package reactor.core.publisher;
import reactor.core.publisher.Sinks.EmissionException;
interface InternalEmptySink<T> extends Sinks.Empty<T>, ContextHolder {
@Override
default void emitEmpty(Sinks.EmitFailureHandler failureHandler) {
for (;;) {
Sinks.EmitResult emitResult = tryEmitEmpty();
if (emitResult.isSuccess()) {
return;
}
boolean shouldRetry = failureHandler.onEmitFailure(SignalType.ON_COMPLETE,
emitResult);
if (shouldRetry) {
continue;
}
switch (emitResult) {
case FAIL_ZERO_SUBSCRIBER:
case FAIL_OVERFLOW:
case FAIL_CANCELLED:
case FAIL_TERMINATED:
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");
}
}
}
@Override
default void emitError(Throwable error, Sinks.EmitFailureHandler failureHandler) {
for (;;) {
Sinks.EmitResult emitResult = tryEmitError(error);
if (emitResult.isSuccess()) {
return;
}
boolean shouldRetry = failureHandler.onEmitFailure(SignalType.ON_ERROR,
emitResult);
if (shouldRetry) {
continue;
}
switch (emitResult) {
case FAIL_ZERO_SUBSCRIBER:
case FAIL_OVERFLOW:
case FAIL_CANCELLED:
return;
case FAIL_TERMINATED:
Operators.onErrorDropped(error, 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");
}
}
}
}