Extract channel close listener into static CloseChannelFutureListener

The close listener now captures only the NettyStream, so the one-shot
connection-open handler can no longer be pinned for the channel lifetime.
Replaces the local-variable capture idiom with a structural guarantee.

JAVA-6250

Author: rozza

driver-core/src/main/com/mongodb/internal/connection/netty/NettyStream.java

@@ -530,11 +530,9 @@ public void operationComplete(final ChannelFuture future) {
                         channelFuture.channel().close();
                     } else {
                         channel = channelFuture.channel();
-                        // capture only the enclosing stream in the close listener: capturing OpenChannelFutureListener.this
-                        // would pin the open handler (and everything it references) for the lifetime of the pooled channel
-                        NettyStream stream = NettyStream.this;
-                        channel.closeFuture().addListener((ChannelFutureListener) future1 ->
-                                stream.handleReadResponse(null, new IOException("The connection to the server was closed")));
+                        // CloseChannelFutureListener captures only the NettyStream, never OpenChannelFutureListener.this,
+                        // so the one-shot connection-open handler is not pinned for the lifetime of the pooled channel (JAVA-6250)
+                        channel.closeFuture().addListener(new CloseChannelFutureListener(NettyStream.this));
                     }
                     handler.completed(null);
                 } else {
@@ -550,6 +548,19 @@ public void operationComplete(final ChannelFuture future) {
         }
     }
 
+    private static final class CloseChannelFutureListener implements ChannelFutureListener {
+        private final NettyStream stream;
+
+        CloseChannelFutureListener(final NettyStream stream) {
+            this.stream = stream;
+        }
+
+        @Override
+        public void operationComplete(final ChannelFuture future) {
+            stream.handleReadResponse(null, new IOException("The connection to the server was closed"));
+        }
+    }
+
     private static void cancel(@Nullable final Future<?> f) {
         if (f != null) {
             f.cancel(false);