Test asynchronous execution of WriteCommandProtocol and WriteProtocol

jyemin · jyemin · commit 2ac1ecf1239f · 2015-12-11T09:08:54.000-05:00
diff --git a/driver-core/src/test/functional/com/mongodb/connection/ProtocolTestHelper.java b/driver-core/src/test/functional/com/mongodb/connection/ProtocolTestHelper.java
@@ -0,0 +1,36 @@
+/*
+ * Copyright 2015 MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.mongodb.connection;
+
+import com.mongodb.ClusterFixture;
+import com.mongodb.async.FutureResultCallback;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+final class ProtocolTestHelper {
+    public static <T> T execute(final Protocol<T> protocol, final InternalConnection connection, final boolean async) throws Throwable{
+        if (async) {
+            final FutureResultCallback<T> futureResultCallback = new FutureResultCallback<T>();
+            protocol.executeAsync(connection, futureResultCallback);
+            return futureResultCallback.get(ClusterFixture.TIMEOUT, SECONDS);
+        } else {
+            return protocol.execute(connection);
+        }
+    }
+
+    private ProtocolTestHelper() {}
+}
diff --git a/driver-core/src/test/functional/com/mongodb/connection/WriteCommandProtocolSpecification.groovy b/driver-core/src/test/functional/com/mongodb/connection/WriteCommandProtocolSpecification.groovy
@@ -38,6 +38,7 @@ import static com.mongodb.ClusterFixture.getPrimary
 import static com.mongodb.ClusterFixture.getSslSettings
 import static com.mongodb.ClusterFixture.serverVersionAtLeast
 import static com.mongodb.WriteConcern.ACKNOWLEDGED
+import static com.mongodb.connection.ProtocolTestHelper.execute
 
 @IgnoreIf({ !serverVersionAtLeast([2, 6, 0]) })
 class WriteCommandProtocolSpecification extends OperationFunctionalSpecification {
@@ -47,7 +48,7 @@ class WriteCommandProtocolSpecification extends OperationFunctionalSpecification
 
     def setupSpec() {
         connection = new InternalStreamConnectionFactory(new NettyStreamFactory(SocketSettings.builder().build(), getSslSettings()),
-                                                         getCredentialList(), new NoOpConnectionListener())
+                getCredentialList(), new NoOpConnectionListener())
                 .create(new ServerId(new ClusterId(), getPrimary()))
         connection.open();
     }
@@ -62,13 +63,17 @@ class WriteCommandProtocolSpecification extends OperationFunctionalSpecification
 
         def insertRequest = [new InsertRequest(document)]
         def protocol = new InsertCommandProtocol(getNamespace(), true, ACKNOWLEDGED, null, insertRequest)
+
         when:
-        def result = protocol.execute(connection)
+        def result = execute(protocol, connection, async)
 
         then:
         result.insertedCount == 1
         result.upserts == []
         collectionHelper.find(document, new BsonDocumentCodec()).first() == document
+
+        where:
+        async << [false, true]
     }
 
     def 'should insert documents'() {
@@ -78,22 +83,25 @@ class WriteCommandProtocolSpecification extends OperationFunctionalSpecification
         def protocol = new InsertCommandProtocol(getNamespace(), true, ACKNOWLEDGED, null, requests)
 
         when:
-        protocol.execute(connection)
+        execute(protocol, connection, async)
 
         then:
         collectionHelper.count() == 2
+
+        where:
+        async << [false, true]
     }
 
     def 'should throw exception'() {
         given:
         def protocol = new InsertCommandProtocol(getNamespace(), false, ACKNOWLEDGED, false,
-                                                 [new InsertRequest(new BsonDocument('_id', new BsonInt32(1))),
-                                                  new InsertRequest(new BsonDocument('_id', new BsonInt32(2)))]
+                [new InsertRequest(new BsonDocument('_id', new BsonInt32(1))),
+                 new InsertRequest(new BsonDocument('_id', new BsonInt32(2)))]
         )
         protocol.execute(connection)
 
         when:
-        protocol.execute(connection)  // now do it again
+        execute(protocol, connection, async)
 
         then:
         def e = thrown(MongoBulkWriteException)
@@ -112,6 +120,9 @@ class WriteCommandProtocolSpecification extends OperationFunctionalSpecification
             code == 11000
             message != null
         }
+
+        where:
+        async << [false, true]
     }
 
     @Category(Slow)
@@ -133,11 +144,14 @@ class WriteCommandProtocolSpecification extends OperationFunctionalSpecification
         def protocol = new InsertCommandProtocol(getNamespace(), true, ACKNOWLEDGED, null, insertList)
 
         when:
-        def result = protocol.execute(connection)
+        def result = execute(protocol, connection, async)
 
         then:
         result.insertedCount == 4
         documents.size() == collectionHelper.count()
+
+        where:
+        async << [false, true]
     }
 
     @Category(Slow)
@@ -164,11 +178,14 @@ class WriteCommandProtocolSpecification extends OperationFunctionalSpecification
         def protocol = new InsertCommandProtocol(getNamespace(), true, ACKNOWLEDGED, null, insertList)
 
         when:
-        protocol.execute(connection)
+        execute(protocol, connection, async)
 
         then:
         def exception = thrown(MongoBulkWriteException)
         exception.writeResult.insertedCount == 1
+
+        where:
+        async << [false, true]
     }
 
     @Category(Slow)
@@ -198,7 +215,7 @@ class WriteCommandProtocolSpecification extends OperationFunctionalSpecification
         def protocol = new InsertCommandProtocol(getNamespace(), false, ACKNOWLEDGED, null, insertList)
 
         when:
-        protocol.execute(connection)
+        execute(protocol, connection, async)
 
         then:
         def e = thrown(MongoBulkWriteException)
@@ -207,26 +224,33 @@ class WriteCommandProtocolSpecification extends OperationFunctionalSpecification
         e.writeErrors[0].index == 0
         e.writeErrors[1].index == 2
         e.writeErrors[2].index == 3
+
+        where:
+        async << [false, true]
     }
 
     def 'should upsert items'() {
         given:
         def protocol = new UpdateCommandProtocol(getNamespace(), true, ACKNOWLEDGED, null,
-                                                 [new UpdateRequest(new BsonDocument('_id', new BsonInt32(1)),
-                                                                    new BsonDocument('$set', new BsonDocument('x', new BsonInt32(1))),
-                                                                    WriteRequest.Type.UPDATE)
-                                                          .upsert(true),
-                                                  new UpdateRequest(new BsonDocument('_id', new BsonInt32(2)),
-                                                                    new BsonDocument('$set', new BsonDocument('x', new BsonInt32(2))),
-                                                                    WriteRequest.Type.UPDATE)
-                                                          .upsert(true)]
+                [new UpdateRequest(new BsonDocument('_id', new BsonInt32(1)),
+                        new BsonDocument('$set', new BsonDocument('x', new BsonInt32(1))),
+                        WriteRequest.Type.UPDATE)
+                         .upsert(true),
+                 new UpdateRequest(new BsonDocument('_id', new BsonInt32(2)),
+                         new BsonDocument('$set', new BsonDocument('x', new BsonInt32(2))),
+                         WriteRequest.Type.UPDATE)
+                         .upsert(true)]
         );
 
         when:
-        def result = protocol.execute(connection);
+        def result = execute(protocol, connection, async)
+
 
         then:
         result.matchedCount == 0;
         result.upserts == [new BulkWriteUpsert(0, new BsonInt32(1)), new BulkWriteUpsert(1, new BsonInt32(2))]
+
+        where:
+        async << [false, true]
     }
 }
diff --git a/driver-core/src/test/functional/com/mongodb/connection/WriteProtocolSpecification.groovy b/driver-core/src/test/functional/com/mongodb/connection/WriteProtocolSpecification.groovy
@@ -35,6 +35,7 @@ import static com.mongodb.ClusterFixture.getPrimary
 import static com.mongodb.ClusterFixture.getSslSettings
 import static com.mongodb.WriteConcern.ACKNOWLEDGED
 import static com.mongodb.WriteConcern.UNACKNOWLEDGED
+import static com.mongodb.connection.ProtocolTestHelper.execute
 
 class WriteProtocolSpecification extends OperationFunctionalSpecification {
     @Shared
@@ -65,7 +66,7 @@ class WriteProtocolSpecification extends OperationFunctionalSpecification {
         getCollectionHelper().insertDocuments(documentOne)
 
         when:
-        protocol.execute(connection)
+        execute(protocol, connection, async)
 
         then:
         getCollectionHelper().find(new BsonDocumentCodec()) == [documentOne]
@@ -74,6 +75,9 @@ class WriteProtocolSpecification extends OperationFunctionalSpecification {
         // force acknowledgement
         new CommandProtocol(getDatabaseName(), new BsonDocument('drop', new BsonString(getCollectionName())),
                             new NoOpFieldNameValidator(), new BsonDocumentCodec()).execute(connection)
+
+        where:
+        async << [false, true]
     }
 
     def 'should report write errors on acknowledged inserts'() {
@@ -90,13 +94,62 @@ class WriteProtocolSpecification extends OperationFunctionalSpecification {
         getCollectionHelper().insertDocuments(documentOne)
 
         when:
-        protocol.execute(connection)
+        execute(protocol, connection, async)
 
         then:
         thrown(DuplicateKeyException)
         getCollectionHelper().count() == 1
+
+        where:
+        async << [false, true]
+    }
+
+    def 'should execute split unacknowledged inserts'() {
+        given:
+        def binary = new BsonBinary(new byte[15000000])
+        def documentOne = new BsonDocument('_id', new BsonInt32(1)).append('b', binary)
+        def documentTwo = new BsonDocument('_id', new BsonInt32(2)).append('b', binary)
+        def documentThree = new BsonDocument('_id', new BsonInt32(3)).append('b', binary)
+        def documentFour = new BsonDocument('_id', new BsonInt32(4)).append('b', binary)
+
+        def insertRequest = [new InsertRequest(documentOne), new InsertRequest(documentTwo),
+                             new InsertRequest(documentThree), new InsertRequest(documentFour)]
+        def protocol = new InsertProtocol(getNamespace(), true, UNACKNOWLEDGED, insertRequest)
+
+        when:
+        execute(protocol, connection, async)
+        // force acknowledgement
+        new CommandProtocol(getDatabaseName(), new BsonDocument('ping', new BsonInt32(1)),
+                new NoOpFieldNameValidator(), new BsonDocumentCodec()).execute(connection)
+
+        then:
+        getCollectionHelper().count() == 4
+
+        where:
+        async << [false, true]
     }
 
+    def 'should execute split acknowledged inserts'() {
+        given:
+        def binary = new BsonBinary(new byte[15000000])
+        def documentOne = new BsonDocument('_id', new BsonInt32(1)).append('b', binary)
+        def documentTwo = new BsonDocument('_id', new BsonInt32(2)).append('b', binary)
+        def documentThree = new BsonDocument('_id', new BsonInt32(3)).append('b', binary)
+        def documentFour = new BsonDocument('_id', new BsonInt32(4)).append('b', binary)
+
+        def insertRequest = [new InsertRequest(documentOne), new InsertRequest(documentTwo),
+                             new InsertRequest(documentThree), new InsertRequest(documentFour)]
+        def protocol = new InsertProtocol(getNamespace(), true, ACKNOWLEDGED, insertRequest)
+
+        when:
+        execute(protocol, connection, async)
+
+        then:
+        getCollectionHelper().count() == 4
+
+        where:
+        async << [false, true]
+    }
     def 'should ignore write errors on split unacknowledged inserts'() {
         given:
         def binary = new BsonBinary(new byte[15000000])
@@ -112,7 +165,7 @@ class WriteProtocolSpecification extends OperationFunctionalSpecification {
         getCollectionHelper().insertDocuments(documentOne)
 
         when:
-        protocol.execute(connection)
+        execute(protocol, connection, async)
 
         then:
         getCollectionHelper().count() == 1
@@ -121,6 +174,9 @@ class WriteProtocolSpecification extends OperationFunctionalSpecification {
         // force acknowledgement
         new CommandProtocol(getDatabaseName(), new BsonDocument('drop', new BsonString(getCollectionName())),
                             new NoOpFieldNameValidator(), new BsonDocumentCodec()).execute(connection)
+
+        where:
+        async << [false, true]
     }
 
     def 'should report write errors on split acknowledged inserts'() {
@@ -138,10 +194,13 @@ class WriteProtocolSpecification extends OperationFunctionalSpecification {
         getCollectionHelper().insertDocuments(documentOne)
 
         when:
-        protocol.execute(connection)
+        execute(protocol, connection, async)
 
         then:
         thrown(DuplicateKeyException)
         getCollectionHelper().count() == 1
+
+        where:
+        async << [false, true]
     }
 }
