class CacheLoader extends MonoOperator<Object, Object> {

        private final AbstractReactiveCache<?> parent;

        private final Object key;
        private Mono<? extends Object> defaultValue;

        private final Sinks.One<Object> holder = Sinks.one();

        private volatile Disposable loading;

        protected CacheLoader(AbstractReactiveCache<?> parent, Object key, Mono<? extends Object> source) {
            super(source.cache());
            this.parent = parent;
            this.key = key;
        }

        protected void defaultValue(Mono<? extends Object> defaultValue, ContextView context) {
            if (this.defaultValue != null) {
                return;
            }
            this.defaultValue = defaultValue;
            tryLoad(context);
        }


        @SuppressWarnings("all")
        private void tryLoad(ContextView context) {
            if (holder.currentSubscriberCount() == 1 && loading == null) {
                Mono<? extends Object> source = this.source;
                if (defaultValue != null) {
                    source = source
                            .switchIfEmpty((Mono) defaultValue
                                    .flatMap(val -> {
                                        return parent.putNow(key, val).thenReturn(val);
                                    }));
                }
                loading = source.subscribe(
                        val -> {
                            complete();
                            holder.emitValue(val, emitFailureHandler);
                        },
                        err -> {
                            complete();
                            holder.emitError(err, emitFailureHandler);
                        },
                        () -> {
                            complete();
                            holder.emitEmpty(emitFailureHandler);
                        },
                        Context.of(context));
            }
        }

        @Override
        public void subscribe(CoreSubscriber<? super Object> actual) {
            holder.asMono().subscribe(actual);
            tryLoad(actual.currentContext());
        }

        private void complete() {
            parent.cacheLoading.remove(key, this);
        }


    }

    protected abstract Mono<Object> getNow(Object key);

    public abstract Mono<Void> putNow(Object key, Object value);

    @Override
    @SuppressWarnings("all")
    public final Mono<E> getMono(Object key) {
        return (Mono<E>) cacheLoading.computeIfAbsent(key, _key -> new CacheLoader(this, _key, getNow(_key)));
    }

    @Override
    @SuppressWarnings("all")
    public final Mono<E> getMono(Object key, Supplier<Mono<E>> loader) {

        return Mono.deferContextual(ctx -> {
            CacheLoader cacheLoader = cacheLoading.compute(key, (_key, old) -> {
                CacheLoader cl = new CacheLoader(this, _key, getNow(_key));
                cl.defaultValue(loader.get(), ctx);
                return cl;
            });
            return (Mono<E>) cacheLoader;
        });
    }


    @Override
    public final Flux<E> getFlux(Object key) {
        return (cacheLoading.computeIfAbsent(key, _key -> new CacheLoader(this, _key, getNow(_key))))
                .flatMapIterable(e -> ((List<E>) e));
    }

    @Override
    public final Flux<E> getFlux(Object key, Supplier<Flux<E>> loader) {<FILL_FUNCTION_BODY>

        return Flux.deferContextual(ctx -> {
            CacheLoader cacheLoader = cacheLoading.compute(key, (_key, old) -> {
                CacheLoader cl = new CacheLoader(this, _key, getNow(_key));
                cl.defaultValue(loader.get().collectList(), ctx);
                return cl;
            });
            return cacheLoader.flatMapIterable(e -> ((List<E>) e));
        });
    
<no_super_class>