import reactor.core.Exceptions; //导入依赖的package包/类
void redirect(String to) {
	String[] redirectedFrom = this.redirectedFrom;
	URI from = activeURI;
	try {
		activeURI = new URI(to);
	}
	catch (URISyntaxException e) {
		throw Exceptions.propagate(e);
	}
	if (redirectedFrom == null) {
		this.redirectedFrom = new String[]{from.toString()};
	}
	else {
		String[] newRedirectedFrom = new String[redirectedFrom.length + 1];
		System.arraycopy(redirectedFrom,
				0,
				newRedirectedFrom,
				0,
				redirectedFrom.length);
		newRedirectedFrom[redirectedFrom.length] = from.toString();
		this.redirectedFrom = newRedirectedFrom;
	}
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
@Test
public void afterTerminateCallbackFailureInterruptsOnNextAndThrows() {
	LongAdder invoked = new LongAdder();
	try {
		StepVerifier.create(Mono.just("foo")
		                        .doAfterTerminate(() -> {
			                        invoked.increment();
			                        throw new IllegalArgumentException("boom");
		                        }))
		            .expectNext("bar") //irrelevant
		            .expectErrorMessage("baz") //irrelevant
		            .verify();
	}
	catch (Throwable t) {
		Throwable e = Exceptions.unwrap(t);
		assertEquals(IllegalArgumentException.class, e.getClass());
		assertEquals("boom", e.getMessage());
	}

	assertEquals(1, invoked.intValue());
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
@Override
final public void channelInactive(ChannelHandlerContext ctx) throws Exception {
	try {
		ChannelOperations<?, ?> ops = ChannelOperations.get(ctx.channel());
		if (ops != null) {
			ops.onInboundClose();
		}
		else {
			if (lastContext != null) {
				lastContext.terminateChannel(ctx.channel());
				lastContext.fireContextError(new AbortedException());
			}
		}
	}
	catch (Throwable err) {
		Exceptions.throwIfFatal(err);
		exceptionCaught(ctx, err);
	}
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
final HttpClientFormEncoder applyChanges(HttpRequest request) {
	if (!needNewEncoder) {
		return this;
	}

	try {
		HttpClientFormEncoder encoder = new HttpClientFormEncoder(newFactory,
				request,
				newMultipart,
				newCharset,
				newMode);

		encoder.setBodyHttpDatas(getBodyListAttributes());

		return encoder;
	}
	catch(ErrorDataEncoderException ee){
		throw Exceptions.propagate(ee);
	}
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
final URI websocketUri() {
	URI uri;
	try {
		String url = uri();
		if (url.startsWith(HttpClient.HTTP_SCHEME) || url.startsWith(HttpClient.WS_SCHEME)) {
			uri = new URI(url);
		}
		else {
			String host = requestHeaders().get(HttpHeaderNames.HOST);
			uri = new URI((isSecure ? HttpClient.WSS_SCHEME :
					HttpClient.WS_SCHEME) + "://" + host + (url.startsWith("/") ?
					url : "/" + url));
		}

	}
	catch (URISyntaxException e) {
		throw Exceptions.bubble(e);
	}
	return uri;
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
@Override
public Publisher<Void> apply(HttpServerRequest request, HttpServerResponse response) {
	final Iterator<HttpRouteHandler> iterator = handlers.iterator();
	HttpRouteHandler cursor;

	try {
		while (iterator.hasNext()) {
			cursor = iterator.next();
			if (cursor.test(request)) {
				return cursor.apply(request, response);
			}
		}
	}
	catch (Throwable t) {
		Exceptions.throwIfFatal(t);
		return Mono.error(t); //500
	}

	return response.sendNotFound();
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
public Flux<JsonElement> queryAsync(JsonObject query) {
  return Flux.just(query)
      .map(q -> validate(q))
      .flatMap(
          s -> {
            Flux<JsonElement> obs = null;
            if ("valid".equals(s)) {
              JsonElement from = query.get("from");
              obs =
                  jsonLever.isArray(from)
                      ? queryAsync(from.getAsJsonArray(), query)
                      : queryAsync(from.getAsJsonObject(), query);
            } else {
              Exceptions.propagate(new Throwable(s));
            }
            return obs;
          });
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
@Override
public void onNext(U t) {
	SampleMainSubscriber<T> m = main;

	T v = m.getAndNullValue();

	if (v != null) {
		if (m.requested != 0L) {
			m.actual.onNext(v);

			if (m.requested != Long.MAX_VALUE) {
				m.decrement();
			}
			return;
		}

		m.cancel();

		m.actual.onError(Exceptions.failWithOverflow("Can't signal value due to lack of requests"));
	}
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
boolean checkTerminated(boolean d, boolean empty, Subscriber<?> a, Queue<?> q) {
	if (cancelled) {
		q.clear();
		return true;
	}
	if (d) {
		Throwable e = Exceptions.terminate(ERROR, this);
		if (e != null && e != Exceptions.TERMINATED) {
			cancel();

			q.clear();

			a.onError(e);
			return true;
		}
		else if (empty) {

			a.onComplete();
			return true;
		}
	}
	return false;
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
boolean onSignal(Signal<T> actualSignal) {
	SignalEvent<T> signalEvent = (SignalEvent<T>) this.script.poll();
	Optional<AssertionError> error = signalEvent.test(actualSignal);
	if (error.isPresent()) {
		Exceptions.addThrowable(ERRORS, this, error.get());
		// #55 ensure the onError is added as a suppressed to the AssertionError
		if(actualSignal.isOnError()) {
			error.get().addSuppressed(actualSignal.getThrowable());
		}
		maybeCancel(actualSignal);
		this.completeLatch.countDown();
		return true;
	}
	if (actualSignal.isOnNext()) {
		unasserted--;
	}
	return false;
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
final boolean onSignalCount(Signal<T> actualSignal, SignalCountEvent<T> event) {
	if (unasserted >= event.count) {
		this.script.poll();
		unasserted -= event.count;
	}
	else {
		if (event.count != 0) {
			Optional<AssertionError> error =
					this.checkCountMismatch(event, actualSignal);

			if (error.isPresent()) {
				Exceptions.addThrowable(ERRORS, this, error.get());
				if(actualSignal.isOnError()) {
					// #55 ensure the onError is added as a suppressed to the AssertionError
					error.get().addSuppressed(actualSignal.getThrowable());
				}
				maybeCancel(actualSignal);
				this.completeLatch.countDown();
			}
		}
		return true;
	}
	return false;
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
@Test
public void completeCallbackError() {
	AssertSubscriber<Integer> ts = AssertSubscriber.create();
	LongAdder state = new LongAdder();

	Throwable err = new Exception("test");

	Flux.just(1)
	    .doOnEach(s -> {
		    if (s.isOnComplete()) {
			    state.increment();
			    throw Exceptions.propagate(err);
		    }
	    })
	    .subscribe(ts);

	ts.assertErrorMessage("test");
	Assert.assertEquals(1, state.intValue());
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
@Override
public void onError(Throwable t) {
	if (!second) {
		second = true;

		Publisher<? extends T> p;

		try {
			p = Objects.requireNonNull(nextFactory.apply(t),
			"The nextFactory returned a null Publisher");
		}
		catch (Throwable e) {
			Throwable _e = Operators.onOperatorError(e, actual.currentContext());
			_e = Exceptions.addSuppressed(_e, t);
			actual.onError(_e);
			return;
		}
		p.subscribe(this);
	}
	else {
		actual.onError(t);
	}
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
@Override
public void onError(Throwable t) {
	if (Exceptions.addThrowable(ERROR, this, t)) {
		inner.cancel();

		if (GUARD.getAndIncrement(this) == 0) {
			t = Exceptions.terminate(ERROR, this);
			if (t != TERMINATED) {
				actual.onError(t);
			}
		}
	}
	else {
		Operators.onErrorDropped(t, actual.currentContext());
	}
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
public void onNext(T t) {
	Subscriber<? super T> a = actual;

	if (!queue.offer(t)) {
		onError(Operators.onOperatorError(this, Exceptions.failWithOverflow(Exceptions.BACKPRESSURE_ERROR_QUEUE_FULL), t,
				actual.currentContext()));
		return;
	}
	if (enableOperatorFusion) {
		if (a != null) {
			a.onNext(null); // in op-fusion, onNext(null) is the indicator of more data
		}
	}
	else {
		drain();
	}
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
boolean emit(C b) {
	long r = requested;
	if (r != 0L) {
		actual.onNext(b);
		if (r != Long.MAX_VALUE) {
			REQUESTED.decrementAndGet(this);
		}
		return true;
	}
	else {
		actual.onError(Operators.onOperatorError(this, Exceptions
				.failWithOverflow(), b, actual.currentContext()));

		return false;
	}
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
@Override
public void onError(Throwable t) {
	if (done) {
		Operators.onErrorDropped(t, cachedContext);
		return;
	}
	done = true;
	try {
		//noinspection ConstantConditions
		onSignal.accept(Signal.error(t, cachedContext));
	}
	catch (Throwable e) {
		//this performs a throwIfFatal or suppresses t in e
		t = Operators.onOperatorError(null, e, t, cachedContext);
	}

	try {
		actual.onError(t);
	}
	catch (UnsupportedOperationException use) {
		if (!Exceptions.isErrorCallbackNotImplemented(use) && use.getCause() != t) {
			throw use;
		}
		//ignore if missing callback
	}
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
@Test
public void callbackThrows() {
	try {
		StepVerifier.create(Flux.just(1)
		                        .doFinally(signal -> {
			                        throw new IllegalStateException();
		                        }))
		            .expectNext(1)
		            .expectComplete()
		            .verify();
	}
	catch (Throwable e) {
		Throwable _e = Exceptions.unwrap(e);
		assertNotSame(e, _e);
		assertThat(_e, is(instanceOf(IllegalStateException.class)));
	}
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
@Test
public void oneRetry() {
	AssertSubscriber<Integer> ts = AssertSubscriber.create();

	AtomicInteger i = new AtomicInteger();
	Mono.fromCallable(() -> {
		int _i = i.getAndIncrement();
		if (_i < 1) {
			throw Exceptions.propagate(new RuntimeException("forced failure"));
		}
		return _i;
	})
	    .retry(1)
	    .subscribe(ts);

	ts.assertValues(1)
	  .assertComplete()
	  .assertNoError();
}
 

import reactor.core.Exceptions; //导入依赖的package包/类
public void onNext(V t) {
	Subscriber<? super V> a = actual;

	if (!queue.offer(t)) {
		onError(Operators.onOperatorError(this, Exceptions.failWithOverflow(Exceptions.BACKPRESSURE_ERROR_QUEUE_FULL), t,
				actual.currentContext()));
		return;
	}
	if (outputFused) {
		if (a != null) {
			a.onNext(null); // in op-fusion, onNext(null) is the indicator of more data
		}
	}
	else {
		drain();
	}
}