Convert OpenChannelFutureListener to a static nested class

Explicit NettyStream dependency instead of the implicit enclosing instance,
removing the this$0 capture footgun that caused JAVA-6250. No behavioral change.

JAVA-6250

Author: rozza

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

@@ -221,7 +221,7 @@ public void initChannel(final SocketChannel ch) {
                 }
             });
             ChannelFuture channelFuture = bootstrap.connect(nextAddress);
-            channelFuture.addListener(new OpenChannelFutureListener(operationContext, socketAddressQueue, channelFuture, handler));
+            channelFuture.addListener(new OpenChannelFutureListener(this, operationContext, socketAddressQueue, channelFuture, handler));
         }
     }
 
@@ -507,15 +507,17 @@ public T get() throws IOException {
         }
     }
 
-    private class OpenChannelFutureListener implements ChannelFutureListener {
+    private static final class OpenChannelFutureListener implements ChannelFutureListener {
+        private final NettyStream stream;
         private final Queue<SocketAddress> socketAddressQueue;
         private final ChannelFuture channelFuture;
         private final AsyncCompletionHandler<Void> handler;
         private final OperationContext operationContext;
 
-        OpenChannelFutureListener(final OperationContext operationContext,
+        OpenChannelFutureListener(final NettyStream stream, final OperationContext operationContext,
                 final Queue<SocketAddress> socketAddressQueue, final ChannelFuture channelFuture,
                 final AsyncCompletionHandler<Void> handler) {
+            this.stream = stream;
             this.operationContext = operationContext;
             this.socketAddressQueue = socketAddressQueue;
             this.channelFuture = channelFuture;
@@ -524,24 +526,24 @@ private class OpenChannelFutureListener implements ChannelFutureListener {
 
         @Override
         public void operationComplete(final ChannelFuture future) {
-            withLock(lock, () -> {
+            withLock(stream.lock, () -> {
                 if (future.isSuccess()) {
-                    if (isClosed) {
+                    if (stream.isClosed) {
                         channelFuture.channel().close();
                     } else {
-                        channel = channelFuture.channel();
+                        stream.channel = channelFuture.channel();
                         // 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));
+                        stream.channel.closeFuture().addListener(new CloseChannelFutureListener(stream));
                     }
                     handler.completed(null);
                 } else {
-                    if (isClosed) {
+                    if (stream.isClosed) {
                         handler.completed(null);
                     } else if (socketAddressQueue.isEmpty()) {
-                        handler.failed(new MongoSocketOpenException("Exception opening socket", getAddress(), future.cause()));
+                        handler.failed(new MongoSocketOpenException("Exception opening socket", stream.getAddress(), future.cause()));
                     } else {
-                        initializeChannel(operationContext, handler, socketAddressQueue);
+                        stream.initializeChannel(operationContext, handler, socketAddressQueue);
                     }
                 }
             });