Added Connection.open method. This will be useful when we want to ensure connection pool minimum size.

jyemin · jyemin · commit afb360a097b3 · 2013-06-13T10:05:48.000-04:00
diff --git a/driver/src/main/org/mongodb/connection/Connection.java b/driver/src/main/org/mongodb/connection/Connection.java
@@ -17,12 +17,47 @@
 package org.mongodb.connection;
 
 import org.bson.ByteBuf;
+import org.mongodb.annotations.NotThreadSafe;
 
 import java.util.List;
 
+/**
+ * A connection to a MongoDB server with blocking operations.
+ * <p>
+ * This class is not completely thread safe.  At most one thread can have an active call to sendMessage, and one thread an active call
+ * to receiveMessage.
+ * </p>
+ *
+ * @since 3.0
+ */
+@NotThreadSafe
 public interface Connection extends BaseConnection {
 
+    /**
+     * Open the connection.  This method can be called multiple times, and all but the first should be a no-op.
+     */
+    void open();
+
+    /**
+     * Send a message to the server. The connection may not make any attempt to validate the integrity of the message.
+     * <p>
+     * This method blocks until all bytes have been written.  This method is not thread safe: only one thread at a time can have an active
+     * call to this method.
+     * </p>
+     *
+     * @param byteBuffers the list of byte buffers to send.
+     */
     void sendMessage(final List<ByteBuf> byteBuffers);
 
+    /**
+     * Receive a response to a sent message from the server.
+     * <p>
+     * This method blocks until the entire message has been read. This method is not thread safe: only one thread at a time can have an
+     * active call to this method.
+     * </p>
+     *
+     * @param responseSettings the settings that the response should conform to.
+     * @return the response
+     */
     ResponseBuffers receiveMessage(final ResponseSettings responseSettings);
 }
diff --git a/driver/src/main/org/mongodb/connection/impl/AuthenticatingConnection.java b/driver/src/main/org/mongodb/connection/impl/AuthenticatingConnection.java
@@ -64,17 +64,22 @@ public ServerAddress getServerAddress() {
         return wrapped.getServerAddress();
     }
 
+    @Override
+    public void open() {
+        isTrue("open", !isClosed());
+        wrapped.open();
+        authenticateAll();
+    }
+
     @Override
     public void sendMessage(final List<ByteBuf> byteBuffers) {
         isTrue("open", wrapped != null);
-        authenticateAll();
         wrapped.sendMessage(byteBuffers);
     }
 
     @Override
     public ResponseBuffers receiveMessage(final ResponseSettings responseSettings) {
         isTrue("open", wrapped != null);
-        authenticateAll();
         return wrapped.receiveMessage(responseSettings);
     }
 
diff --git a/driver/src/main/org/mongodb/connection/impl/ConnectingServerConnection.java b/driver/src/main/org/mongodb/connection/impl/ConnectingServerConnection.java
@@ -45,6 +45,11 @@ public ServerDescription getDescription() {
         return serverDescription;
     }
 
+    @Override
+    public void open() {
+        connection.open();
+    }
+
     @Override
     public void sendMessage(final List<ByteBuf> byteBuffers) {
         connection.sendMessage(byteBuffers);
diff --git a/driver/src/main/org/mongodb/connection/impl/DefaultConnection.java b/driver/src/main/org/mongodb/connection/impl/DefaultConnection.java
@@ -38,6 +38,7 @@
 import java.nio.channels.ClosedByInterruptException;
 import java.util.List;
 
+import static org.mongodb.assertions.Assertions.isTrue;
 import static org.mongodb.connection.ReplyHeader.REPLY_HEADER_LENGTH;
 
 abstract class DefaultConnection implements Connection {
@@ -66,12 +67,14 @@ public ServerAddress getServerAddress() {
         return serverAddress;
     }
 
-    public DefaultConnectionSettings getSettings() {
-        return settings;
+    @Override
+    public void open() {
+       isTrue("open", !isClosed());
+       ensureOpen();
     }
 
     public void sendMessage(final List<ByteBuf> byteBuffers) {
-        check();
+        isTrue("open", !isClosed());
         try {
             write(byteBuffers);
         } catch (IOException e) {
@@ -82,7 +85,7 @@ public void sendMessage(final List<ByteBuf> byteBuffers) {
 
     @Override
     public ResponseBuffers receiveMessage(final ResponseSettings responseSettings) {
-        check();
+        isTrue("open", !isClosed());
         try {
             return receiveMessage(responseSettings, System.nanoTime());
         } catch (IOException e) {
@@ -149,10 +152,6 @@ private ResponseBuffers receiveMessage(final ResponseSettings responseSettings,
         return new ResponseBuffers(replyHeader, bodyByteBuffer, System.nanoTime() - start);
     }
 
-    private void check() {
-        ensureOpen();
-    }
-
     protected void initializeSocket(final Socket socket) throws IOException {
         socket.setTcpNoDelay(true);
         socket.setSoTimeout(settings.getReadTimeoutMS());
diff --git a/driver/src/main/org/mongodb/connection/impl/DefaultConnectionProvider.java b/driver/src/main/org/mongodb/connection/impl/DefaultConnectionProvider.java
@@ -62,6 +62,7 @@ public Connection get(final long timeout, final TimeUnit timeUnit) {
             if (connection == null) {
                 throw new MongoTimeoutException(String.format("Timeout waiting for a connection after %d %s", timeout, timeUnit));
             }
+            connection.open();
             return wrap(connection);
         } finally {
             waitQueueSize.decrementAndGet();
@@ -91,6 +92,7 @@ public SimpleConnectionPool(final ServerAddress serverAddress, final ConnectionF
         @Override
         protected Connection createNew() {
             return connectionFactory.create(serverAddress);
+
         }
 
         @Override
@@ -121,10 +123,16 @@ public boolean isClosed() {
 
         @Override
         public ServerAddress getServerAddress() {
-            isTrue("open", wrapped != null);
+            isTrue("open", !isClosed());
             return wrapped.getServerAddress();
         }
 
+        @Override
+        public void open() {
+            isTrue("open", !isClosed());
+            wrapped.open();
+        }
+
         @Override
         public void sendMessage(final List<ByteBuf> byteBuffers) {
             isTrue("open", wrapped != null);
diff --git a/driver/src/main/org/mongodb/connection/impl/DefaultServer.java b/driver/src/main/org/mongodb/connection/impl/DefaultServer.java
@@ -167,6 +167,17 @@ public ServerAddress getServerAddress() {
             return wrapped.getServerAddress();
         }
 
+        @Override
+        public void open() {
+            isTrue("open", !isClosed());
+            try {
+                wrapped.open();
+            } catch (MongoException e) {
+                handleException();
+                throw e;
+            }
+        }
+
         @Override
         public void sendMessage(final List<ByteBuf> byteBuffers) {
             isTrue("open", !isClosed());
diff --git a/driver/src/main/org/mongodb/connection/impl/IsMasterServerStateNotifier.java b/driver/src/main/org/mongodb/connection/impl/IsMasterServerStateNotifier.java
@@ -91,6 +91,7 @@ public synchronized void run() {
         try {
             if (connection == null) {
                 connection = connectionFactory.create(serverAddress);
+                connection.open();
             }
             try {
                 final CommandResult commandResult = new CommandOperation("admin",
diff --git a/driver/src/main/org/mongodb/session/DelayedCloseConnection.java b/driver/src/main/org/mongodb/session/DelayedCloseConnection.java
@@ -37,6 +37,12 @@ public DelayedCloseConnection(final ServerConnection wrapped) {
         this.wrapped = notNull("wrapped", wrapped);
     }
 
+    @Override
+    public void open() {
+        isTrue("open", !isClosed());
+        wrapped.open();
+    }
+
     @Override
     public void sendMessage(final List<ByteBuf> byteBuffers) {
         isTrue("open", !isClosed());
diff --git a/driver/src/test/functional/org/mongodb/MongoQueryCursorExhaustTest.java b/driver/src/test/functional/org/mongodb/MongoQueryCursorExhaustTest.java
@@ -163,6 +163,12 @@ public ServerAddress getServerAddress() {
             return getWrapped().getServerAddress();
         }
 
+        @Override
+        public void open() {
+            isTrue("open", !isClosed());
+            wrapped.open();
+        }
+
         @Override
         public void sendMessage(final List<ByteBuf> byteBuffers) {
             isTrue("open", !isClosed());
diff --git a/driver/src/test/unit/org/mongodb/connection/DefaultConnectionProviderTest.java b/driver/src/test/unit/org/mongodb/connection/DefaultConnectionProviderTest.java
@@ -115,6 +115,10 @@ private static class TestConnectionFactory implements ConnectionFactory {
         @Override
         public Connection create(final ServerAddress serverAddress) {
             return new Connection() {
+                @Override
+                public void open() {
+                }
+
                 @Override
                 public void sendMessage(final List<ByteBuf> byteBuffers) {
                     throw new UnsupportedOperationException();

Original file line number	Diff line number	Diff line change
`@@ -91,6 +91,7 @@ public synchronized void run() {`
`91`	`91`	`try {`
`92`	`92`	`if (connection == null) {`
`93`	`93`	`connection = connectionFactory.create(serverAddress);`
	`94`	`+ connection.open();`
`94`	`95`	`}`
`95`	`96`	`try {`
`96`	`97`	`final CommandResult commandResult = new CommandOperation("admin",`