package reactor.core.publisher;
import java.util.Objects;
import org.reactivestreams.Publisher;
import reactor.core.CorePublisher;
import reactor.core.CoreSubscriber;
import reactor.core.Scannable;
import reactor.util.annotation.Nullable;
abstract class FluxFromMonoOperator<I, O> extends Flux<O> implements Scannable,
OptimizableOperator<O, I> {
protected final Mono<? extends I> source;
@Nullable
final OptimizableOperator<?, I> optimizableOperator;
@SuppressWarnings("unchecked")
protected FluxFromMonoOperator(Mono<? extends I> source) {
this.source = Objects.requireNonNull(source);
if (source instanceof OptimizableOperator) {
@SuppressWarnings("unchecked")
OptimizableOperator<?, I> optimSource = (OptimizableOperator<?, I>) source;
this.optimizableOperator = optimSource;
}
else {
this.optimizableOperator = null;
}
}
@Override
@Nullable
public Object scanUnsafe(Attr key) {
if (key == Attr.PREFETCH) return getPrefetch();
if (key == Attr.PARENT) return source;
return null;
}
@Override
@SuppressWarnings("unchecked")
public final void subscribe(CoreSubscriber<? super O> subscriber) {
OptimizableOperator operator = this;
try {
while (true) {
subscriber = operator.subscribeOrReturn(subscriber);
if (subscriber == null) {
return;
}
OptimizableOperator newSource = operator.nextOptimizableSource();
if (newSource == null) {
operator.source().subscribe(subscriber);
return;
}
operator = newSource;
}
}
catch (Throwable e) {
Operators.reportThrowInSubscribe(subscriber, e);
return;
}
}
@Override
@Nullable
public abstract CoreSubscriber<? super I> subscribeOrReturn(CoreSubscriber<? super O> actual) throws Throwable;
@Override
public final CorePublisher<? extends I> source() {
return source;
}
@Override
public final OptimizableOperator<?, ? extends I> nextOptimizableSource() {
return optimizableOperator;
}
}