DATAMONGO-1166 - ReadPreference is now be used for aggregations.

We now use MongoTemplate.readPreference(…) when executing commands such as geoNear(…) and aggregate(…).

Original pull request: spring-projects#303.

Author: christophstrobl

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java

@@ -335,7 +335,7 @@ public CommandResult executeCommand(final DBObject command, final int options) {
 
 		CommandResult result = execute(new DbCallback<CommandResult>() {
 			public CommandResult doInDB(DB db) throws MongoException, DataAccessException {
-				return db.command(command, options);
+				return readPreference != null ? db.command(command, readPreference) : db.command(command);
 			}
 		});
 
@@ -566,7 +566,7 @@ public <T> GeoResults<T> geoNear(NearQuery near, Class<T> entityClass, String co
 		BasicDBObject command = new BasicDBObject("geoNear", collection);
 		command.putAll(near.toDBObject());
 
-		CommandResult commandResult = executeCommand(command);
+		CommandResult commandResult = executeCommand(command, getDb().getOptions());
 		List<Object> results = (List<Object>) commandResult.get("results");
 		results = results == null ? Collections.emptyList() : results;
 
@@ -1422,7 +1422,7 @@ protected <O> AggregationResults<O> aggregate(Aggregation aggregation, String co
 			LOGGER.debug("Executing aggregation: {}", serializeToJsonSafely(command));
 		}
 
-		CommandResult commandResult = executeCommand(command);
+		CommandResult commandResult = executeCommand(command, getDb().getOptions());
 		handleCommandError(commandResult, command);
 
 		// map results

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateUnitTests.java

@@ -43,7 +43,9 @@
 import org.springframework.data.annotation.Id;
 import org.springframework.data.annotation.Version;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.geo.Point;
 import org.springframework.data.mongodb.MongoDbFactory;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.convert.CustomConversions;
 import org.springframework.data.mongodb.core.convert.DefaultDbRefResolver;
 import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
@@ -52,18 +54,21 @@
 import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
 import org.springframework.data.mongodb.core.query.BasicQuery;
 import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
 import org.springframework.test.util.ReflectionTestUtils;
 
 import com.mongodb.BasicDBObject;
 import com.mongodb.BasicDBObjectBuilder;
+import com.mongodb.CommandResult;
 import com.mongodb.DB;
 import com.mongodb.DBCollection;
 import com.mongodb.DBCursor;
 import com.mongodb.DBObject;
 import com.mongodb.Mongo;
 import com.mongodb.MongoException;
+import com.mongodb.ReadPreference;
 
 /**
  * Unit tests for {@link MongoTemplate}.
@@ -352,6 +357,74 @@ public void processDocument(DBObject dbObject) throws MongoException, DataAccess
 		assertThat(captor.getValue(), equalTo(new BasicDBObjectBuilder().add("foo", 1).get()));
 	}
 
+	/**
+	 * @see DATAMONGO-1166
+	 */
+	@Test
+	public void aggregateShouldHonorReadPreferenceWhenSet() {
+
+		CommandResult result = mock(CommandResult.class);
+
+		when(result.get("result")).thenReturn(Collections.emptySet());
+		when(db.command(Mockito.any(DBObject.class), Mockito.any(ReadPreference.class))).thenReturn(result);
+		when(db.command(Mockito.any(DBObject.class))).thenReturn(result);
+		template.setReadPreference(ReadPreference.secondary());
+
+		template.aggregate(Aggregation.newAggregation(Aggregation.unwind("foo")), "collection-1", Wrapper.class);
+
+		verify(this.db, times(1)).command(Mockito.any(DBObject.class), eq(ReadPreference.secondary()));
+	}
+
+	/**
+	 * @see DATAMONGO-1166
+	 */
+	@Test
+	public void aggregateShouldIgnoreReadPreferenceWhenNotSet() {
+
+		CommandResult result = mock(CommandResult.class);
+
+		when(result.get("result")).thenReturn(Collections.emptySet());
+		when(db.command(Mockito.any(DBObject.class), Mockito.any(ReadPreference.class))).thenReturn(result);
+		when(db.command(Mockito.any(DBObject.class))).thenReturn(result);
+
+		template.aggregate(Aggregation.newAggregation(Aggregation.unwind("foo")), "collection-1", Wrapper.class);
+
+		verify(this.db, times(1)).command(Mockito.any(DBObject.class));
+	}
+
+	/**
+	 * @see DATAMONGO-1166
+	 */
+	@Test
+	public void geoNearShouldHonorReadPreferenceWhenSet() {
+
+		when(db.command(Mockito.any(DBObject.class), Mockito.any(ReadPreference.class)))
+				.thenReturn(mock(CommandResult.class));
+		when(db.command(Mockito.any(DBObject.class))).thenReturn(mock(CommandResult.class));
+		template.setReadPreference(ReadPreference.secondary());
+
+		NearQuery query = NearQuery.near(new Point(1, 1));
+		template.geoNear(query, Wrapper.class);
+
+		verify(this.db, times(1)).command(Mockito.any(DBObject.class), eq(ReadPreference.secondary()));
+	}
+
+	/**
+	 * @see DATAMONGO-1166
+	 */
+	@Test
+	public void geoNearShouldIgnoreReadPreferenceWhenNotSet() {
+
+		when(db.command(Mockito.any(DBObject.class), Mockito.any(ReadPreference.class)))
+				.thenReturn(mock(CommandResult.class));
+		when(db.command(Mockito.any(DBObject.class))).thenReturn(mock(CommandResult.class));
+
+		NearQuery query = NearQuery.near(new Point(1, 1));
+		template.geoNear(query, Wrapper.class);
+
+		verify(this.db, times(1)).command(Mockito.any(DBObject.class));
+	}
+
 	class AutogenerateableId {
 
 		@Id BigInteger id;