package reactor.core.publisher;
import java.util.Objects;
import org.reactivestreams.Subscription;
import reactor.core.CoreSubscriber;
import reactor.core.Fuseable;
import reactor.util.annotation.Nullable;
final class MonoElementAt<T> extends MonoFromFluxOperator<T, T>
implements Fuseable {
final long index;
final T defaultValue;
MonoElementAt(Flux<? extends T> source, long index) {
super(source);
if (index < 0) {
throw new IndexOutOfBoundsException("index >= required but it was " + index);
}
this.index = index;
this.defaultValue = null;
}
MonoElementAt(Flux<? extends T> source, long index, T defaultValue) {
super(source);
if (index < 0) {
throw new IndexOutOfBoundsException("index >= required but it was " + index);
}
this.index = index;
this.defaultValue = Objects.requireNonNull(defaultValue, "defaultValue");
}
@Override
public CoreSubscriber<? super T> subscribeOrReturn(CoreSubscriber<? super T> actual) {
return new ElementAtSubscriber<>(actual, index, defaultValue);
}
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
return super.scanUnsafe(key);
}
static final class ElementAtSubscriber<T>
extends Operators.MonoSubscriber<T, T> {
final T defaultValue;
long index;
Subscription s;
boolean done;
ElementAtSubscriber(CoreSubscriber<? super T> actual, long index,
T defaultValue) {
super(actual);
this.index = index;
this.defaultValue = defaultValue;
}
@Override
@Nullable
public Object scanUnsafe(Attr key) {
if (key == Attr.TERMINATED) return done;
if (key == Attr.PARENT) return s;
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
return super.scanUnsafe(key);
}
@Override
public void request(long n) {
super.request(n);
if (n > 0L) {
s.request(Long.MAX_VALUE);
}
}
@Override
public void cancel() {
super.cancel();
s.cancel();
}
@Override
public void onSubscribe(Subscription s) {
if (Operators.validate(this.s, s)) {
this.s = s;
actual.onSubscribe(this);
}
}
@Override
public void onNext(T t) {
if (done) {
Operators.onNextDropped(t, actual.currentContext());
return;
}
long i = index;
if (i == 0) {
done = true;
s.cancel();
actual.onNext(t);
actual.onComplete();
return;
}
index = i - 1;
Operators.onDiscard(t, actual.currentContext());
}
@Override
public void onError(Throwable t) {
if (done) {
Operators.onErrorDropped(t, actual.currentContext());
return;
}
done = true;
actual.onError(t);
}
@Override
public void onComplete() {
if (done) {
return;
}
done = true;
if(defaultValue != null) {
complete(defaultValue);
}
else{
actual.onError(Operators.onOperatorError(new
IndexOutOfBoundsException(), actual.currentContext()));
}
}
}
}