From 7500c49a16c9fa5164a07ccc809f97bec85b2dd1 Mon Sep 17 00:00:00 2001
From: Violeta Georgieva <violetag@vmware.com>
Date: Mon, 6 Jul 2020 17:24:31 +0300
Subject: [PATCH] fix #1165 Change the state after notifying that the
 connection is acquired successfully.

---
 .../java/reactor/netty/channel/ChannelOperations.java  |  4 ++++
 .../netty/resources/PooledConnectionProvider.java      | 10 +++++++++-
 2 files changed, 13 insertions(+), 1 deletion(-)

diff --git a/src/main/java/reactor/netty/channel/ChannelOperations.java b/src/main/java/reactor/netty/channel/ChannelOperations.java
index 9214d0ae31..2dda408f6f 100644
--- a/src/main/java/reactor/netty/channel/ChannelOperations.java
+++ b/src/main/java/reactor/netty/channel/ChannelOperations.java
@@ -157,6 +157,10 @@ public ChannelOperations<INBOUND, OUTBOUND> withConnection(Consumer<? super Conn
 
 	@Override
 	public void dispose() {
+		if (log.isTraceEnabled()) {
+			log.trace(format(channel(), "Disposing ChannelOperation from a channel"),
+					new Exception("ChannelOperation dispose stack"));
+		}
 		if (!inbound.isDisposed()) {
 			inbound.cancel();
 		}
diff --git a/src/main/java/reactor/netty/resources/PooledConnectionProvider.java b/src/main/java/reactor/netty/resources/PooledConnectionProvider.java
index 23e94a60c3..b1f83b1de9 100644
--- a/src/main/java/reactor/netty/resources/PooledConnectionProvider.java
+++ b/src/main/java/reactor/netty/resources/PooledConnectionProvider.java
@@ -531,6 +531,10 @@ public void onUncaughtException(Connection connection, Throwable error) {
 		@Override
 		public void onStateChange(Connection connection, State newState) {
 			if (newState == State.CONFIGURED) {
+				// First send a notification that the connection is ready and then change the state
+				// In case a cancellation was received, ChannelOperations will be disposed
+				// and there will be no subscription to the I/O handler at all.
+				// https://github.com/reactor/reactor-netty/issues/1165
 				sink.success(connection);
 			}
 			obs.onStateChange(connection, newState);
@@ -600,8 +604,12 @@ else if (current == null) {
 				ChannelOperations<?, ?> ops = opsFactory.create(pooledConnection, pooledConnection, null);
 				if (ops != null) {
 					ops.bind();
-					obs.onStateChange(ops, State.CONFIGURED);
+					// First send a notification that the connection is ready and then change the state
+					// In case a cancellation was received, ChannelOperations will be disposed
+					// and there will be no subscription to the I/O handler at all.
+					// https://github.com/reactor/reactor-netty/issues/1165
 					sink.success(ops);
+					obs.onStateChange(ops, State.CONFIGURED);
 				}
 				else {
 					//already configured, just forward the connection