DATAMONGO-1682 - Polishing.

Require non-null arguments in DefaultReactiveIndexOperations constructor. Remove superfluous publisher creation indirections. Use StepVerifier.verifyComplete() to verify the step sequence.

Use provided entity type in template API to construct index operations.

Original pull request: #474.

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

@@ -23,8 +23,8 @@ import java.util.List;
 
 import org.bson.Document;
 import org.springframework.dao.DataAccessException;
-import org.springframework.data.mongodb.core.convert.QueryMapper;
 import org.springframework.data.mongodb.MongoDbFactory;
+import org.springframework.data.mongodb.core.convert.QueryMapper;
 import org.springframework.data.mongodb.core.index.IndexDefinition;
 import org.springframework.data.mongodb.core.index.IndexInfo;
 import org.springframework.data.mongodb.core.index.IndexOperations;
@@ -98,24 +98,22 @@ public class DefaultIndexOperations implements IndexOperations {
 
 			Document indexOptions = indexDefinition.getIndexOptions();
 
-			if (indexOptions != null) {
+			if (indexOptions == null) {
-
+				return collection.createIndex(indexDefinition.getIndexKeys());
-				IndexOptions ops = IndexConverters.indexDefinitionToIndexOptionsConverter().convert(indexDefinition);
-
-				if (indexOptions.containsKey(PARTIAL_FILTER_EXPRESSION_KEY)) {
-
-					Assert.isInstanceOf(Document.class, indexOptions.get(PARTIAL_FILTER_EXPRESSION_KEY));
-
-					ops.partialFilterExpression( mapper.getMappedObject(
-							(Document) indexOptions.get(PARTIAL_FILTER_EXPRESSION_KEY), lookupPersistentEntity(type, collectionName)));
-				}
-
-				return collection.createIndex(indexDefinition.getIndexKeys(), ops);
 			}
-			return collection.createIndex(indexDefinition.getIndexKeys());
-		}
 
-		);
+			IndexOptions ops = IndexConverters.indexDefinitionToIndexOptionsConverter().convert(indexDefinition);
+
+			if (indexOptions.containsKey(PARTIAL_FILTER_EXPRESSION_KEY)) {
+
+				Assert.isInstanceOf(Document.class, indexOptions.get(PARTIAL_FILTER_EXPRESSION_KEY));
+
+				ops.partialFilterExpression(mapper.getMappedObject((Document) indexOptions.get(PARTIAL_FILTER_EXPRESSION_KEY),
+						lookupPersistentEntity(type, collectionName)));
+			}
+
+			return collection.createIndex(indexDefinition.getIndexKeys(), ops);
+		});
 	}
 
 	private MongoPersistentEntity<?> lookupPersistentEntity(Class<?> entityType, String collection) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -25,15 +25,13 @@ import org.bson.Document;
 import org.springframework.data.mongodb.core.convert.QueryMapper;
 import org.springframework.data.mongodb.core.index.IndexDefinition;
 import org.springframework.data.mongodb.core.index.IndexInfo;
-import org.springframework.data.mongodb.core.index.IndexOperations;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
 import org.springframework.util.Assert;
 
 import com.mongodb.client.model.IndexOptions;
-import com.mongodb.reactivestreams.client.ListIndexesPublisher;
 
 /**
- * Default implementation of {@link IndexOperations}.
+ * Default implementation of {@link ReactiveIndexOperations}.
  *
  * @author Mark Paluch
  * @author Christoph Strobl
@@ -53,11 +51,11 @@ public class DefaultReactiveIndexOperations implements ReactiveIndexOperations {
 	 *
 	 * @param mongoOperations must not be {@literal null}.
 	 * @param collectionName must not be {@literal null}.
+	 * @param queryMapper must not be {@literal null}.
 	 */
 	public DefaultReactiveIndexOperations(ReactiveMongoOperations mongoOperations, String collectionName,
 			QueryMapper queryMapper) {
-
+		this(mongoOperations, collectionName, queryMapper, Optional.empty());
-		this(mongoOperations, collectionName, queryMapper, null);
 	}
 
 	/**
@@ -65,9 +63,16 @@ public class DefaultReactiveIndexOperations implements ReactiveIndexOperations {
 	 *
 	 * @param mongoOperations must not be {@literal null}.
 	 * @param collectionName must not be {@literal null}.
+	 * @param queryMapper must not be {@literal null}.
+	 * @param type used for mapping potential partial index filter expression, must not be {@literal null}.
 	 */
 	public DefaultReactiveIndexOperations(ReactiveMongoOperations mongoOperations, String collectionName,
 			QueryMapper queryMapper, Class<?> type) {
+		this(mongoOperations, collectionName, queryMapper, Optional.of(type));
+	}
+
+	private DefaultReactiveIndexOperations(ReactiveMongoOperations mongoOperations, String collectionName,
+			QueryMapper queryMapper, Optional<Class<?>> type) {
 
 		Assert.notNull(mongoOperations, "ReactiveMongoOperations must not be null!");
 		Assert.notNull(collectionName, "Collection must not be null!");
@@ -76,7 +81,7 @@ public class DefaultReactiveIndexOperations implements ReactiveIndexOperations {
 		this.mongoOperations = mongoOperations;
 		this.collectionName = collectionName;
 		this.queryMapper = queryMapper;
-		this.type = Optional.ofNullable(type);
+		this.type = type;
 	}
 
 	/* (non-Javadoc)
@@ -88,26 +93,26 @@ public class DefaultReactiveIndexOperations implements ReactiveIndexOperations {
 
 			Document indexOptions = indexDefinition.getIndexOptions();
 
-			if (indexOptions != null) {
+			if (indexOptions == null) {
-
+				return collection.createIndex(indexDefinition.getIndexKeys());
-				IndexOptions ops = IndexConverters.indexDefinitionToIndexOptionsConverter().convert(indexDefinition);
-
-				if (indexOptions.containsKey(PARTIAL_FILTER_EXPRESSION_KEY)) {
-
-					Assert.isInstanceOf(Document.class, indexOptions.get(PARTIAL_FILTER_EXPRESSION_KEY));
-
-					MongoPersistentEntity<?> entity = type
-							.map(val -> (MongoPersistentEntity) queryMapper.getMappingContext().getRequiredPersistentEntity(val))
-							.orElseGet(() -> lookupPersistentEntity(collectionName));
-
-					ops = ops.partialFilterExpression(
-							queryMapper.getMappedObject((Document) indexOptions.get(PARTIAL_FILTER_EXPRESSION_KEY), entity));
-				}
-
-				return collection.createIndex(indexDefinition.getIndexKeys(), ops);
 			}
 
-			return collection.createIndex(indexDefinition.getIndexKeys());
+			IndexOptions ops = IndexConverters.indexDefinitionToIndexOptionsConverter().convert(indexDefinition);
+
+			if (indexOptions.containsKey(PARTIAL_FILTER_EXPRESSION_KEY)) {
+
+				Assert.isInstanceOf(Document.class, indexOptions.get(PARTIAL_FILTER_EXPRESSION_KEY));
+
+				MongoPersistentEntity<?> entity = type
+						.map(val -> (MongoPersistentEntity) queryMapper.getMappingContext().getRequiredPersistentEntity(val))
+						.orElseGet(() -> lookupPersistentEntity(collectionName));
+
+				ops = ops.partialFilterExpression(
+						queryMapper.getMappedObject(indexOptions.get(PARTIAL_FILTER_EXPRESSION_KEY, Document.class), entity));
+			}
+
+			return collection.createIndex(indexDefinition.getIndexKeys(), ops);
+
 		}).next();
 	}
 
@@ -115,24 +120,17 @@ public class DefaultReactiveIndexOperations implements ReactiveIndexOperations {
 
 		Collection<? extends MongoPersistentEntity<?>> entities = queryMapper.getMappingContext().getPersistentEntities();
 
-		for (MongoPersistentEntity<?> entity : entities) {
+		return entities.stream() //
-			if (entity.getCollection().equals(collection)) {
+				.filter(entity -> entity.getCollection().equals(collection)) //
-				return entity;
+				.findFirst() //
-			}
+				.orElse(null);
-		}
-
-		return null;
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.mongodb.core.ReactiveIndexOperations#dropIndex(java.lang.String)
 	 */
 	public Mono<Void> dropIndex(final String name) {
-
+		return mongoOperations.execute(collectionName, collection -> collection.dropIndex(name)).then();
-		return mongoOperations.execute(collectionName, collection -> {
-
-			return Mono.from(collection.dropIndex(name));
-		}).flatMap(success -> Mono.<Void> empty()).next();
 	}
 
 	/* (non-Javadoc)
@@ -147,11 +145,7 @@ public class DefaultReactiveIndexOperations implements ReactiveIndexOperations {
 	 */
 	public Flux<IndexInfo> getIndexInfo() {
 
-		return mongoOperations.execute(collectionName, collection -> {
+		return mongoOperations.execute(collectionName, collection -> collection.listIndexes(Document.class)) //
-
+				.map(IndexConverters.documentToIndexInfoConverter()::convert);
-			ListIndexesPublisher<Document> indexesPublisher = collection.listIndexes(Document.class);
-
-			return Flux.from(indexesPublisher).map(IndexConverters.documentToIndexInfoConverter()::convert);
-		});
 	}
 }

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

@@ -548,7 +548,8 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 	}
 
 	public IndexOperations indexOps(Class<?> entityClass) {
-		return new DefaultIndexOperations(getMongoDbFactory(), determineCollectionName(entityClass), queryMapper);
+		return new DefaultIndexOperations(getMongoDbFactory(), determineCollectionName(entityClass), queryMapper,
+				entityClass);
 	}
 
 	public BulkOperations bulkOps(BulkMode bulkMode, String collectionName) {

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

@@ -56,5 +56,4 @@ public interface ReactiveIndexOperations {
 	 * @return index information on the collection
 	 */
 	Flux<IndexInfo> getIndexInfo();
-
 }

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

@@ -309,7 +309,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 	 * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#reactiveIndexOps(java.lang.Class)
 	 */
 	public ReactiveIndexOperations indexOps(Class<?> entityClass) {
-		return new DefaultReactiveIndexOperations(this, determineCollectionName(entityClass), this.queryMapper);
+		return new DefaultReactiveIndexOperations(this, determineCollectionName(entityClass), this.queryMapper,
+				entityClass);
 	}
 
 	public String getCollectionName(Class<?> entityClass) {
@@ -1915,8 +1916,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 					"No class parameter provided, entity collection can't be determined!");
 		}
 
-		MongoPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(entityClass);
+		return mappingContext.getRequiredPersistentEntity(entityClass).getCollection();
-		return entity.getCollection();
 	}
 
 	private static MappingMongoConverter getDefaultMongoConverter() {

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

@@ -16,12 +16,10 @@
 package org.springframework.data.mongodb.core;
 
 import static org.assertj.core.api.Assertions.*;
-import static org.hamcrest.core.Is.*;
 import static org.junit.Assume.*;
 import static org.springframework.data.mongodb.core.index.PartialIndexFilter.*;
 import static org.springframework.data.mongodb.core.query.Criteria.*;
 
-import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import java.util.function.Predicate;
@@ -34,13 +32,13 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.data.domain.Sort.Direction;
 import org.springframework.data.mongodb.config.AbstractReactiveMongoConfiguration;
-import org.springframework.data.mongodb.core.query.Collation;
-import org.springframework.data.mongodb.core.query.Collation.CaseFirst;
 import org.springframework.data.mongodb.core.convert.QueryMapper;
 import org.springframework.data.mongodb.core.index.Index;
 import org.springframework.data.mongodb.core.index.IndexDefinition;
 import org.springframework.data.mongodb.core.index.IndexInfo;
 import org.springframework.data.mongodb.core.mapping.Field;
+import org.springframework.data.mongodb.core.query.Collation;
+import org.springframework.data.mongodb.core.query.Collation.CaseFirst;
 import org.springframework.data.util.Version;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -87,10 +85,10 @@ public class DefaultReactiveIndexOperationsTests {
 		String collectionName = this.template.getCollectionName(DefaultIndexOperationsIntegrationTestsSample.class);
 
 		this.collection = this.template.getMongoDatabase().getCollection(collectionName, Document.class);
-		Mono.from(this.collection.dropIndexes()).subscribe();
-
 		this.indexOps = new DefaultReactiveIndexOperations(template, collectionName,
 				new QueryMapper(template.getConverter()));
+
+		StepVerifier.create(this.collection.dropIndexes()).expectNextCount(1).verifyComplete();
 	}
 
 	private void queryMongoVersionIfNecessary() {
@@ -104,12 +102,12 @@ public class DefaultReactiveIndexOperationsTests {
 	@Test // DATAMONGO-1518
 	public void shouldCreateIndexWithCollationCorrectly() {
 
-		assumeThat(mongoVersion.isGreaterThanOrEqualTo(THREE_DOT_FOUR), is(true));
+		assumeTrue(mongoVersion.isGreaterThanOrEqualTo(THREE_DOT_FOUR));
 
 		IndexDefinition id = new Index().named("with-collation").on("xyz", Direction.ASC)
 				.collation(Collation.of("de_AT").caseFirst(CaseFirst.off()));
 
-		indexOps.ensureIndex(id).subscribe();
+		StepVerifier.create(indexOps.ensureIndex(id)).expectNextCount(1).verifyComplete();
 
 		Document expected = new Document("locale", "de_AT") //
 				.append("caseLevel", false) //
@@ -132,64 +130,64 @@ public class DefaultReactiveIndexOperationsTests {
 
 					assertThat(result).isEqualTo(expected);
 				}) //
-				.thenAwait();
+				.verifyComplete();
 	}
 
 	@Test // DATAMONGO-1682
 	public void shouldApplyPartialFilterCorrectly() {
 
-		assumeThat(mongoVersion.isGreaterThanOrEqualTo(THREE_DOT_TWO), is(true));
+		assumeTrue(mongoVersion.isGreaterThanOrEqualTo(THREE_DOT_TWO));
 
 		IndexDefinition id = new Index().named("partial-with-criteria").on("k3y", Direction.ASC)
 				.partial(of(where("q-t-y").gte(10)));
 
-		indexOps.ensureIndex(id).subscribe();
+		StepVerifier.create(indexOps.ensureIndex(id)).expectNextCount(1).verifyComplete();
 
 		StepVerifier.create(indexOps.getIndexInfo().filter(this.indexByName("partial-with-criteria"))) //
 				.consumeNextWith(indexInfo -> {
 					assertThat(indexInfo.getPartialFilterExpression()).isEqualTo("{ \"q-t-y\" : { \"$gte\" : 10 } }");
 				}) //
-				.thenAwait();
+				.verifyComplete();
 	}
 
 	@Test // DATAMONGO-1682
 	public void shouldApplyPartialFilterWithMappedPropertyCorrectly() {
 
-		assumeThat(mongoVersion.isGreaterThanOrEqualTo(THREE_DOT_TWO), is(true));
+		assumeTrue(mongoVersion.isGreaterThanOrEqualTo(THREE_DOT_TWO));
 
 		IndexDefinition id = new Index().named("partial-with-mapped-criteria").on("k3y", Direction.ASC)
 				.partial(of(where("quantity").gte(10)));
 
-		indexOps.ensureIndex(id).subscribe();
+		StepVerifier.create(indexOps.ensureIndex(id)).expectNextCount(1).verifyComplete();
 
 		StepVerifier.create(indexOps.getIndexInfo().filter(this.indexByName("partial-with-mapped-criteria"))) //
 				.consumeNextWith(indexInfo -> {
 					assertThat(indexInfo.getPartialFilterExpression()).isEqualTo("{ \"qty\" : { \"$gte\" : 10 } }");
-				}).thenAwait();
+				}).verifyComplete();
 	}
 
 	@Test // DATAMONGO-1682
 	public void shouldApplyPartialDBOFilterCorrectly() {
 
-		assumeThat(mongoVersion.isGreaterThanOrEqualTo(THREE_DOT_TWO), is(true));
+		assumeTrue(mongoVersion.isGreaterThanOrEqualTo(THREE_DOT_TWO));
 
 		IndexDefinition id = new Index().named("partial-with-dbo").on("k3y", Direction.ASC)
 				.partial(of(new org.bson.Document("qty", new org.bson.Document("$gte", 10))));
 
-		indexOps.ensureIndex(id).subscribe();
+		StepVerifier.create(indexOps.ensureIndex(id)).expectNextCount(1).verifyComplete();
 
 		StepVerifier.create(indexOps.getIndexInfo().filter(this.indexByName("partial-with-dbo"))) //
 				.consumeNextWith(indexInfo -> {
 					assertThat(indexInfo.getPartialFilterExpression()).isEqualTo("{ \"qty\" : { \"$gte\" : 10 } }");
 				}) //
-				.thenAwait();
+				.verifyComplete();
 
 	}
 
 	@Test // DATAMONGO-1682
 	public void shouldFavorExplicitMappingHintViaClass() {
 
-		assumeThat(mongoVersion.isGreaterThanOrEqualTo(THREE_DOT_TWO), is(true));
+		assumeTrue(mongoVersion.isGreaterThanOrEqualTo(THREE_DOT_TWO));
 
 		IndexDefinition id = new Index().named("partial-with-inheritance").on("k3y", Direction.ASC)
 				.partial(of(where("age").gte(10)));
@@ -198,7 +196,7 @@ public class DefaultReactiveIndexOperationsTests {
 				this.template.getCollectionName(DefaultIndexOperationsIntegrationTestsSample.class),
 				new QueryMapper(template.getConverter()), MappingToSameCollection.class);
 
-		indexOps.ensureIndex(id).subscribe();
+		StepVerifier.create(indexOps.ensureIndex(id)).expectNextCount(1).verifyComplete();
 
 		StepVerifier.create(indexOps.getIndexInfo().filter(this.indexByName("partial-with-inheritance"))) //
 				.consumeNextWith(indexInfo -> {