package reactor.core.publisher;
import reactor.core.Exceptions;
import reactor.core.publisher.Sinks.EmissionException;
import reactor.util.annotation.Nullable;
interface InternalOneSink<T> extends Sinks.One<T>, InternalEmptySink<T> {
@Override
default void emitValue(@Nullable T value, Sinks.EmitFailureHandler failureHandler) {
if (value == null) {
emitEmpty(failureHandler);
return;
}
for (;;) {
Sinks.EmitResult emitResult = tryEmitValue(value);
if (emitResult.isSuccess()) {
return;
}
boolean shouldRetry = failureHandler.onEmitFailure(SignalType.ON_NEXT,
emitResult);
if (shouldRetry) {
continue;
}
switch (emitResult) {
case FAIL_ZERO_SUBSCRIBER:
return;
case FAIL_OVERFLOW:
Operators.onDiscard(value, currentContext());
emitError(Exceptions.failWithOverflow("Backpressure overflow during Sinks.Many#emitNext"),
failureHandler);
return;
case FAIL_CANCELLED:
Operators.onDiscard(value, currentContext());
return;
case FAIL_TERMINATED:
Operators.onNextDropped(value, currentContext());
return;
case FAIL_NON_SERIALIZED:
throw new EmissionException(emitResult,
"Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially."
);
default:
throw new EmissionException(emitResult, "Unknown emitResult value");
}
}
}
}