JAVA-2227: When using aggregate helper methods that take an AggregationOptions parameter, ensure that slaveOk bit is set for non-primary read preferences

Author: jyemin

src/main/com/mongodb/DBCollectionImpl.java

@@ -97,7 +97,7 @@ public Cursor aggregate(final List<DBObject> pipeline, final AggregationOptions
 
         DBObject last = pipeline.get(pipeline.size() - 1);
         String outCollection = (String) last.get("$out");
-        ReadPreference appliedReadPreference = outCollection != null ? ReadPreference.primary() : readPreference;
+        final ReadPreference appliedReadPreference = outCollection != null ? ReadPreference.primary() : readPreference;
 
         final DBPort port = db.getConnector().getPort(appliedReadPreference);
         try {
@@ -106,7 +106,8 @@ public Cursor aggregate(final List<DBObject> pipeline, final AggregationOptions
             CommandResult res = db.getConnector().doOperation(db, port, new DBPort.Operation<CommandResult>() {
                 @Override
                 public CommandResult execute() throws IOException {
-                    return port.runCommand(db, command, db.getMongo().getMaxBsonObjectSize() + QUERY_DOCUMENT_HEADROOM);
+                    return port.runCommand(db, command, appliedReadPreference,
+                            db.getMongo().getMaxBsonObjectSize() + QUERY_DOCUMENT_HEADROOM);
                 }
             });
 

src/main/com/mongodb/DBPort.java

@@ -211,8 +211,14 @@ synchronized CommandResult runCommand( DB db , DBObject cmd ) throws IOException
     }
 
     synchronized CommandResult runCommand(final DB db, final DBObject cmd, final int maxBsonObjectSize) throws IOException {
+        return runCommand(db, cmd, null, maxBsonObjectSize);
+    }
+
+    synchronized CommandResult runCommand(final DB db, final DBObject cmd, final ReadPreference readPreference,
+                                          final int maxBsonObjectSize) throws IOException {
         isTrue("open", !closed);
-        OutMessage msg = OutMessage.query( db.getCollection("$cmd") , 0 , 0 , -1 , cmd , null, maxBsonObjectSize);
+        OutMessage msg = OutMessage.query( db.getCollection("$cmd") , 0 , 0 , -1 , cmd, null, readPreference,
+                DefaultDBEncoder.FACTORY.create(), maxBsonObjectSize);
         try {
             return convertToCommandResult(cmd, call(msg, db.getCollection("$cmd"), null));
         } finally {

src/test/com/mongodb/AggregationTest.java

@@ -19,7 +19,6 @@
 import com.mongodb.AggregationOptions.OutputMode;
 import com.mongodb.util.TestCase;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Test;
 
 import java.net.UnknownHostException;
@@ -188,7 +187,7 @@ public void testDollarOutOnSecondary() throws UnknownHostException {
         checkServerVersion(2.6);
         assumeTrue(isReplicaSet(cleanupMongo));
 
-        ServerAddress primary = new ServerAddress("localhost");
+        ServerAddress primary = new ServerAddress(getPrimaryAsString(cleanupMongo));
         MongoClient rsClient = new MongoClient(getMongoClientURI());
         DB rsDatabase = rsClient.getDB(database.getName());
         DBCollection aggCollection = rsDatabase.getCollection(collection.getName());
@@ -205,25 +204,27 @@ public void testDollarOutOnSecondary() throws UnknownHostException {
     }
 
     @Test
-    @Ignore
     public void testAggregateOnSecondary() throws UnknownHostException {
         checkServerVersion(2.6);
         assumeTrue(isReplicaSet(cleanupMongo));
 
-        ServerAddress primary = new ServerAddress("localhost");
-        ServerAddress secondary = new ServerAddress("localhost", 27018);
+        ServerAddress primary = new ServerAddress(getPrimaryAsString(cleanupMongo));
+        ServerAddress secondary = new ServerAddress(getASecondaryAsString(cleanupMongo));
         MongoClient rsClient = new MongoClient(asList(primary, secondary));
         DB rsDatabase = rsClient.getDB(database.getName());
         rsDatabase.dropDatabase();
         DBCollection aggCollection = rsDatabase.getCollection(collection.getName());
         aggCollection.drop();
 
-        final List<DBObject> pipeline = new ArrayList<DBObject>(prepareData());
+        List<DBObject> pipeline = new ArrayList<DBObject>(prepareData());
         AggregationOptions options = AggregationOptions.builder()
                 .outputMode(OutputMode.INLINE)
                 .build();
-        Cursor cursor = verify(pipeline, options, ReadPreference.secondary(), aggCollection);
+        Cursor cursor = aggCollection.aggregate(pipeline, options, ReadPreference.secondary());
         assertNotEquals(primary, cursor.getServerAddress());
+
+        AggregationOutput aggregationOutput = aggCollection.aggregate(pipeline, ReadPreference.secondary());
+        assertNotEquals(primary, aggregationOutput.getServerUsed());
     }
 
     @Test