Use the AsyncReadBinding's readPreference with the ReadConnectionSource

JAVA-1954 JAVA-1944

Author: rozza

driver-core/src/main/com/mongodb/operation/CommandOperationHelper.java

@@ -212,7 +212,7 @@ static <D, T> void executeWrappedCommandProtocolAsync(final String database, fin
                                                           final Function<D, T> transformer,
                                                           final SingleResultCallback<T> callback) {
         binding.getReadConnectionSource(new CommandProtocolExecutingCallback<D, T>(database, command, new NoOpFieldNameValidator(),
-                                                                                   decoder, primary(), transformer,
+                                                                                   decoder, binding.getReadPreference(), transformer,
                                                                                    errorHandlingCallback(callback)));
     }
 

driver-core/src/test/unit/com/mongodb/operation/CommandOperationHelperSpecification.groovy

@@ -15,15 +15,31 @@
  */
 
 package com.mongodb.operation
-
+import com.mongodb.Function
 import com.mongodb.MongoCommandException
+import com.mongodb.ReadPreference
 import com.mongodb.ServerAddress
+import com.mongodb.async.SingleResultCallback
+import com.mongodb.binding.AsyncConnectionSource
+import com.mongodb.binding.AsyncReadBinding
+import com.mongodb.binding.AsyncWriteBinding
+import com.mongodb.binding.ConnectionSource
+import com.mongodb.binding.ReadBinding
+import com.mongodb.binding.WriteBinding
+import com.mongodb.connection.AsyncConnection
+import com.mongodb.connection.ClusterId
+import com.mongodb.connection.Connection
+import com.mongodb.connection.ConnectionDescription
+import com.mongodb.connection.ServerId
 import org.bson.BsonBoolean
 import org.bson.BsonDocument
 import org.bson.BsonInt32
 import org.bson.BsonString
+import org.bson.codecs.Decoder
 import spock.lang.Specification
 
+import static com.mongodb.operation.CommandOperationHelper.executeWrappedCommandProtocol
+import static com.mongodb.operation.CommandOperationHelper.executeWrappedCommandProtocolAsync
 import static com.mongodb.operation.CommandOperationHelper.isNamespaceError
 import static com.mongodb.operation.CommandOperationHelper.rethrowIfNotNamespaceError
 
@@ -89,4 +105,121 @@ class CommandOperationHelperSpecification extends Specification {
                                                                      .append('code', new BsonInt32(26)),
                                                              new ServerAddress()), 'some value') == 'some value'
     }
+
+    def 'should set slaveOK to false when using WriteBinding'() {
+        given:
+        def dbName = "db"
+        def command = new BsonDocument()
+        def decoder = Stub(Decoder)
+        def writeBinding = Mock(WriteBinding)
+        def function = Mock(Function)
+        def connectionSource = Mock(ConnectionSource)
+        def connection = Mock(Connection)
+        def connectionDescription = new ConnectionDescription(new ServerId(new ClusterId("cluster"), new ServerAddress("localhost")))
+
+        when:
+        executeWrappedCommandProtocol(dbName, command, decoder, writeBinding, function)
+
+        then:
+        1 * writeBinding.getWriteConnectionSource() >> connectionSource
+
+        then:
+        1 * connectionSource.getConnection() >> connection
+        1 * connection.getDescription() >> connectionDescription
+
+        then:
+        1 * connection.command(dbName, command, false, _, decoder)
+
+        then:
+        1 * connection.release()
+        1 * function.apply(_)
+        1 * connectionSource.release()
+    }
+
+    def 'should use the ReadBindings readPreference to set slaveOK'() {
+        given:
+        def dbName = "db"
+        def command = new BsonDocument()
+        def decoder = Stub(Decoder)
+        def readBinding = Mock(ReadBinding)
+        def readPreference = Mock(ReadPreference)
+        def function = Mock(Function)
+        def connectionSource = Mock(ConnectionSource)
+        def connection = Mock(Connection)
+        def connectionDescription = new ConnectionDescription(new ServerId(new ClusterId("cluster"), new ServerAddress("localhost")))
+
+        when:
+        executeWrappedCommandProtocol(dbName, command, decoder, readBinding, function)
+
+        then:
+        1 * readBinding.getReadConnectionSource() >> connectionSource
+        1 * readBinding.getReadPreference() >> readPreference
+
+        then:
+        1 * connectionSource.getConnection() >> connection
+        1 * connection.getDescription() >> connectionDescription
+
+        then:
+        1 * readPreference.slaveOk >> true
+        1 * connection.command(dbName, command, true, _, decoder)
+
+        then:
+        1 * connection.release()
+        1 * function.apply(_)
+        1 * connectionSource.release()
+    }
+
+    def 'should set slaveOK to false when using AsyncWriteBinding'() {
+        given:
+        def dbName = "db"
+        def command = new BsonDocument()
+        def decoder = Stub(Decoder)
+        def asyncWriteBinding = Mock(AsyncWriteBinding)
+        def function = Mock(Function)
+        def callback = Stub(SingleResultCallback)
+        def connectionSource = Mock(AsyncConnectionSource)
+        def connection = Mock(AsyncConnection)
+        def connectionDescription = new ConnectionDescription(new ServerId(new ClusterId("cluster"), new ServerAddress("localhost")))
+
+        when:
+        executeWrappedCommandProtocolAsync(dbName, command, decoder, asyncWriteBinding, function, callback)
+
+        then:
+        1 * asyncWriteBinding.getWriteConnectionSource(_) >> { it[0].onResult(connectionSource, null) }
+        1 * connectionSource.getConnection(_) >> { it[0].onResult(connection, null)}
+        1 * connection.getDescription() >> connectionDescription
+        1 * connection.commandAsync(dbName, command, false, _, decoder, _) >> { it[5].onResult(1, null)}
+        1 * connection.release()
+        1 * connectionSource.release()
+    }
+
+    def 'should use the AsyncReadBindings readPreference to set slaveOK'() {
+        given:
+        def dbName = "db"
+        def command = new BsonDocument()
+        def decoder = Stub(Decoder)
+        def asyncReadBinding = Mock(AsyncReadBinding)
+        def readPreference = Mock(ReadPreference)
+        def function = Mock(Function)
+        def callback = Stub(SingleResultCallback)
+        def connectionSource = Mock(AsyncConnectionSource)
+        def connection = Mock(AsyncConnection)
+        def connectionDescription = new ConnectionDescription(new ServerId(new ClusterId("cluster"), new ServerAddress("localhost")))
+
+        when:
+        executeWrappedCommandProtocolAsync(dbName, command, decoder, asyncReadBinding, function, callback)
+
+        then:
+        1 * asyncReadBinding.getReadPreference() >> readPreference
+
+        then:
+        1 * asyncReadBinding.getReadConnectionSource(_) >> { it[0].onResult(connectionSource, null) }
+        1 * connectionSource.getConnection(_) >> { it[0].onResult(connection, null)}
+        1 * connection.getDescription() >> connectionDescription
+        1 * readPreference.slaveOk >> true
+        1 * connection.commandAsync(dbName, command, true, _, decoder, _) >> { it[5].onResult(1, null)}
+        1 * connection.release()
+        1 * connectionSource.release()
+    }
+
 }