package reactor.core.publisher;
import java.util.Objects;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import org.reactivestreams.Publisher;
import reactor.core.CoreSubscriber;
import reactor.core.Disposable;
import reactor.core.Fuseable;
import reactor.core.Scannable;
import reactor.util.annotation.Nullable;
final class ConnectableLiftFuseable<I, O> extends ConnectableFlux<O>
implements Scannable, Fuseable {
final BiFunction<Publisher, ? super CoreSubscriber<? super O>, ? extends CoreSubscriber<? super I>>
lifter;
final ConnectableFlux<I> source;
ConnectableLiftFuseable(ConnectableFlux<I> p,
BiFunction<Publisher, ? super CoreSubscriber<? super O>, ? extends CoreSubscriber<? super I>> lifter) {
this.source = Objects.requireNonNull(p, "source");
this.lifter = lifter;
}
@Override
public int getPrefetch() {
return source.getPrefetch();
}
@Override
public void connect(Consumer<? super Disposable> cancelSupport) {
this.source.connect();
}
@Override
@Nullable
public Object scanUnsafe(Attr key) {
if (key == Attr.PREFETCH) return source.getPrefetch();
if (key == Attr.PARENT) return source;
return null;
}
@Override
public void subscribe(CoreSubscriber<? super O> actual) {
CoreSubscriber<? super I> input =
lifter.apply(source, actual);
Objects.requireNonNull(input, "Lifted subscriber MUST NOT be null");
if (actual instanceof QueueSubscription
&& !(input instanceof QueueSubscription)) {
input = new FluxHide.SuppressFuseableSubscriber<>(input);
}
source.subscribe(input);
}
}