DATAMONGO-1682 - Add support partialFilterExpression for reactive index creation.

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-mongodb-parent</artifactId>
-	<version>2.0.0.BUILD-SNAPSHOT</version>
+	<version>2.0.0.DATAMONGO-1682-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data MongoDB</name>

spring-data-mongodb-benchmarks/pom.xml

@@ -7,7 +7,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.0.0.BUILD-SNAPSHOT</version>
+		<version>2.0.0.DATAMONGO-1682-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb-cross-store/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.0.0.BUILD-SNAPSHOT</version>
+		<version>2.0.0.DATAMONGO-1682-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 
@@ -48,7 +48,7 @@
 		<dependency>
 			<groupId>org.springframework.data</groupId>
 			<artifactId>spring-data-mongodb</artifactId>
-			<version>2.0.0.BUILD-SNAPSHOT</version>
+			<version>2.0.0.DATAMONGO-1682-SNAPSHOT</version>
 		</dependency>
 
 		<!-- reactive -->

spring-data-mongodb-distribution/pom.xml

@@ -13,7 +13,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.0.0.BUILD-SNAPSHOT</version>
+		<version>2.0.0.DATAMONGO-1682-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/pom.xml

@@ -11,7 +11,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.0.0.BUILD-SNAPSHOT</version>
+		<version>2.0.0.DATAMONGO-1682-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

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

@@ -23,8 +23,8 @@
 
 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;
@@ -38,7 +38,7 @@
 
 /**
  * Default implementation of {@link IndexOperations}.
- * 
+ *
  * @author Mark Pollack
  * @author Oliver Gierke
  * @author Komi Innocent
@@ -56,7 +56,7 @@ public class DefaultIndexOperations implements IndexOperations {
 
 	/**
 	 * Creates a new {@link DefaultIndexOperations}.
-	 * 
+	 *
 	 * @param mongoDbFactory must not be {@literal null}.
 	 * @param collectionName must not be {@literal null}.
 	 * @param queryMapper must not be {@literal null}.
@@ -98,24 +98,22 @@ public String ensureIndex(final IndexDefinition indexDefinition) {
 
 			Document indexOptions = indexDefinition.getIndexOptions();
 
-			if (indexOptions != null) {
-
-				IndexOptions ops = IndexConverters.indexDefinitionToIndexOptionsConverter().convert(indexDefinition);
+			if (indexOptions == null) {
+				return collection.createIndex(indexDefinition.getIndexKeys());
+			}
 
-				if (indexOptions.containsKey(PARTIAL_FILTER_EXPRESSION_KEY)) {
+			IndexOptions ops = IndexConverters.indexDefinitionToIndexOptionsConverter().convert(indexDefinition);
 
-					Assert.isInstanceOf(Document.class, indexOptions.get(PARTIAL_FILTER_EXPRESSION_KEY));
+			if (indexOptions.containsKey(PARTIAL_FILTER_EXPRESSION_KEY)) {
 
-					ops.partialFilterExpression( mapper.getMappedObject(
-							(Document) indexOptions.get(PARTIAL_FILTER_EXPRESSION_KEY), lookupPersistentEntity(type, collectionName)));
-				}
+				Assert.isInstanceOf(Document.class, indexOptions.get(PARTIAL_FILTER_EXPRESSION_KEY));
 
-				return collection.createIndex(indexDefinition.getIndexKeys(), ops);
+				ops.partialFilterExpression(mapper.getMappedObject((Document) indexOptions.get(PARTIAL_FILTER_EXPRESSION_KEY),
+						lookupPersistentEntity(type, collectionName)));
 			}
-			return collection.createIndex(indexDefinition.getIndexKeys());
-		}
 
-		);
+			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.
@@ -15,70 +15,122 @@
  */
 package org.springframework.data.mongodb.core;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.Collection;
+import java.util.Optional;
+
 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.reactivestreams.client.ListIndexesPublisher;
-
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
+import com.mongodb.client.model.IndexOptions;
 
 /**
- * Default implementation of {@link IndexOperations}.
+ * Default implementation of {@link ReactiveIndexOperations}.
  *
  * @author Mark Paluch
- * @since 1.11
+ * @author Christoph Strobl
+ * @since 2.0
  */
 public class DefaultReactiveIndexOperations implements ReactiveIndexOperations {
 
+	private static final String PARTIAL_FILTER_EXPRESSION_KEY = "partialFilterExpression";
+
 	private final ReactiveMongoOperations mongoOperations;
 	private final String collectionName;
+	private final QueryMapper queryMapper;
+	private final Optional<Class<?>> type;
+
+	/**
+	 * Creates a new {@link DefaultReactiveIndexOperations}.
+	 *
+	 * @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());
+	}
 
 	/**
 	 * Creates a new {@link DefaultReactiveIndexOperations}.
 	 *
 	 * @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) {
+	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!");
+		Assert.notNull(queryMapper, "QueryMapper must not be null!");
 
 		this.mongoOperations = mongoOperations;
 		this.collectionName = collectionName;
+		this.queryMapper = queryMapper;
+		this.type = type;
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.mongodb.core.ReactiveIndexOperations#ensureIndex(org.springframework.data.mongodb.core.index.IndexDefinition)
 	 */
 	public Mono<String> ensureIndex(final IndexDefinition indexDefinition) {
 
-		return mongoOperations.execute(collectionName, (ReactiveCollectionCallback<String>) collection -> {
+		return mongoOperations.execute(collectionName, collection -> {
 
 			Document indexOptions = indexDefinition.getIndexOptions();
 
-			if (indexOptions != null) {
-				return collection.createIndex(indexDefinition.getIndexKeys(),
-						IndexConverters.indexDefinitionToIndexOptionsConverter().convert(indexDefinition));
+			if (indexOptions == null) {
+				return collection.createIndex(indexDefinition.getIndexKeys());
 			}
 
-			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();
 	}
 
+	private MongoPersistentEntity<?> lookupPersistentEntity(String collection) {
+
+		Collection<? extends MongoPersistentEntity<?>> entities = queryMapper.getMappingContext().getPersistentEntities();
+
+		return entities.stream() //
+				.filter(entity -> entity.getCollection().equals(collection)) //
+				.findFirst() //
+				.orElse(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 -> {
-
-			return Mono.from(collection.dropIndex(name));
-		}).flatMap(success -> Mono.<Void>empty()).next();
+		return mongoOperations.execute(collectionName, collection -> collection.dropIndex(name)).then();
 	}
 
 	/* (non-Javadoc)
@@ -93,11 +145,7 @@ public Mono<Void> dropAllIndexes() {
 	 */
 	public Flux<IndexInfo> getIndexInfo() {
 
-		return mongoOperations.execute(collectionName, collection -> {
-
-			ListIndexesPublisher<Document> indexesPublisher = collection.listIndexes(Document.class);
-
-			return Flux.from(indexesPublisher).map(IndexConverters.documentToIndexInfoConverter()::convert);
-		});
+		return mongoOperations.execute(collectionName, collection -> collection.listIndexes(Document.class)) //
+				.map(IndexConverters.documentToIndexInfoConverter()::convert);
 	}
 }

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

@@ -548,7 +548,8 @@ public IndexOperations indexOps(String collectionName) {
 	}
 
 	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

@@ -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.
@@ -15,8 +15,6 @@
  */
 package org.springframework.data.mongodb.core;
 
-import java.util.List;
-
 import org.springframework.data.mongodb.core.index.IndexDefinition;
 import org.springframework.data.mongodb.core.index.IndexInfo;
 
@@ -58,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

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