fix #1165 Change the state after notifying that the connection is acq…

…uired successfully.

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();
 		}

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