Merge #3448 into 1.2.0-RC1

reactor-netty-core/src/main/java/reactor/netty/ReactorNetty.java

@@ -23,6 +23,7 @@
 import java.util.Objects;
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
@@ -721,42 +722,43 @@ public void onStateChange(Connection connection, State newState) {
 	 * An appending write that delegates to its origin context and append the passed
 	 * publisher after the origin success if any.
 	 */
-	static final class OutboundThen implements NettyOutbound {
+	static final class OutboundThen extends AtomicBoolean implements NettyOutbound {
 
 		final NettyOutbound source;
 		final Mono<Void> thenMono;
 
 		static final Runnable EMPTY_CLEANUP = () -> {};
 
-
 		OutboundThen(NettyOutbound source, Publisher<Void> thenPublisher) {
 			this(source, thenPublisher, EMPTY_CLEANUP);
 		}
 
+		// This construction is used only with ChannelOperations#sendObject
+		// The implementation relies on Netty's promise that Channel#writeAndFlush will release the buffer on success/error
+		// The onCleanup callback is invoked only in case when we are sure that the processing doesn't delegate to Netty
+		// because of some failure before the exchange can be continued in the thenPublisher
 		OutboundThen(NettyOutbound source, Publisher<Void> thenPublisher, Runnable onCleanup) {
 			this.source = source;
 			Objects.requireNonNull(onCleanup, "onCleanup");
 
 			Mono<Void> parentMono = source.then();
 
 			if (parentMono == Mono.<Void>empty()) {
-				if (onCleanup == EMPTY_CLEANUP) {
-					this.thenMono = Mono.from(thenPublisher);
-				}
-				else {
-					this.thenMono = Mono.from(thenPublisher)
-					                    .doOnCancel(onCleanup)
-					                    .doOnError(t -> onCleanup.run());
-				}
+				this.thenMono = Mono.from(thenPublisher);
 			}
 			else {
 				if (onCleanup == EMPTY_CLEANUP) {
 					this.thenMono = parentMono.thenEmpty(thenPublisher);
 				}
 				else {
-					this.thenMono = parentMono.thenEmpty(thenPublisher)
-					                          .doOnCancel(onCleanup)
-					                          .doOnError(t -> onCleanup.run());
+					this.thenMono = parentMono
+							.doFinally(signalType -> {
+								if ((signalType == SignalType.CANCEL || signalType == SignalType.ON_ERROR) &&
+										compareAndSet(false, true)) {
+									onCleanup.run();
+								}
+							})
+							.thenEmpty(thenPublisher);
 				}
 			}
 		}

reactor-netty-core/src/test/java/reactor/netty/tcp/TcpServerTests.java

@@ -53,6 +53,7 @@
 import io.netty.channel.group.DefaultChannelGroup;
 import io.netty.channel.unix.DomainSocketAddress;
 import io.netty.handler.codec.LineBasedFrameDecoder;
+import io.netty.handler.codec.MessageToMessageEncoder;
 import io.netty.handler.codec.json.JsonObjectDecoder;
 import io.netty.handler.ssl.SniCompletionEvent;
 import io.netty.handler.ssl.SslContext;
@@ -67,6 +68,8 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.reactivestreams.Publisher;
 import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
@@ -84,6 +87,7 @@
 import reactor.netty.SocketUtils;
 import reactor.netty.channel.ChannelOperations;
 import reactor.netty.resources.LoopResources;
+import reactor.test.StepVerifier;
 import reactor.util.Logger;
 import reactor.util.Loggers;
 
@@ -1234,4 +1238,55 @@ void testTcpServerCancelled() throws InterruptedException {
 			assertThat(serverMessages.size()).isEqualTo(0);
 		}
 	}
+
+	@ParameterizedTest
+	@ValueSource(booleans = {true, false})
+	void testIssue3406(boolean singleInvocation) {
+		DisposableServer server = null;
+		Connection client = null;
+
+		try {
+			server =
+					TcpServer.create()
+					         .wiretap(true)
+					         .handle((in, out) -> out.sendString(Mono.just("testIssue3406"))
+					                                 .then(in.receive().then()))
+					         .bindNow();
+
+			Sinks.One<Void> result = Sinks.one();
+			client =
+					TcpClient.create()
+					         .remoteAddress(server::address)
+					         .wiretap(true)
+					         .doOnConnected(conn ->
+					                 conn.addHandlerFirst(
+					                     new MessageToMessageEncoder<Object>() {
+					                         @Override
+					                         protected void encode(ChannelHandlerContext ctx, Object msg, List<Object> out) {
+					                             // This is no-op in order to force Netty to release the 'msg'
+					                             // and to throw Exception
+					                         }
+					                     }))
+					         .handle((in, out) ->
+					                 in.receive()
+					                   .retain()
+					                   .doOnError(result::tryEmitError)
+					                   .doOnComplete(result::tryEmitEmpty)
+					                   .flatMap(b -> singleInvocation ? out.sendObject(b) : out.sendObject(b.retain()).sendObject(b)))
+					         .connectNow();
+
+			result.asMono()
+			      .as(StepVerifier::create)
+			      .expectComplete()
+			      .verify(Duration.ofSeconds(5));
+		}
+		finally {
+			if (client != null) {
+				client.disposeNow();
+			}
+			if (server != null) {
+				server.disposeNow();
+			}
+		}
+	}
 }