adapt to latest core changes... (reactor#160)

  - Operators hooks now take a Context
  - Mono.doAfterTerminate(BiConsumer) is now doAfterSuccessOrError
  - Mono.when methods that produce a Tuple have been renamed zip

Fixed a test that was failing on CI by adding a bit more leeway via sleep.

* attempt at unblocking the startRouterAndAwait test on travis

src/main/java/reactor/ipc/netty/FutureMono.java

@@ -107,8 +107,8 @@ public void subscribe(CoreSubscriber<? super Void> s) {
 
 			if (f == null) {
 				Operators.error(s,
-						Operators.onOperatorError(new NullPointerException(
-								"Deferred supplied null")));
+						Operators.onOperatorError(new NullPointerException("Deferred supplied null"),
+								s.currentContext()));
 				return;
 			}
 

src/main/java/reactor/ipc/netty/channel/FluxReceive.java

@@ -33,6 +33,7 @@
 import reactor.util.Logger;
 import reactor.util.Loggers;
 import reactor.util.concurrent.Queues;
+import reactor.util.context.Context;
 
 /**
  * @author Stephane Maldini
@@ -340,7 +341,8 @@ final boolean onInboundComplete() {
 
 	final boolean onInboundError(Throwable err) {
 		if (isCancelled() || inboundDone) {
-			Operators.onErrorDropped(err);
+			Context c = receiver == null ? Context.empty() : receiver.currentContext();
+			Operators.onErrorDropped(err, c);
 			return false;
 		}
 		CoreSubscriber<?> receiver = this.receiver;

src/main/java/reactor/ipc/netty/http/client/HttpClientOperations.java

@@ -681,7 +681,7 @@ final Mono<Void> withWebsocketSupport(URI url,
 						                                 ops,
 						                                 ops))));
 				if (websocketHandler != noopHandler()) {
-					handshake = handshake.doAfterTerminate(ops);
+					handshake = handshake.doAfterSuccessOrError(ops);
 				}
 				return handshake;
 			}
@@ -695,7 +695,7 @@ else if (isWebsocket()) {
 				if (websocketHandler != noopHandler()) {
 					handshake =
 							handshake.then(Mono.defer(() -> Mono.from(websocketHandler.apply(ops, ops)))
-							         .doAfterTerminate(ops));
+							         .doAfterSuccessOrError(ops));
 				}
 				return handshake;
 			}

src/main/java/reactor/ipc/netty/http/multipart/MultipartParser.java

@@ -73,7 +73,7 @@ public void onSubscribe(Subscription s) {
 	@Override
 	public void onNext(MultipartTokenizer.Token token) {
 		if (done) {
-			Operators.onNextDropped(token);
+			Operators.onNextDropped(token, actual.currentContext());
 			return;
 		}
 
@@ -116,7 +116,7 @@ public void onNext(MultipartTokenizer.Token token) {
 	@Override
 	public void onError(Throwable t) {
 		if (done) {
-			Operators.onErrorDropped(t);
+			Operators.onErrorDropped(t, actual.currentContext());
 			return;
 		}
 		UnicastProcessor<ByteBuf> w = window;

src/main/java/reactor/ipc/netty/http/multipart/MultipartTokenizer.java

@@ -101,7 +101,7 @@ public void onComplete() {
 	@Override
 	public void onError(Throwable throwable) {
 		if (this.done) {
-			Operators.onErrorDropped(throwable);
+			Operators.onErrorDropped(throwable, actual.currentContext());
 			return;
 		}
 
@@ -112,7 +112,7 @@ public void onError(Throwable throwable) {
 	@Override
 	public void onNext(ByteBuf byteBuf) {
 		if (this.done) {
-			Operators.onNextDropped(byteBuf);
+			Operators.onNextDropped(byteBuf, actual.currentContext());
 			return;
 		}
 
@@ -168,15 +168,17 @@ public void request(long n) {
 			if (Integer.MAX_VALUE > n) {  // TODO: Support smaller request sizes
 				actual.onError(Operators.onOperatorError(this, new
 						IllegalArgumentException(
-						"This operation only supports unbounded requests, was " + n), n));
+						"This operation only supports unbounded requests, was " + n), n,
+						actual.currentContext()));
 				return;
 			}
 			Subscription s = this.subscription;
 			if (s != null) {
 				s.request(n);
 			}
 		} catch (Throwable throwable) {
-			actual.onError(Operators.onOperatorError(this, throwable, n));
+			actual.onError(Operators.onOperatorError(this, throwable, n,
+					actual.currentContext()));
 		}
 	}
 

src/main/java/reactor/ipc/netty/http/server/HttpServerOperations.java

@@ -456,7 +456,7 @@ final Mono<Void> withWebsocketSupport(String url,
 			if (replace(ops)) {
 				return FutureMono.from(ops.handshakerResult)
 				                 .then(Mono.defer(() -> Mono.from(websocketHandler.apply(ops, ops))))
-				                 .doAfterTerminate(ops);
+				                 .doAfterSuccessOrError(ops);
 			}
 		}
 		else {

src/main/java/reactor/ipc/netty/resources/DefaultLoopResources.java

@@ -129,7 +129,7 @@ public Mono<Void> disposeLater() {
 				cnsrvlMono = FutureMono.from((Future) cacheNativeServerGroup.terminationFuture());
 			}
 
-			return Mono.when(clMono, sslMono, slMono, cnclMono, cnslMono, cnsrvlMono).then();
+			return Mono.zip(clMono, sslMono, slMono, cnclMono, cnslMono, cnsrvlMono).then();
 		});
 	}
 

src/main/java/reactor/ipc/netty/tcp/TcpResources.java

@@ -137,7 +137,7 @@ protected void _dispose(){
 	 * @return the Mono that represents the end of disposal
 	 */
 	protected Mono<Void> _disposeLater() {
-		return Mono.when(
+		return Mono.zip(
 				defaultLoops.disposeLater(),
 				defaultPools.disposeLater())
 		           .then();

src/test/java/reactor/ipc/netty/http/client/HttpClientWSOperationsTest.java

@@ -104,7 +104,7 @@ private void failOnClientServerError(boolean clientError, boolean serverError,
 	}
 
 	private Mono<HttpClientRequest> doLoginFirst(Mono<HttpClientRequest> request, int port) {
-		return Mono.when(request, login(port))
+		return Mono.zip(request, login(port))
 		           .map(tuple -> {
 		               HttpClientRequest req = tuple.getT1();
 		               req.addHeader("Authorization", tuple.getT2());

src/test/java/reactor/ipc/netty/http/server/HttpServerTests.java

@@ -443,6 +443,7 @@ public void startRouterAndAwait()
 
 		//shutdown the router to unblock the thread
 		ref.get().shutdown();
+		Thread.sleep(100);
 		assertThat(f.isDone()).isTrue();
 	}
 }