本文整理了Java中reactor.core.publisher.Operators.onOperatorError()方法的一些代码示例，展示了Operators.onOperatorError()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Operators.onOperatorError()方法的具体详情如下：
包路径：reactor.core.publisher.Operators
类名称：Operators
方法名：onOperatorError

Operators.onOperatorError介绍

[英]Map an "operator" error. The result error will be passed via onError to the operator downstream after checking for fatal error via Exceptions#throwIfFatal(Throwable).
[中]映射“操作员”错误。在通过异常#throwIfFatal（Throwable）检查致命错误后，结果错误将通过OneError传递给下游操作员。

代码示例

代码示例来源：origin: reactor/reactor-core

@Override
public void subscribe(CoreSubscriber<? super T> actual) {
  Throwable error = Objects.requireNonNull(errorSupplier.get(), "the errorSupplier returned null");
  Operators.error(actual, Operators.onOperatorError(error, actual.currentContext()));
}

代码示例来源：origin: reactor/reactor-core

@Override
public void subscribe(CoreSubscriber<? super T> actual) {
  Throwable error = Objects.requireNonNull(errorSupplier.get(), "errorSupplier produced a null Throwable");
  Operators.error(actual, Operators.onOperatorError(error, actual.currentContext()));
}

代码示例来源：origin: reactor/reactor-core

@Override
public void onError(Throwable e) {
  done = true;
  Throwable e_ = Operators.onOperatorError(e, parent.currentContext());
  Throwable commitError = new RuntimeException("Async resource cleanup failed after onComplete", e_);
  parent.deferredError(commitError);
}

代码示例来源：origin: resilience4j/resilience4j

@Override
public final void onNext(T value) {
  Objects.requireNonNull(value, "onNext");
  try {
    hookOnNext(value);
  }
  catch (Throwable throwable) {
    onError(Operators.onOperatorError(subscription, throwable, value, currentContext()));
  }
}

代码示例来源：origin: reactor/reactor-core

@Override
public final void onNext(T value) {
  Objects.requireNonNull(value, "onNext");
  try {
    hookOnNext(value);
  }
  catch (Throwable throwable) {
    onError(Operators.onOperatorError(subscription, throwable, value, currentContext()));
  }
}

代码示例来源：origin: reactor/reactor-core

@Override
public void subscribe(CoreSubscriber<? super T> actual) {
  Iterator<? extends T> it;
  try {
    it = iterable.iterator();
  }
  catch (Throwable e) {
    Operators.error(actual, Operators.onOperatorError(e, actual.currentContext()));
    return;
  }
  subscribe(actual, it, onClose);
}

代码示例来源：origin: reactor/reactor-core

@Override
public void subscribe(CoreSubscriber<? super T> actual) {
  try {
    run.run();
  } catch (Throwable ex) {
    Operators.error(actual, Operators.onOperatorError(ex, actual.currentContext()));
    return;
  }
  Operators.complete(actual);
}

代码示例来源：origin: reactor/reactor-core

@Override
public void request(long n) {
  final LongConsumer requestHook = parent.onRequestCall();
  if (requestHook != null) {
    try {
      requestHook.accept(n);
    }
    catch (Throwable e) {
      Operators.onOperatorError(e, actual.currentContext());
    }
  }
  s.request(n);
}

代码示例来源：origin: reactor/reactor-core

@Override
public void onError(Throwable t) {
  if (isCancelled()) {
    Operators.onOperatorError(t, currentContext());
  } else {
    actual.onError(t);
  }
}

代码示例来源：origin: reactor/reactor-core

@Override
public void onNext(T t) {
  if (!queue.offer(t)) {
    onError(Operators.onOperatorError(cachedSubscription, Exceptions
        .failWithOverflow(Exceptions.BACKPRESSURE_ERROR_QUEUE_FULL), t,
        currentContext()));
    return;
  }
  parent.drain();
}

代码示例来源：origin: reactor/reactor-core

@Override
public void subscribe(CoreSubscriber<? super T> actual) {
  DefaultMonoSink<T> emitter = new DefaultMonoSink<>(actual);
  actual.onSubscribe(emitter);
  try {
    callback.accept(emitter);
  }
  catch (Throwable ex) {
    emitter.error(Operators.onOperatorError(ex, actual.currentContext()));
  }
}

代码示例来源：origin: reactor/reactor-core

@Override
public void onNext(T t) {
  if (sourceMode != ASYNC) {
    if (!queue.offer(t)) {
      onError(Operators.onOperatorError(s, Exceptions.failWithOverflow
          (Exceptions.BACKPRESSURE_ERROR_QUEUE_FULL), currentContext()));
      return;
    }
  }
  parent.drain();
}

代码示例来源：origin: reactor/reactor-core

@Override
public void subscribe(CoreSubscriber<? super T> actual) {
  S state;
  try {
    state = stateSupplier.call();
  } catch (Throwable e) {
    Operators.error(actual, Operators.onOperatorError(e, actual.currentContext()));
    return;
  }
  actual.onSubscribe(new GenerateSubscription<>(actual, state, generator, stateConsumer));
}

代码示例来源：origin: reactor/reactor-core

@Override
public void subscribe(CoreSubscriber<? super T> actual) {
  Context c;
  try {
    c = doOnContext.apply(actual.currentContext());
  }
  catch (Throwable t) {
    Operators.error(actual, Operators.onOperatorError(t, actual.currentContext()));
    return;
  }
  source.subscribe(new ContextStartSubscriber<>(actual, c));
}

代码示例来源：origin: reactor/reactor-core

@Override
public void onNext(T t) {
  if (!queue.offer(t)) {
    Operators.terminate(S, this);
    onError(Operators.onOperatorError(null,
        Exceptions.failWithOverflow(Exceptions.BACKPRESSURE_ERROR_QUEUE_FULL),
        t, currentContext()));
  }
  else {
    signalConsumer();
  }
}

代码示例来源：origin: reactor/reactor-core

@Override
public void subscribe(CoreSubscriber<? super T> actual) {
  Context c;
  try {
    c = doOnContext.apply(actual.currentContext());
  }
  catch (Throwable t) {
    Operators.error(actual, Operators.onOperatorError(t, actual.currentContext()));
    return;
  }
  source.subscribe(new FluxContextStart.ContextStartSubscriber<>(actual, c));
}

代码示例来源：origin: reactor/reactor-core

@Override
public final void onSubscribe(Subscription s) {
  if (Operators.setOnce(S, this, s)) {
    try {
      hookOnSubscribe(s);
    }
    catch (Throwable throwable) {
      onError(Operators.onOperatorError(s, throwable, currentContext()));
    }
  }
}

代码示例来源：origin: reactor/reactor-core

@Test
public void onOperatorErrorLocal() {
  BiFunction<Throwable, Object, Throwable> localHook = (e, v) ->
      new IllegalStateException("boom_" + v, e);
  Context c = Context.of(Hooks.KEY_ON_OPERATOR_ERROR, localHook);
  IllegalArgumentException failure = new IllegalArgumentException("foo");
  final Throwable throwable = Operators.onOperatorError(null, failure,
      "foo", c);
  assertThat(throwable).isInstanceOf(IllegalStateException.class)
             .hasMessage("boom_foo")
             .hasCause(failure);
}

代码示例来源：origin: reactor/reactor-core

void innerNext(MergeSequentialInner<R> inner, R value) {
  if (inner.queue().offer(value)) {
    drain();
  }
  else {
    inner.cancel();
    onError(Operators.onOperatorError(null, Exceptions.failWithOverflow(Exceptions.BACKPRESSURE_ERROR_QUEUE_FULL), value,
        actual.currentContext()));
  }
}

代码示例来源：origin: reactor/reactor-core

boolean failOverflow(R v, Subscription toCancel){
  Throwable e = Operators.onOperatorError(toCancel,
      Exceptions.failWithOverflow(Exceptions.BACKPRESSURE_ERROR_QUEUE_FULL),
      v, actual.currentContext());
  if (!Exceptions.addThrowable(ERROR, this, e)) {
    Operators.onErrorDropped(e, actual.currentContext());
    return false;
  }
  return true;
}