Ensure that socket is closed on handshake failure

JAVA-3680

Author: jyemin

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

@@ -269,7 +269,7 @@ public ServerAddress getAddress() {
     }
 
     @Override
-    public void close() {
+    public synchronized void close() {
         isClosed = true;
         if (channel != null) {
             channel.close();
@@ -395,20 +395,28 @@ private class OpenChannelFutureListener implements ChannelFutureListener {
 
         @Override
         public void operationComplete(final ChannelFuture future) {
-            if (future.isSuccess()) {
-                channel = channelFuture.channel();
-                channel.closeFuture().addListener(new ChannelFutureListener() {
-                    @Override
-                    public void operationComplete(final ChannelFuture future) {
-                        handleReadResponse(null, new IOException("The connection to the server was closed"));
+            synchronized (NettyStream.this) {
+                if (future.isSuccess()) {
+                    if (isClosed) {
+                        channelFuture.channel().close();
+                    } else {
+                        channel = channelFuture.channel();
+                        channel.closeFuture().addListener(new ChannelFutureListener() {
+                            @Override
+                            public void operationComplete(final ChannelFuture future) {
+                                handleReadResponse(null, new IOException("The connection to the server was closed"));
+                            }
+                        });
                     }
-                });
-                handler.completed(null);
-            } else {
-                if (socketAddressQueue.isEmpty()) {
-                    handler.failed(new MongoSocketOpenException("Exception opening socket", getAddress(), future.cause()));
+                    handler.completed(null);
                 } else {
-                    initializeChannel(handler, socketAddressQueue);
+                    if (isClosed) {
+                        handler.completed(null);
+                    } else if (socketAddressQueue.isEmpty()) {
+                        handler.failed(new MongoSocketOpenException("Exception opening socket", getAddress(), future.cause()));
+                    } else {
+                        initializeChannel(handler, socketAddressQueue);
+                    }
                 }
             }
         }

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

@@ -69,13 +69,17 @@ public BufferProvider getBufferProvider() {
         return bufferProvider;
     }
 
-    public ExtendedAsynchronousByteChannel getChannel() {
+    public synchronized ExtendedAsynchronousByteChannel getChannel() {
         return channel;
     }
 
-    protected void setChannel(final ExtendedAsynchronousByteChannel channel) {
+    protected synchronized void setChannel(final ExtendedAsynchronousByteChannel channel) {
         isTrue("current channel is null", this.channel == null);
-        this.channel = channel;
+        if (isClosed) {
+            closeChannel(channel);
+        } else {
+            this.channel = channel;
+        }
     }
 
     @Override
@@ -133,16 +137,22 @@ public ServerAddress getAddress() {
     }
 
     @Override
-    public void close() {
+    public synchronized void close() {
+        isClosed = true;
+        try {
+            closeChannel(channel);
+        } finally {
+            channel = null;
+        }
+    }
+
+    private void closeChannel(final ExtendedAsynchronousByteChannel channel) {
         try {
             if (channel != null) {
                 channel.close();
             }
         } catch (IOException e) {
             // ignore
-        } finally {
-            channel = null;
-            isClosed = true;
         }
     }