DATAMONGO-1869 - Release version 2.1 M2 (Lovelace).

Related tickets: DATACMNS-1275.

.travis.yml

@@ -5,6 +5,14 @@ jdk:
 
 before_script:
   - mongod --version
+  - |-
+    echo "replication:
+      replSetName: rs0" | sudo tee -a /etc/mongod.conf
+  - sudo service mongod restart
+  - sleep 20
+  - |-
+    mongo --eval "rs.initiate({_id: 'rs0', members:[{_id: 0, host: '127.0.0.1:27017'}]});"
+  - sleep 15
 
 services:
   - mongodb
@@ -31,6 +39,8 @@ cache:
   directories:
     - $HOME/.m2
 
-install: true
+install:
+  - |-
+    mongo admin --eval  "db.adminCommand({setFeatureCompatibilityVersion: '3.4'});"
 
 script: "mvn clean dependency:list test -P${PROFILE} -Dsort"

README.md

@@ -83,7 +83,7 @@ You can have Spring automatically create a proxy for the interface by using the
 class ApplicationConfig extends AbstractMongoConfiguration {
 
   @Override
-  public Mongo mongo() throws Exception {
+  public MongoClient mongoClient() throws Exception {
     return new MongoClient();
   }
 

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-mongodb-parent</artifactId>
-	<version>2.0.0.RELEASE</version>
+	<version>2.1.0.M2</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data MongoDB</name>
@@ -15,7 +15,7 @@
 	<parent>
 		<groupId>org.springframework.data.build</groupId>
 		<artifactId>spring-data-parent</artifactId>
-		<version>2.0.0.RELEASE</version>
+		<version>2.1.0.M2</version>
 	</parent>
 
 	<modules>
@@ -27,9 +27,9 @@
 	<properties>
 		<project.type>multi</project.type>
 		<dist.id>spring-data-mongodb</dist.id>
-		<springdata.commons>2.0.0.RELEASE</springdata.commons>
-		<mongo>3.5.0</mongo>
-		<mongo.reactivestreams>1.6.0</mongo.reactivestreams>
+		<springdata.commons>2.1.0.M2</springdata.commons>
+		<mongo>3.6.3</mongo>
+		<mongo.reactivestreams>1.7.1</mongo.reactivestreams>
 		<jmh.version>1.19</jmh.version>
 	</properties>
 
@@ -183,8 +183,8 @@
 
 	<repositories>
 		<repository>
-			<id>spring-libs-release</id>
-			<url>https://repo.spring.io/libs-release</url>
+			<id>spring-libs-milestone</id>
+			<url>https://repo.spring.io/libs-milestone</url>
 		</repository>
 	</repositories>
 
@@ -193,6 +193,11 @@
 			<id>spring-plugins-release</id>
 			<url>https://repo.spring.io/plugins-release</url>
 		</pluginRepository>
+		<pluginRepository>
+			<id>spring-libs-milestone</id>
+			<url>https://repo.spring.io/libs-milestone</url>
+		</pluginRepository>
+
 	</pluginRepositories>
 
 </project>

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.RELEASE</version>
+		<version>2.1.0.M2</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

spring-data-mongodb-benchmarks/src/main/java/org/springframework/data/mongodb/core/convert/DbRefMappingBenchmark.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

spring-data-mongodb-benchmarks/src/main/java/org/springframework/data/mongodb/core/convert/MappingMongoConverterBenchmark.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

spring-data-mongodb-benchmarks/src/main/java/org/springframework/data/mongodb/microbenchmark/AbstractMicrobenchmark.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

spring-data-mongodb-benchmarks/src/main/java/org/springframework/data/mongodb/microbenchmark/HttpResultsWriter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

spring-data-mongodb-benchmarks/src/main/java/org/springframework/data/mongodb/microbenchmark/MongoResultsWriter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

spring-data-mongodb-benchmarks/src/main/java/org/springframework/data/mongodb/microbenchmark/ResultsWriter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

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.RELEASE</version>
+		<version>2.1.0.M2</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 
@@ -17,6 +17,7 @@
 		<jpa>2.1.1</jpa>
 		<hibernate>5.2.1.Final</hibernate>
 		<java-module-name>spring.data.mongodb.cross.store</java-module-name>
+		<project.root>${basedir}/..</project.root>
 	</properties>
 
 	<dependencies>
@@ -49,7 +50,7 @@
 		<dependency>
 			<groupId>org.springframework.data</groupId>
 			<artifactId>spring-data-mongodb</artifactId>
-			<version>2.0.0.RELEASE</version>
+			<version>2.1.0.M2</version>
 		</dependency>
 
 		<!-- reactive -->

spring-data-mongodb-cross-store/src/main/java/org/springframework/data/mongodb/crossstore/DocumentBacked.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 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.

spring-data-mongodb-cross-store/src/main/java/org/springframework/data/mongodb/crossstore/MongoChangeSetPersister.java

@@ -1,214 +1,214 @@
-/*
- * Copyright 2011-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.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.springframework.data.mongodb.crossstore;
-
-import javax.persistence.EntityManagerFactory;
-
-import org.bson.Document;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.dao.DataAccessException;
-import org.springframework.dao.DataAccessResourceFailureException;
-import org.springframework.dao.DataIntegrityViolationException;
-import org.springframework.data.crossstore.ChangeSet;
-import org.springframework.data.crossstore.ChangeSetBacked;
-import org.springframework.data.crossstore.ChangeSetPersister;
-import org.springframework.data.mongodb.core.CollectionCallback;
-import org.springframework.data.mongodb.core.MongoTemplate;
-import org.springframework.util.ClassUtils;
-
-import com.mongodb.MongoException;
-import com.mongodb.client.MongoCollection;
-import com.mongodb.client.model.Filters;
-import com.mongodb.client.result.DeleteResult;
-
-/**
- * @author Thomas Risberg
- * @author Oliver Gierke
- * @author Alex Vengrovsk
- * @author Mark Paluch
- * @deprecated will be removed without replacement.
- */
-@Deprecated
-public class MongoChangeSetPersister implements ChangeSetPersister<Object> {
-
-	private static final String ENTITY_CLASS = "_entity_class";
-	private static final String ENTITY_ID = "_entity_id";
-	private static final String ENTITY_FIELD_NAME = "_entity_field_name";
-	private static final String ENTITY_FIELD_CLASS = "_entity_field_class";
-
-	private final Logger log = LoggerFactory.getLogger(getClass());
-
-	private MongoTemplate mongoTemplate;
-	private EntityManagerFactory entityManagerFactory;
-
-	public void setMongoTemplate(MongoTemplate mongoTemplate) {
-		this.mongoTemplate = mongoTemplate;
-	}
-
-	public void setEntityManagerFactory(EntityManagerFactory entityManagerFactory) {
-		this.entityManagerFactory = entityManagerFactory;
-	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.crossstore.ChangeSetPersister#getPersistentState(java.lang.Class, java.lang.Object, org.springframework.data.crossstore.ChangeSet)
-	 */
-	public void getPersistentState(Class<? extends ChangeSetBacked> entityClass, Object id, final ChangeSet changeSet)
-			throws DataAccessException, NotFoundException {
-
-		if (id == null) {
-			log.debug("Unable to load MongoDB data for null id");
-			return;
-		}
-
-		String collName = getCollectionNameForEntity(entityClass);
-
-		final Document dbk = new Document();
-		dbk.put(ENTITY_ID, id);
-		dbk.put(ENTITY_CLASS, entityClass.getName());
-		if (log.isDebugEnabled()) {
-			log.debug("Loading MongoDB data for {}", dbk);
-		}
-		mongoTemplate.execute(collName, new CollectionCallback<Object>() {
-			public Object doInCollection(MongoCollection<Document> collection) throws MongoException, DataAccessException {
-				for (Document dbo : collection.find(dbk)) {
-					String key = (String) dbo.get(ENTITY_FIELD_NAME);
-					if (log.isDebugEnabled()) {
-						log.debug("Processing key: {}", key);
-					}
-					if (!changeSet.getValues().containsKey(key)) {
-						String className = (String) dbo.get(ENTITY_FIELD_CLASS);
-						if (className == null) {
-							throw new DataIntegrityViolationException(
-									"Unble to convert property " + key + ": Invalid metadata, " + ENTITY_FIELD_CLASS + " not available");
-						}
-						Class<?> clazz = ClassUtils.resolveClassName(className, ClassUtils.getDefaultClassLoader());
-						Object value = mongoTemplate.getConverter().read(clazz, dbo);
-						if (log.isDebugEnabled()) {
-							log.debug("Adding to ChangeSet: {}", key);
-						}
-						changeSet.set(key, value);
-					}
-				}
-				return null;
-			}
-		});
-	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.crossstore.ChangeSetPersister#getPersistentId(org.springframework.data.crossstore.ChangeSetBacked, org.springframework.data.crossstore.ChangeSet)
-	 */
-	public Object getPersistentId(ChangeSetBacked entity, ChangeSet cs) throws DataAccessException {
-		if (log.isDebugEnabled()) {
-			log.debug("getPersistentId called on {}", entity);
-		}
-		if (entityManagerFactory == null) {
-			throw new DataAccessResourceFailureException("EntityManagerFactory cannot be null");
-		}
-
-		return entityManagerFactory.getPersistenceUnitUtil().getIdentifier(entity);
-	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.crossstore.ChangeSetPersister#persistState(org.springframework.data.crossstore.ChangeSetBacked, org.springframework.data.crossstore.ChangeSet)
-	 */
-	public Object persistState(ChangeSetBacked entity, ChangeSet cs) throws DataAccessException {
-		if (cs == null) {
-			log.debug("Flush: changeset was null, nothing to flush.");
-			return 0L;
-		}
-
-		if (log.isDebugEnabled()) {
-			log.debug("Flush: changeset: {}", cs.getValues());
-		}
-
-		String collName = getCollectionNameForEntity(entity.getClass());
-		if (mongoTemplate.getCollection(collName) == null) {
-			mongoTemplate.createCollection(collName);
-		}
-
-		for (String key : cs.getValues().keySet()) {
-			if (key != null && !key.startsWith("_") && !key.equals(ChangeSetPersister.ID_KEY)) {
-				Object value = cs.getValues().get(key);
-				final Document dbQuery = new Document();
-				dbQuery.put(ENTITY_ID, getPersistentId(entity, cs));
-				dbQuery.put(ENTITY_CLASS, entity.getClass().getName());
-				dbQuery.put(ENTITY_FIELD_NAME, key);
-				final Document dbId = mongoTemplate.execute(collName, new CollectionCallback<Document>() {
-					public Document doInCollection(MongoCollection<Document> collection)
-							throws MongoException, DataAccessException {
-						Document id = collection.find(dbQuery).first();
-						return id;
-					}
-				});
-
-				if (value == null) {
-					if (log.isDebugEnabled()) {
-						log.debug("Flush: removing: {}", dbQuery);
-					}
-					mongoTemplate.execute(collName, new CollectionCallback<Object>() {
-						public Object doInCollection(MongoCollection<Document> collection)
-								throws MongoException, DataAccessException {
-							DeleteResult dr = collection.deleteMany(dbQuery);
-							return null;
-						}
-					});
-				} else {
-					final Document dbDoc = new Document();
-					dbDoc.putAll(dbQuery);
-					if (log.isDebugEnabled()) {
-						log.debug("Flush: saving: {}", dbQuery);
-					}
-					mongoTemplate.getConverter().write(value, dbDoc);
-					dbDoc.put(ENTITY_FIELD_CLASS, value.getClass().getName());
-					if (dbId != null) {
-						dbDoc.put("_id", dbId.get("_id"));
-					}
-					mongoTemplate.execute(collName, new CollectionCallback<Object>() {
-						public Object doInCollection(MongoCollection<Document> collection)
-								throws MongoException, DataAccessException {
-
-							if (dbId != null) {
-								collection.replaceOne(Filters.eq("_id", dbId.get("_id")), dbDoc);
-							} else {
-
-								if (dbDoc.containsKey("_id") && dbDoc.get("_id") == null) {
-									dbDoc.remove("_id");
-								}
-								collection.insertOne(dbDoc);
-							}
-							return null;
-						}
-					});
-				}
-			}
-		}
-		return 0L;
-	}
-
-	/**
-	 * Returns the collection the given entity type shall be persisted to.
-	 * 
-	 * @param entityClass must not be {@literal null}.
-	 * @return
-	 */
-	private String getCollectionNameForEntity(Class<? extends ChangeSetBacked> entityClass) {
-		return mongoTemplate.getCollectionName(entityClass);
-	}
-}
+/*
+ * Copyright 2011-2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.mongodb.crossstore;
+
+import javax.persistence.EntityManagerFactory;
+
+import org.bson.Document;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.dao.DataAccessException;
+import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.data.crossstore.ChangeSet;
+import org.springframework.data.crossstore.ChangeSetBacked;
+import org.springframework.data.crossstore.ChangeSetPersister;
+import org.springframework.data.mongodb.core.CollectionCallback;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.util.ClassUtils;
+
+import com.mongodb.MongoException;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.model.Filters;
+import com.mongodb.client.result.DeleteResult;
+
+/**
+ * @author Thomas Risberg
+ * @author Oliver Gierke
+ * @author Alex Vengrovsk
+ * @author Mark Paluch
+ * @deprecated will be removed without replacement.
+ */
+@Deprecated
+public class MongoChangeSetPersister implements ChangeSetPersister<Object> {
+
+	private static final String ENTITY_CLASS = "_entity_class";
+	private static final String ENTITY_ID = "_entity_id";
+	private static final String ENTITY_FIELD_NAME = "_entity_field_name";
+	private static final String ENTITY_FIELD_CLASS = "_entity_field_class";
+
+	private final Logger log = LoggerFactory.getLogger(getClass());
+
+	private MongoTemplate mongoTemplate;
+	private EntityManagerFactory entityManagerFactory;
+
+	public void setMongoTemplate(MongoTemplate mongoTemplate) {
+		this.mongoTemplate = mongoTemplate;
+	}
+
+	public void setEntityManagerFactory(EntityManagerFactory entityManagerFactory) {
+		this.entityManagerFactory = entityManagerFactory;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.crossstore.ChangeSetPersister#getPersistentState(java.lang.Class, java.lang.Object, org.springframework.data.crossstore.ChangeSet)
+	 */
+	public void getPersistentState(Class<? extends ChangeSetBacked> entityClass, Object id, final ChangeSet changeSet)
+			throws DataAccessException, NotFoundException {
+
+		if (id == null) {
+			log.debug("Unable to load MongoDB data for null id");
+			return;
+		}
+
+		String collName = getCollectionNameForEntity(entityClass);
+
+		final Document dbk = new Document();
+		dbk.put(ENTITY_ID, id);
+		dbk.put(ENTITY_CLASS, entityClass.getName());
+		if (log.isDebugEnabled()) {
+			log.debug("Loading MongoDB data for {}", dbk);
+		}
+		mongoTemplate.execute(collName, new CollectionCallback<Object>() {
+			public Object doInCollection(MongoCollection<Document> collection) throws MongoException, DataAccessException {
+				for (Document dbo : collection.find(dbk)) {
+					String key = (String) dbo.get(ENTITY_FIELD_NAME);
+					if (log.isDebugEnabled()) {
+						log.debug("Processing key: {}", key);
+					}
+					if (!changeSet.getValues().containsKey(key)) {
+						String className = (String) dbo.get(ENTITY_FIELD_CLASS);
+						if (className == null) {
+							throw new DataIntegrityViolationException(
+									"Unble to convert property " + key + ": Invalid metadata, " + ENTITY_FIELD_CLASS + " not available");
+						}
+						Class<?> clazz = ClassUtils.resolveClassName(className, ClassUtils.getDefaultClassLoader());
+						Object value = mongoTemplate.getConverter().read(clazz, dbo);
+						if (log.isDebugEnabled()) {
+							log.debug("Adding to ChangeSet: {}", key);
+						}
+						changeSet.set(key, value);
+					}
+				}
+				return null;
+			}
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.crossstore.ChangeSetPersister#getPersistentId(org.springframework.data.crossstore.ChangeSetBacked, org.springframework.data.crossstore.ChangeSet)
+	 */
+	public Object getPersistentId(ChangeSetBacked entity, ChangeSet cs) throws DataAccessException {
+		if (log.isDebugEnabled()) {
+			log.debug("getPersistentId called on {}", entity);
+		}
+		if (entityManagerFactory == null) {
+			throw new DataAccessResourceFailureException("EntityManagerFactory cannot be null");
+		}
+
+		return entityManagerFactory.getPersistenceUnitUtil().getIdentifier(entity);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.crossstore.ChangeSetPersister#persistState(org.springframework.data.crossstore.ChangeSetBacked, org.springframework.data.crossstore.ChangeSet)
+	 */
+	public Object persistState(ChangeSetBacked entity, ChangeSet cs) throws DataAccessException {
+		if (cs == null) {
+			log.debug("Flush: changeset was null, nothing to flush.");
+			return 0L;
+		}
+
+		if (log.isDebugEnabled()) {
+			log.debug("Flush: changeset: {}", cs.getValues());
+		}
+
+		String collName = getCollectionNameForEntity(entity.getClass());
+		if (mongoTemplate.getCollection(collName) == null) {
+			mongoTemplate.createCollection(collName);
+		}
+
+		for (String key : cs.getValues().keySet()) {
+			if (key != null && !key.startsWith("_") && !key.equals(ChangeSetPersister.ID_KEY)) {
+				Object value = cs.getValues().get(key);
+				final Document dbQuery = new Document();
+				dbQuery.put(ENTITY_ID, getPersistentId(entity, cs));
+				dbQuery.put(ENTITY_CLASS, entity.getClass().getName());
+				dbQuery.put(ENTITY_FIELD_NAME, key);
+				final Document dbId = mongoTemplate.execute(collName, new CollectionCallback<Document>() {
+					public Document doInCollection(MongoCollection<Document> collection)
+							throws MongoException, DataAccessException {
+						Document id = collection.find(dbQuery).first();
+						return id;
+					}
+				});
+
+				if (value == null) {
+					if (log.isDebugEnabled()) {
+						log.debug("Flush: removing: {}", dbQuery);
+					}
+					mongoTemplate.execute(collName, new CollectionCallback<Object>() {
+						public Object doInCollection(MongoCollection<Document> collection)
+								throws MongoException, DataAccessException {
+							DeleteResult dr = collection.deleteMany(dbQuery);
+							return null;
+						}
+					});
+				} else {
+					final Document dbDoc = new Document();
+					dbDoc.putAll(dbQuery);
+					if (log.isDebugEnabled()) {
+						log.debug("Flush: saving: {}", dbQuery);
+					}
+					mongoTemplate.getConverter().write(value, dbDoc);
+					dbDoc.put(ENTITY_FIELD_CLASS, value.getClass().getName());
+					if (dbId != null) {
+						dbDoc.put("_id", dbId.get("_id"));
+					}
+					mongoTemplate.execute(collName, new CollectionCallback<Object>() {
+						public Object doInCollection(MongoCollection<Document> collection)
+								throws MongoException, DataAccessException {
+
+							if (dbId != null) {
+								collection.replaceOne(Filters.eq("_id", dbId.get("_id")), dbDoc);
+							} else {
+
+								if (dbDoc.containsKey("_id") && dbDoc.get("_id") == null) {
+									dbDoc.remove("_id");
+								}
+								collection.insertOne(dbDoc);
+							}
+							return null;
+						}
+					});
+				}
+			}
+		}
+		return 0L;
+	}
+
+	/**
+	 * Returns the collection the given entity type shall be persisted to.
+	 *
+	 * @param entityClass must not be {@literal null}.
+	 * @return
+	 */
+	private String getCollectionNameForEntity(Class<? extends ChangeSetBacked> entityClass) {
+		return mongoTemplate.getCollectionName(entityClass);
+	}
+}

spring-data-mongodb-cross-store/src/main/java/org/springframework/data/mongodb/crossstore/RelatedDocument.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 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.

spring-data-mongodb-cross-store/src/test/java/org/springframework/data/mongodb/crossstore/CrossStoreMongoTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2013 the original author or authors.
+ * Copyright 2011-2018 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.
@@ -39,7 +39,7 @@ import org.springframework.transaction.support.TransactionTemplate;
 
 /**
  * Integration tests for MongoDB cross-store persistence (mainly {@link MongoChangeSetPersister}).
- * 
+ *
  * @author Thomas Risberg
  * @author Oliver Gierke
  */

spring-data-mongodb-cross-store/src/test/java/org/springframework/data/mongodb/crossstore/test/Address.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011 the original author or authors.
+ * Copyright 2011-2018 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.

spring-data-mongodb-cross-store/src/test/java/org/springframework/data/mongodb/crossstore/test/Person.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011 the original author or authors.
+ * Copyright 2011-2018 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.

spring-data-mongodb-cross-store/src/test/java/org/springframework/data/mongodb/crossstore/test/Resume.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011 the original author or authors.
+ * Copyright 2011-2018 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.

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.RELEASE</version>
+		<version>2.1.0.M2</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.RELEASE</version>
+		<version>2.1.0.M2</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 
@@ -19,6 +19,8 @@
 		<objenesis>1.3</objenesis>
 		<equalsverifier>1.7.8</equalsverifier>
 		<java-module-name>spring.data.mongodb</java-module-name>
+		<project.root>${basedir}/..</project.root>
+		<multithreadedtc>1.01</multithreadedtc>
 	</properties>
 
 	<dependencies>
@@ -145,7 +147,7 @@
 			<version>1.0.1</version>
 			<scope>test</scope>
 		</dependency>
-		
+
 		<dependency>
 			<groupId>javax.interceptor</groupId>
 			<artifactId>javax.interceptor-api</artifactId>
@@ -214,7 +216,6 @@
 		<dependency>
 			<groupId>com.fasterxml.jackson.core</groupId>
 			<artifactId>jackson-databind</artifactId>
-			<version>${jackson}</version>
 			<optional>true</optional>
 		</dependency>
 
@@ -245,6 +246,13 @@
 			<scope>test</scope>
 		</dependency>
 
+		<dependency>
+			<groupId>edu.umd.cs.mtc</groupId>
+			<artifactId>multithreadedtc</artifactId>
+			<version>${multithreadedtc}</version>
+			<scope>test</scope>
+		</dependency>
+
 		<!-- Kotlin extension -->
 		<dependency>
 			<groupId>org.jetbrains.kotlin</groupId>
@@ -288,74 +296,9 @@
 	</dependencies>
 
 	<build>
+
 		<plugins>
 
-			<plugin>
-				<artifactId>kotlin-maven-plugin</artifactId>
-				<groupId>org.jetbrains.kotlin</groupId>
-				<version>${kotlin}</version>
-				<configuration>
-					<jvmTarget>${source.level}</jvmTarget>
-				</configuration>
-				<executions>
-					<execution>
-						<id>compile</id>
-						<phase>compile</phase>
-						<goals>
-							<goal>compile</goal>
-						</goals>
-						<configuration>
-							<sourceDirs>
-								<sourceDir>${project.basedir}/src/main/kotlin</sourceDir>
-								<sourceDir>${project.basedir}/src/main/java</sourceDir>
-							</sourceDirs>
-						</configuration>
-					</execution>
-					<execution>
-						<id>test-compile</id>
-						<phase>test-compile</phase>
-						<goals>
-							<goal>test-compile</goal>
-						</goals>
-						<configuration>
-							<sourceDirs>
-								<sourceDir>${project.basedir}/src/test/kotlin</sourceDir>
-								<sourceDir>${project.basedir}/src/test/java</sourceDir>
-							</sourceDirs>
-						</configuration>
-					</execution>
-				</executions>
-			</plugin>
-
-			<plugin>
-				<groupId>org.apache.maven.plugins</groupId>
-				<artifactId>maven-compiler-plugin</artifactId>
-				<executions>
-					<execution>
-						<id>default-compile</id>
-						<phase>none</phase>
-					</execution>
-					<execution>
-						<id>default-testCompile</id>
-						<phase>none</phase>
-					</execution>
-					<execution>
-						<id>java-compile</id>
-						<phase>compile</phase>
-						<goals>
-							<goal>compile</goal>
-						</goals>
-					</execution>
-					<execution>
-						<id>java-test-compile</id>
-						<phase>test-compile</phase>
-						<goals>
-							<goal>testCompile</goal>
-						</goals>
-					</execution>
-				</executions>
-			</plugin>
-
 			<plugin>
 				<groupId>com.mysema.maven</groupId>
 				<artifactId>apt-maven-plugin</artifactId>
@@ -384,7 +327,6 @@
 			<plugin>
 				<groupId>org.apache.maven.plugins</groupId>
 				<artifactId>maven-surefire-plugin</artifactId>
-				<version>2.12</version>
 				<configuration>
 					<useFile>false</useFile>
 					<includes>
@@ -406,6 +348,8 @@
 					</properties>
 				</configuration>
 			</plugin>
+
 		</plugins>
+
 	</build>
 </project>

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/BulkOperationException.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2018 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,7 +25,7 @@ import com.mongodb.BulkWriteResult;
 
 /**
  * Is thrown when errors occur during bulk operations.
- * 
+ *
  * @author Tobias Trelle
  * @author Oliver Gierke
  * @since 1.9
@@ -39,7 +39,7 @@ public class BulkOperationException extends DataAccessException {
 
 	/**
 	 * Creates a new {@link BulkOperationException} with the given message and source {@link BulkWriteException}.
-	 * 
+	 *
 	 * @param message must not be {@literal null}.
 	 * @param source must not be {@literal null}.
 	 */

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/CannotGetMongoDbConnectionException.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2017 the original author or authors.
+ * Copyright 2010-2018 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.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ClientSessionException.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.mongodb;
+
+import org.springframework.dao.NonTransientDataAccessException;
+import org.springframework.lang.Nullable;
+
+/**
+ * {@link NonTransientDataAccessException} specific to MongoDB {@link com.mongodb.session.ClientSession} related data
+ * access failures such as reading data using an already closed session.
+ *
+ * @author Christoph Strobl
+ * @since 2.1
+ */
+public class ClientSessionException extends NonTransientDataAccessException {
+
+	/**
+	 * Constructor for {@link ClientSessionException}.
+	 *
+	 * @param msg the detail message. Must not be {@literal null}.
+	 */
+	public ClientSessionException(String msg) {
+		super(msg);
+	}
+
+	/**
+	 * Constructor for {@link ClientSessionException}.
+	 *
+	 * @param msg the detail message. Can be {@literal null}.
+	 * @param cause the root cause. Can be {@literal null}.
+	 */
+	public ClientSessionException(@Nullable String msg, @Nullable Throwable cause) {
+		super(msg, cause);
+	}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/CodecRegistryProvider.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2017-2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.mongodb;
+
+import java.util.Optional;
+
+import org.bson.codecs.Codec;
+import org.bson.codecs.configuration.CodecConfigurationException;
+import org.bson.codecs.configuration.CodecRegistry;
+import org.springframework.util.Assert;
+
+/**
+ * Provider interface to obtain {@link CodecRegistry} from the underlying MongoDB Java driver.
+ *
+ * @author Christoph Strobl
+ * @author Mark Paluch
+ * @since 2.1
+ */
+@FunctionalInterface
+public interface CodecRegistryProvider {
+
+	/**
+	 * Get the underlying {@link CodecRegistry} used by the MongoDB Java driver.
+	 *
+	 * @return never {@literal null}.
+	 * @throws IllegalStateException if {@link CodecRegistry} cannot be obtained.
+	 */
+	CodecRegistry getCodecRegistry();
+
+	/**
+	 * Checks if a {@link Codec} is registered for a given type.
+	 *
+	 * @param type must not be {@literal null}.
+	 * @return true if {@link #getCodecRegistry()} holds a {@link Codec} for given type.
+	 * @throws IllegalStateException if {@link CodecRegistry} cannot be obtained.
+	 */
+	default boolean hasCodecFor(Class<?> type) {
+		return getCodecFor(type).isPresent();
+	}
+
+	/**
+	 * Get the {@link Codec} registered for the given {@literal type} or an {@link Optional#empty() empty Optional}
+	 * instead.
+	 *
+	 * @param type must not be {@literal null}.
+	 * @param <T>
+	 * @return never {@literal null}.
+	 * @throws IllegalArgumentException if {@literal type} is {@literal null}.
+	 */
+	default <T> Optional<Codec<T>> getCodecFor(Class<T> type) {
+
+		Assert.notNull(type, "Type must not be null!");
+
+		try {
+			return Optional.of(getCodecRegistry().get(type));
+		} catch (CodecConfigurationException e) {
+			// ignore
+		}
+		return Optional.empty();
+	}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/InvalidMongoDbApiUsageException.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2011 the original author or authors.
+ * Copyright 2010-2018 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.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/LazyLoadingException.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013 the original author or authors.
+ * Copyright 2013-2018 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.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/MongoCollectionUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011 the original author or authors.
+ * Copyright 2011-2018 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.
@@ -23,7 +23,7 @@ import org.springframework.util.StringUtils;
  * <p/>
  * <p/>
  * Mainly intended for internal use within the framework.
- * 
+ *
  * @author Thomas Risberg
  * @since 1.0
  */
@@ -38,7 +38,7 @@ public abstract class MongoCollectionUtils {
 
 	/**
 	 * Obtains the collection name to use for the provided class
-	 * 
+	 *
 	 * @param entityClass The class to determine the preferred collection name for
 	 * @return The preferred collection name
 	 */

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/MongoDbFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2013 the original author or authors.
+ * Copyright 2011-2018 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,24 +15,28 @@
  */
 package org.springframework.data.mongodb;
 
+import org.bson.codecs.configuration.CodecRegistry;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
 import org.springframework.data.mongodb.core.MongoExceptionTranslator;
 
+import com.mongodb.ClientSessionOptions;
 import com.mongodb.DB;
 import com.mongodb.client.MongoDatabase;
+import com.mongodb.session.ClientSession;
 
 /**
- * Interface for factories creating {@link DB} instances.
- * 
+ * Interface for factories creating {@link MongoDatabase} instances.
+ *
  * @author Mark Pollack
  * @author Thomas Darimont
+ * @author Christoph Strobl
  */
-public interface MongoDbFactory {
+public interface MongoDbFactory extends CodecRegistryProvider {
 
 	/**
-	 * Creates a default {@link DB} instance.
-	 * 
+	 * Creates a default {@link MongoDatabase} instance.
+	 *
 	 * @return
 	 * @throws DataAccessException
 	 */
@@ -40,7 +44,7 @@ public interface MongoDbFactory {
 
 	/**
 	 * Creates a {@link DB} instance to access the database with the given name.
-	 * 
+	 *
 	 * @param dbName must not be {@literal null} or empty.
 	 * @return
 	 * @throws DataAccessException
@@ -49,10 +53,51 @@ public interface MongoDbFactory {
 
 	/**
 	 * Exposes a shared {@link MongoExceptionTranslator}.
-	 * 
+	 *
 	 * @return will never be {@literal null}.
 	 */
 	PersistenceExceptionTranslator getExceptionTranslator();
 
 	DB getLegacyDb();
+
+	/**
+	 * Get the underlying {@link CodecRegistry} used by the MongoDB Java driver.
+	 *
+	 * @return never {@literal null}.
+	 */
+	@Override
+	default CodecRegistry getCodecRegistry() {
+		return getDb().getCodecRegistry();
+	}
+
+	/**
+	 * Obtain a {@link ClientSession} for given ClientSessionOptions.
+	 *
+	 * @param options must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	ClientSession getSession(ClientSessionOptions options);
+
+	/**
+	 * Obtain a {@link ClientSession} bound instance of {@link MongoDbFactory} returning {@link MongoDatabase} instances
+	 * that are aware and bound to a new session with given {@link ClientSessionOptions options}.
+	 *
+	 * @param options must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	default MongoDbFactory withSession(ClientSessionOptions options) {
+		return withSession(getSession(options));
+	}
+
+	/**
+	 * Obtain a {@link ClientSession} bound instance of {@link MongoDbFactory} returning {@link MongoDatabase} instances
+	 * that are aware and bound to the given session.
+	 *
+	 * @param session must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	MongoDbFactory withSession(ClientSession session);
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -16,19 +16,25 @@
 
 package org.springframework.data.mongodb;
 
+import reactor.core.publisher.Mono;
+
+import org.bson.codecs.configuration.CodecRegistry;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
 import org.springframework.data.mongodb.core.MongoExceptionTranslator;
 
+import com.mongodb.ClientSessionOptions;
 import com.mongodb.reactivestreams.client.MongoDatabase;
+import com.mongodb.session.ClientSession;
 
 /**
  * Interface for factories creating reactive {@link MongoDatabase} instances.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.0
  */
-public interface ReactiveMongoDatabaseFactory {
+public interface ReactiveMongoDatabaseFactory extends CodecRegistryProvider {
 
 	/**
 	 * Creates a default {@link MongoDatabase} instance.
@@ -53,4 +59,33 @@ public interface ReactiveMongoDatabaseFactory {
 	 * @return will never be {@literal null}.
 	 */
 	PersistenceExceptionTranslator getExceptionTranslator();
+
+	/**
+	 * Get the underlying {@link CodecRegistry} used by the reactive MongoDB Java driver.
+	 *
+	 * @return never {@literal null}.
+	 */
+	@Override
+	default CodecRegistry getCodecRegistry() {
+		return getMongoDatabase().getCodecRegistry();
+	}
+
+	/**
+	 * Obtain a {@link Mono} emitting a {@link ClientSession} for given {@link ClientSessionOptions options}.
+	 *
+	 * @param options must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	Mono<ClientSession> getSession(ClientSessionOptions options);
+
+	/**
+	 * Obtain a {@link ClientSession} bound instance of {@link ReactiveMongoDatabaseFactory} returning
+	 * {@link MongoDatabase} instances that are aware and bound to the given session.
+	 *
+	 * @param session must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	ReactiveMongoDatabaseFactory withSession(ClientSession session);
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/SessionAwareMethodInterceptor.java

@@ -0,0 +1,211 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.mongodb;
+
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.util.Optional;
+import java.util.function.BiFunction;
+
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+import org.springframework.core.MethodClassKey;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.ConcurrentReferenceHashMap;
+import org.springframework.util.ReflectionUtils;
+
+import com.mongodb.WriteConcern;
+import com.mongodb.session.ClientSession;
+
+/**
+ * {@link MethodInterceptor} implementation looking up and invoking an alternative target method having
+ * {@link ClientSession} as its first argument. This allows seamless integration with the existing code base.
+ * <p />
+ * The {@link MethodInterceptor} is aware of methods on {@code MongoCollection} that my return new instances of itself
+ * like (eg. {@link com.mongodb.reactivestreams.client.MongoCollection#withWriteConcern(WriteConcern)} and decorate them
+ * if not already proxied.
+ *
+ * @param <D> Type of the actual Mongo Database.
+ * @param <C> Type of the actual Mongo Collection.
+ * @author Christoph Strobl
+ * @author Mark Paluch
+ * @since 2.1
+ */
+public class SessionAwareMethodInterceptor<D, C> implements MethodInterceptor {
+
+	private static final MethodCache METHOD_CACHE = new MethodCache();
+
+	private final ClientSession session;
+	private final ClientSessionOperator collectionDecorator;
+	private final ClientSessionOperator databaseDecorator;
+	private final Object target;
+	private final Class<?> targetType;
+	private final Class<?> collectionType;
+	private final Class<?> databaseType;
+
+	/**
+	 * Create a new SessionAwareMethodInterceptor for given target.
+	 *
+	 * @param session the {@link ClientSession} to be used on invocation.
+	 * @param target the original target object.
+	 * @param databaseType the MongoDB database type
+	 * @param databaseDecorator a {@link ClientSessionOperator} used to create the proxy for an imperative / reactive
+	 *          {@code MongoDatabase}.
+	 * @param collectionType the MongoDB collection type.
+	 * @param collectionDecorator a {@link ClientSessionOperator} used to create the proxy for an imperative / reactive
+	 *          {@code MongoCollection}.
+	 * @param <T> target object type.
+	 */
+	public <T> SessionAwareMethodInterceptor(ClientSession session, T target, Class<D> databaseType,
+			ClientSessionOperator<D> databaseDecorator, Class<C> collectionType,
+			ClientSessionOperator<C> collectionDecorator) {
+
+		Assert.notNull(session, "ClientSession must not be null!");
+		Assert.notNull(target, "Target must not be null!");
+		Assert.notNull(databaseType, "Database type must not be null!");
+		Assert.notNull(databaseDecorator, "Database ClientSessionOperator must not be null!");
+		Assert.notNull(collectionType, "Collection type must not be null!");
+		Assert.notNull(collectionDecorator, "Collection ClientSessionOperator must not be null!");
+
+		this.session = session;
+		this.target = target;
+		this.databaseType = ClassUtils.getUserClass(databaseType);
+		this.collectionType = ClassUtils.getUserClass(collectionType);
+		this.collectionDecorator = collectionDecorator;
+		this.databaseDecorator = databaseDecorator;
+
+		this.targetType = ClassUtils.isAssignable(databaseType, target.getClass()) ? databaseType : collectionType;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.aopalliance.intercept.MethodInterceptor(org.aopalliance.intercept.MethodInvocation)
+	 */
+	@Nullable
+	@Override
+	public Object invoke(MethodInvocation methodInvocation) throws Throwable {
+
+		if (requiresDecoration(methodInvocation.getMethod())) {
+
+			Object target = methodInvocation.proceed();
+			if (target instanceof Proxy) {
+				return target;
+			}
+
+			return decorate(target);
+		}
+
+		if (!requiresSession(methodInvocation.getMethod())) {
+			return methodInvocation.proceed();
+		}
+
+		Optional<Method> targetMethod = METHOD_CACHE.lookup(methodInvocation.getMethod(), targetType);
+
+		return !targetMethod.isPresent() ? methodInvocation.proceed()
+				: ReflectionUtils.invokeMethod(targetMethod.get(), target,
+						prependSessionToArguments(session, methodInvocation));
+	}
+
+	private boolean requiresDecoration(Method method) {
+
+		return ClassUtils.isAssignable(databaseType, method.getReturnType())
+				|| ClassUtils.isAssignable(collectionType, method.getReturnType());
+	}
+
+	@SuppressWarnings("unchecked")
+	protected Object decorate(Object target) {
+
+		return ClassUtils.isAssignable(databaseType, target.getClass()) ? databaseDecorator.apply(session, target)
+				: collectionDecorator.apply(session, target);
+	}
+
+	private static boolean requiresSession(Method method) {
+
+		if (method.getParameterCount() == 0
+				|| !ClassUtils.isAssignable(ClientSession.class, method.getParameterTypes()[0])) {
+			return true;
+		}
+
+		return false;
+	}
+
+	private static Object[] prependSessionToArguments(ClientSession session, MethodInvocation invocation) {
+
+		Object[] args = new Object[invocation.getArguments().length + 1];
+
+		args[0] = session;
+		System.arraycopy(invocation.getArguments(), 0, args, 1, invocation.getArguments().length);
+
+		return args;
+	}
+
+	/**
+	 * Simple {@link Method} to {@link Method} caching facility for {@link ClientSession} overloaded targets.
+	 *
+	 * @since 2.1
+	 * @author Christoph Strobl
+	 */
+	static class MethodCache {
+
+		private final ConcurrentReferenceHashMap<MethodClassKey, Optional<Method>> cache = new ConcurrentReferenceHashMap<>();
+
+		/**
+		 * Lookup the target {@link Method}.
+		 *
+		 * @param method
+		 * @param targetClass
+		 * @return
+		 */
+		Optional<Method> lookup(Method method, Class<?> targetClass) {
+
+			return cache.computeIfAbsent(new MethodClassKey(method, targetClass),
+					val -> Optional.ofNullable(findTargetWithSession(method, targetClass)));
+		}
+
+		@Nullable
+		private Method findTargetWithSession(Method sourceMethod, Class<?> targetType) {
+
+			Class<?>[] argTypes = sourceMethod.getParameterTypes();
+			Class<?>[] args = new Class<?>[argTypes.length + 1];
+			args[0] = ClientSession.class;
+			System.arraycopy(argTypes, 0, args, 1, argTypes.length);
+
+			return ReflectionUtils.findMethod(targetType, sourceMethod.getName(), args);
+		}
+
+		/**
+		 * Check whether the cache contains an entry for {@link Method} and {@link Class}.
+		 *
+		 * @param method
+		 * @param targetClass
+		 * @return
+		 */
+		boolean contains(Method method, Class<?> targetClass) {
+			return cache.containsKey(new MethodClassKey(method, targetClass));
+		}
+	}
+
+	/**
+	 * Represents an operation upon two operands of the same type, producing a result of the same type as the operands
+	 * accepting {@link ClientSession}. This is a specialization of {@link BiFunction} for the case where the operands and
+	 * the result are all of the same type.
+	 *
+	 * @param <T> the type of the operands and result of the operator
+	 */
+	public interface ClientSessionOperator<T> extends BiFunction<ClientSession, T, T> {}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/UncategorizedMongoDbException.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2011 the original author or authors.
+ * Copyright 2010-2018 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.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/AbstractMongoConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 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.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/AbstractReactiveMongoConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/BeanNames.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 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.
@@ -17,7 +17,7 @@ package org.springframework.data.mongodb.config;
 
 /**
  * Constants to declare bean names used by the namespace configuration.
- * 
+ *
  * @author Jon Brisbin
  * @author Oliver Gierke
  * @author Martin Baumgartner

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/EnableMongoAuditing.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013 the original author or authors.
+ * Copyright 2013-2018 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.
@@ -28,7 +28,7 @@ import org.springframework.data.domain.AuditorAware;
 
 /**
  * Annotation to enable auditing in MongoDB via annotation configuration.
- * 
+ *
  * @author Thomas Darimont
  * @author Oliver Gierke
  */
@@ -41,21 +41,21 @@ public @interface EnableMongoAuditing {
 
 	/**
 	 * Configures the {@link AuditorAware} bean to be used to lookup the current principal.
-	 * 
+	 *
 	 * @return
 	 */
 	String auditorAwareRef() default "";
 
 	/**
 	 * Configures whether the creation and modification dates are set. Defaults to {@literal true}.
-	 * 
+	 *
 	 * @return
 	 */
 	boolean setDates() default true;
 
 	/**
 	 * Configures whether the entity shall be marked as modified on creation. Defaults to {@literal true}.
-	 * 
+	 *
 	 * @return
 	 */
 	boolean modifyOnCreate() default true;
@@ -63,7 +63,7 @@ public @interface EnableMongoAuditing {
 	/**
 	 * Configures a {@link DateTimeProvider} bean name that allows customizing the {@link org.joda.time.DateTime} to be
 	 * used for setting creation and modification dates.
-	 * 
+	 *
 	 * @return
 	 */
 	String dateTimeProviderRef() default "";

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/GeoJsonConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -21,7 +21,7 @@ import org.springframework.data.web.config.SpringDataJacksonModules;
 
 /**
  * Configuration class to expose {@link GeoJsonModule} as a Spring bean.
- * 
+ *
  * @author Oliver Gierke
  * @author Jens Schauder
  */

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/GridFsTemplateParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2014 the original author or authors.
+ * Copyright 2013-2018 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.
@@ -29,7 +29,7 @@ import org.w3c.dom.Element;
 
 /**
  * {@link BeanDefinitionParser} to parse {@code gridFsTemplate} elements into {@link BeanDefinition}s.
- * 
+ *
  * @author Martin Baumgartner
  */
 class GridFsTemplateParser extends AbstractBeanDefinitionParser {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MappingMongoConverterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 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.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoAuditingBeanDefinitionParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2014 the original author or authors.
+ * Copyright 2012-2018 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.
@@ -33,12 +33,12 @@ import org.w3c.dom.Element;
 /**
  * {@link BeanDefinitionParser} to register a {@link AuditingEventListener} to transparently set auditing information on
  * an entity.
- * 
+ *
  * @author Oliver Gierke
  */
 public class MongoAuditingBeanDefinitionParser extends AbstractSingleBeanDefinitionParser {
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser#getBeanClass(org.w3c.dom.Element)
 	 */
@@ -47,7 +47,7 @@ public class MongoAuditingBeanDefinitionParser extends AbstractSingleBeanDefinit
 		return AuditingEventListener.class;
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.beans.factory.xml.AbstractBeanDefinitionParser#shouldGenerateId()
 	 */
@@ -56,7 +56,7 @@ public class MongoAuditingBeanDefinitionParser extends AbstractSingleBeanDefinit
 		return true;
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser#doParse(org.w3c.dom.Element, org.springframework.beans.factory.xml.ParserContext, org.springframework.beans.factory.support.BeanDefinitionBuilder)
 	 */

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoAuditingRegistrar.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2016 the original author or authors.
+ * Copyright 2013-2018 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.
@@ -37,13 +37,13 @@ import org.springframework.util.Assert;
 
 /**
  * {@link ImportBeanDefinitionRegistrar} to enable {@link EnableMongoAuditing} annotation.
- * 
+ *
  * @author Thomas Darimont
  * @author Oliver Gierke
  */
 class MongoAuditingRegistrar extends AuditingBeanDefinitionRegistrarSupport {
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.auditing.config.AuditingBeanDefinitionRegistrarSupport#getAnnotation()
 	 */
@@ -52,7 +52,7 @@ class MongoAuditingRegistrar extends AuditingBeanDefinitionRegistrarSupport {
 		return EnableMongoAuditing.class;
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.auditing.config.AuditingBeanDefinitionRegistrarSupport#getAuditingHandlerBeanName()
 	 */
@@ -61,7 +61,7 @@ class MongoAuditingRegistrar extends AuditingBeanDefinitionRegistrarSupport {
 		return "mongoAuditingHandler";
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.auditing.config.AuditingBeanDefinitionRegistrarSupport#registerBeanDefinitions(org.springframework.core.type.AnnotationMetadata, org.springframework.beans.factory.support.BeanDefinitionRegistry)
 	 */
@@ -74,7 +74,7 @@ class MongoAuditingRegistrar extends AuditingBeanDefinitionRegistrarSupport {
 		super.registerBeanDefinitions(annotationMetadata, registry);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.auditing.config.AuditingBeanDefinitionRegistrarSupport#getAuditHandlerBeanDefinitionBuilder(org.springframework.data.auditing.config.AuditingConfiguration)
 	 */
@@ -92,7 +92,7 @@ class MongoAuditingRegistrar extends AuditingBeanDefinitionRegistrarSupport {
 		return configureDefaultAuditHandlerAttributes(configuration, builder);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.auditing.config.AuditingBeanDefinitionRegistrarSupport#registerAuditListener(org.springframework.beans.factory.config.BeanDefinition, org.springframework.beans.factory.support.BeanDefinitionRegistry)
 	 */
@@ -125,14 +125,14 @@ class MongoAuditingRegistrar extends AuditingBeanDefinitionRegistrarSupport {
 
 		/**
 		 * Creates a new {@link MongoMappingContextLookup} for the given {@link MappingMongoConverter}.
-		 * 
+		 *
 		 * @param converter must not be {@literal null}.
 		 */
 		public MongoMappingContextLookup(MappingMongoConverter converter) {
 			this.converter = converter;
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.beans.factory.FactoryBean#getObject()
 		 */
@@ -141,7 +141,7 @@ class MongoAuditingRegistrar extends AuditingBeanDefinitionRegistrarSupport {
 			return converter.getMappingContext();
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.beans.factory.FactoryBean#getObjectType()
 		 */
@@ -150,7 +150,7 @@ class MongoAuditingRegistrar extends AuditingBeanDefinitionRegistrarSupport {
 			return MappingContext.class;
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.beans.factory.FactoryBean#isSingleton()
 		 */

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoClientParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -29,13 +29,13 @@ import org.w3c.dom.Element;
 
 /**
  * Parser for {@code mongo-client} definitions.
- * 
+ *
  * @author Christoph Strobl
  * @since 1.7
  */
 public class MongoClientParser implements BeanDefinitionParser {
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.beans.factory.xml.BeanDefinitionParser#parse(org.w3c.dom.Element, org.springframework.beans.factory.xml.ParserContext)
 	 */

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoConfigurationSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoCredentialPropertyEditor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoDbFactoryParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 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.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoJmxParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 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.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoNamespaceHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 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.
@@ -19,7 +19,7 @@ import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
 
 /**
  * {@link org.springframework.beans.factory.xml.NamespaceHandler} for Mongo DB configuration.
- * 
+ *
  * @author Oliver Gierke
  * @author Martin Baumgartner
  * @author Christoph Strobl

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoParsingUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 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.
@@ -30,7 +30,7 @@ import org.w3c.dom.Element;
 
 /**
  * Utility methods for {@link BeanDefinitionParser} implementations for MongoDB.
- * 
+ *
  * @author Mark Pollack
  * @author Oliver Gierke
  * @author Thomas Darimont
@@ -43,7 +43,7 @@ abstract class MongoParsingUtils {
 
 	/**
 	 * Parses the mongo replica-set element.
-	 * 
+	 *
 	 * @param parserContext the parser context
 	 * @param element the mongo element
 	 * @param mongoBuilder the bean definition builder to populate
@@ -56,7 +56,7 @@ abstract class MongoParsingUtils {
 	/**
 	 * Parses the {@code mongo:client-options} sub-element. Populates the given attribute factory with the proper
 	 * attributes.
-	 * 
+	 *
 	 * @param element must not be {@literal null}.
 	 * @param mongoClientBuilder must not be {@literal null}.
 	 * @return
@@ -102,7 +102,7 @@ abstract class MongoParsingUtils {
 	/**
 	 * Returns the {@link BeanDefinitionBuilder} to build a {@link BeanDefinition} for a
 	 * {@link WriteConcernPropertyEditor}.
-	 * 
+	 *
 	 * @return
 	 */
 	static BeanDefinitionBuilder getWriteConcernPropertyEditorBuilder() {
@@ -135,7 +135,7 @@ abstract class MongoParsingUtils {
 	/**
 	 * Returns the {@link BeanDefinitionBuilder} to build a {@link BeanDefinition} for a
 	 * {@link ReadPreferencePropertyEditor}.
-	 * 
+	 *
 	 * @return
 	 * @since 1.7
 	 */
@@ -153,7 +153,7 @@ abstract class MongoParsingUtils {
 	/**
 	 * Returns the {@link BeanDefinitionBuilder} to build a {@link BeanDefinition} for a
 	 * {@link MongoCredentialPropertyEditor}.
-	 * 
+	 *
 	 * @return
 	 * @since 1.7
 	 */

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoTemplateParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2014 the original author or authors.
+ * Copyright 2011-2018 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.
@@ -33,13 +33,13 @@ import org.w3c.dom.Element;
 
 /**
  * {@link BeanDefinitionParser} to parse {@code template} elements into {@link BeanDefinition}s.
- * 
+ *
  * @author Martin Baumgartner
  * @author Oliver Gierke
  */
 class MongoTemplateParser extends AbstractBeanDefinitionParser {
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.beans.factory.xml.AbstractBeanDefinitionParser#resolveId(org.w3c.dom.Element, org.springframework.beans.factory.support.AbstractBeanDefinition, org.springframework.beans.factory.xml.ParserContext)
 	 */

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/ReadPreferencePropertyEditor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -23,7 +23,7 @@ import com.mongodb.ReadPreference;
 
 /**
  * Parse a {@link String} to a {@link ReadPreference}.
- * 
+ *
  * @author Christoph Strobl
  * @since 1.7
  */

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/ServerAddressPropertyEditor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 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.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/StringToWriteConcernConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012 the original author or authors.
+ * Copyright 2012-2018 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.
@@ -21,12 +21,12 @@ import com.mongodb.WriteConcern;
 
 /**
  * Converter to create {@link WriteConcern} instances from String representations.
- * 
+ *
  * @author Oliver Gierke
  */
 public class StringToWriteConcernConverter implements Converter<String, WriteConcern> {
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.core.convert.converter.Converter#convert(java.lang.Object)
 	 */

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/WriteConcernPropertyEditor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 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.
@@ -27,7 +27,7 @@ import com.mongodb.WriteConcern;
  * {@link WriteConcern#valueOf(String)}, use the well known {@link WriteConcern} value, otherwise pass the string as is
  * to the constructor of the write concern. There is no support for other constructor signatures when parsing from a
  * string value.
- * 
+ *
  * @author Mark Pollack
  * @author Christoph Strobl
  */

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.

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

@@ -0,0 +1,117 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.mongodb.core;
+
+import lombok.EqualsAndHashCode;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.bson.Document;
+import org.springframework.data.mongodb.core.convert.MongoConverter;
+import org.springframework.data.mongodb.core.messaging.Message;
+import org.springframework.lang.Nullable;
+import org.springframework.util.ClassUtils;
+
+import com.mongodb.client.model.changestream.ChangeStreamDocument;
+
+/**
+ * {@link Message} implementation specific to MongoDB <a href="https://docs.mongodb.com/manual/changeStreams/">Change
+ * Streams</a>.
+ *
+ * @author Christoph Strobl
+ * @author Mark Paluch
+ * @since 2.1
+ */
+@EqualsAndHashCode
+public class ChangeStreamEvent<T> {
+
+	private final @Nullable ChangeStreamDocument<Document> raw;
+
+	private final Class<T> targetType;
+	private final MongoConverter converter;
+	private final AtomicReference<T> converted = new AtomicReference<>();
+
+	/**
+	 * @param raw can be {@literal null}.
+	 * @param targetType must not be {@literal null}.
+	 * @param converter must not be {@literal null}.
+	 */
+	public ChangeStreamEvent(@Nullable ChangeStreamDocument<Document> raw, Class<T> targetType,
+			MongoConverter converter) {
+
+		this.raw = raw;
+		this.targetType = targetType;
+		this.converter = converter;
+	}
+
+	/**
+	 * Get the raw {@link ChangeStreamDocument} as emitted by the driver.
+	 *
+	 * @return can be {@literal null}.
+	 */
+	@Nullable
+	public ChangeStreamDocument<Document> getRaw() {
+		return raw;
+	}
+
+	/**
+	 * Get the potentially converted {@link ChangeStreamDocument#getFullDocument()}.
+	 *
+	 * @return {@literal null} when {@link #getRaw()} or {@link ChangeStreamDocument#getFullDocument()} is
+	 *         {@literal null}.
+	 */
+	@Nullable
+	public T getBody() {
+
+		if (raw == null) {
+			return null;
+		}
+
+		if (raw.getFullDocument() == null) {
+			return targetType.cast(raw.getFullDocument());
+		}
+
+		return getConverted();
+	}
+
+	private T getConverted() {
+
+		T result = converted.get();
+		if (result != null) {
+			return result;
+		}
+
+		if (ClassUtils.isAssignable(Document.class, raw.getFullDocument().getClass())) {
+
+			result = converter.read(targetType, raw.getFullDocument());
+			return converted.compareAndSet(null, result) ? result : converted.get();
+		}
+
+		if (converter.getConversionService().canConvert(raw.getFullDocument().getClass(), targetType)) {
+
+			result = converter.getConversionService().convert(raw.getFullDocument(), targetType);
+			return converted.compareAndSet(null, result) ? result : converted.get();
+		}
+
+		throw new IllegalArgumentException(String.format("No converter found capable of converting %s to %s",
+				raw.getFullDocument().getClass(), targetType));
+	}
+
+	@Override
+	public String toString() {
+		return "ChangeStreamEvent {" + "raw=" + raw + ", targetType=" + targetType + '}';
+	}
+}

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

@@ -0,0 +1,218 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.mongodb.core;
+
+import lombok.EqualsAndHashCode;
+
+import java.util.Arrays;
+import java.util.Optional;
+
+import org.bson.BsonValue;
+import org.bson.Document;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.query.Collation;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+import com.mongodb.client.model.changestream.ChangeStreamDocument;
+import com.mongodb.client.model.changestream.FullDocument;
+
+/**
+ * Options applicable to MongoDB <a href="https://docs.mongodb.com/manual/changeStreams/">Change Streams</a>. Intended
+ * to be used along with {@link org.springframework.data.mongodb.core.messaging.ChangeStreamRequest} in a sync world as
+ * well {@link ReactiveMongoOperations} if you prefer it that way.
+ *
+ * @author Christoph Strobl
+ * @author Mark Paluch
+ * @since 2.1
+ */
+@EqualsAndHashCode
+public class ChangeStreamOptions {
+
+	private @Nullable Object filter;
+	private @Nullable BsonValue resumeToken;
+	private @Nullable FullDocument fullDocumentLookup;
+	private @Nullable Collation collation;
+
+	protected ChangeStreamOptions() {}
+
+	/**
+	 * @return {@link Optional#empty()} if not set.
+	 */
+	public Optional<Object> getFilter() {
+		return Optional.ofNullable(filter);
+	}
+
+	/**
+	 * @return {@link Optional#empty()} if not set.
+	 */
+	public Optional<BsonValue> getResumeToken() {
+		return Optional.ofNullable(resumeToken);
+	}
+
+	/**
+	 * @return {@link Optional#empty()} if not set.
+	 */
+	public Optional<FullDocument> getFullDocumentLookup() {
+		return Optional.ofNullable(fullDocumentLookup);
+	}
+
+	/**
+	 * @return {@link Optional#empty()} if not set.
+	 */
+	public Optional<Collation> getCollation() {
+		return Optional.ofNullable(collation);
+	}
+
+	/**
+	 * @return empty {@link ChangeStreamOptions}.
+	 */
+	public static ChangeStreamOptions empty() {
+		return ChangeStreamOptions.builder().build();
+	}
+
+	/**
+	 * Obtain a shiny new {@link ChangeStreamOptionsBuilder} and start defining options in this fancy fluent way. Just
+	 * don't forget to call {@link ChangeStreamOptionsBuilder#build() build()} when your're done.
+	 *
+	 * @return new instance of {@link ChangeStreamOptionsBuilder}.
+	 */
+	public static ChangeStreamOptionsBuilder builder() {
+		return new ChangeStreamOptionsBuilder();
+	}
+
+	/**
+	 * Builder for creating {@link ChangeStreamOptions}.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	public static class ChangeStreamOptionsBuilder {
+
+		private @Nullable Object filter;
+		private @Nullable BsonValue resumeToken;
+		private @Nullable FullDocument fullDocumentLookup;
+		private @Nullable Collation collation;
+
+		private ChangeStreamOptionsBuilder() {}
+
+		/**
+		 * Set the collation to use.
+		 *
+		 * @param collation must not be {@literal null} nor {@literal empty}.
+		 * @return this.
+		 */
+		public ChangeStreamOptionsBuilder collation(Collation collation) {
+
+			Assert.notNull(collation, "Collation must not be null nor empty!");
+
+			this.collation = collation;
+			return this;
+		}
+
+		/**
+		 * Set the filter to apply.
+		 * <p/>
+		 * Fields on aggregation expression root level are prefixed to map to fields contained in
+		 * {@link ChangeStreamDocument#getFullDocument() fullDocument}. However {@literal operationType}, {@literal ns},
+		 * {@literal documentKey} and {@literal fullDocument} are reserved words that will be omitted, and therefore taken
+		 * as given, during the mapping procedure. You may want to have a look at the
+		 * <a href="https://docs.mongodb.com/manual/reference/change-events/">structure of Change Events</a>.
+		 * <p/>
+		 * Use {@link org.springframework.data.mongodb.core.aggregation.TypedAggregation} to ensure filter expressions are
+		 * mapped to domain type fields.
+		 *
+		 * @param filter the {@link Aggregation Aggregation pipeline} to apply for filtering events. Must not be
+		 *          {@literal null}.
+		 * @return this.
+		 */
+		public ChangeStreamOptionsBuilder filter(Aggregation filter) {
+
+			Assert.notNull(filter, "Filter must not be null!");
+
+			this.filter = filter;
+			return this;
+		}
+
+		/**
+		 * Set the plain filter chain to apply.
+		 *
+		 * @param filter must not be {@literal null} nor contain {@literal null} values.
+		 * @return this.
+		 */
+		public ChangeStreamOptionsBuilder filter(Document... filter) {
+
+			Assert.noNullElements(filter, "Filter must not contain null values");
+
+			this.filter = Arrays.asList(filter);
+			return this;
+		}
+
+		/**
+		 * Set the resume token (typically a {@link org.bson.BsonDocument} containing a {@link org.bson.BsonBinary binary
+		 * token}) after which to start with listening.
+		 *
+		 * @param resumeToken must not be {@literal null}.
+		 * @return this.
+		 */
+		public ChangeStreamOptionsBuilder resumeToken(BsonValue resumeToken) {
+
+			Assert.notNull(resumeToken, "ResumeToken must not be null!");
+
+			this.resumeToken = resumeToken;
+			return this;
+		}
+
+		/**
+		 * Set the {@link FullDocument} lookup to {@link FullDocument#UPDATE_LOOKUP}.
+		 *
+		 * @return this.
+		 * @see #fullDocumentLookup(FullDocument)
+		 */
+		public ChangeStreamOptionsBuilder returnFullDocumentOnUpdate() {
+			return fullDocumentLookup(FullDocument.UPDATE_LOOKUP);
+		}
+
+		/**
+		 * Set the {@link FullDocument} lookup to use.
+		 *
+		 * @param lookup must not be {@literal null}.
+		 * @return this.
+		 */
+		public ChangeStreamOptionsBuilder fullDocumentLookup(FullDocument lookup) {
+
+			Assert.notNull(lookup, "Lookup must not be null!");
+
+			this.fullDocumentLookup = lookup;
+			return this;
+		}
+
+		/**
+		 * @return the built {@link ChangeStreamOptions}
+		 */
+		public ChangeStreamOptions build() {
+
+			ChangeStreamOptions options = new ChangeStreamOptions();
+
+			options.filter = filter;
+			options.resumeToken = resumeToken;
+			options.fullDocumentLookup = fullDocumentLookup;
+			options.collation = collation;
+
+			return options;
+		}
+	}
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2017 the original author or authors.
+ * Copyright 2010-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2017 the original author or authors.
+ * Copyright 2010-2018 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,18 +15,27 @@
  */
 package org.springframework.data.mongodb.core;
 
+import lombok.RequiredArgsConstructor;
+
 import java.util.Optional;
 
 import org.springframework.data.mongodb.core.query.Collation;
+import org.springframework.data.mongodb.core.schema.MongoJsonSchema;
+import org.springframework.data.mongodb.core.validation.Validator;
+import org.springframework.data.util.Optionals;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
+import com.mongodb.client.model.ValidationAction;
+import com.mongodb.client.model.ValidationLevel;
+
 /**
  * Provides a simple wrapper to encapsulate the variety of settings you can use when creating a collection.
  *
  * @author Thomas Risberg
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author Andreas Zink
  */
 public class CollectionOptions {
 
@@ -34,6 +43,7 @@ public class CollectionOptions {
 	private @Nullable Long size;
 	private @Nullable Boolean capped;
 	private @Nullable Collation collation;
+	private ValidationOptions validationOptions;
 
 	/**
 	 * Constructs a new <code>CollectionOptions</code> instance.
@@ -46,16 +56,17 @@ public class CollectionOptions {
 	 */
 	@Deprecated
 	public CollectionOptions(@Nullable Long size, @Nullable Long maxDocuments, @Nullable Boolean capped) {
-		this(size, maxDocuments, capped, null);
+		this(size, maxDocuments, capped, null, ValidationOptions.none());
 	}
 
 	private CollectionOptions(@Nullable Long size, @Nullable Long maxDocuments, @Nullable Boolean capped,
-			@Nullable Collation collation) {
+			@Nullable Collation collation, ValidationOptions validationOptions) {
 
 		this.maxDocuments = maxDocuments;
 		this.size = size;
 		this.capped = capped;
 		this.collation = collation;
+		this.validationOptions = validationOptions;
 	}
 
 	/**
@@ -69,7 +80,7 @@ public class CollectionOptions {
 
 		Assert.notNull(collation, "Collation must not be null!");
 
-		return new CollectionOptions(null, null, null, collation);
+		return new CollectionOptions(null, null, null, collation, ValidationOptions.none());
 	}
 
 	/**
@@ -79,7 +90,7 @@ public class CollectionOptions {
 	 * @since 2.0
 	 */
 	public static CollectionOptions empty() {
-		return new CollectionOptions(null, null, null, null);
+		return new CollectionOptions(null, null, null, null, ValidationOptions.none());
 	}
 
 	/**
@@ -90,7 +101,7 @@ public class CollectionOptions {
 	 * @since 2.0
 	 */
 	public CollectionOptions capped() {
-		return new CollectionOptions(size, maxDocuments, true, collation);
+		return new CollectionOptions(size, maxDocuments, true, collation, validationOptions);
 	}
 
 	/**
@@ -101,7 +112,7 @@ public class CollectionOptions {
 	 * @since 2.0
 	 */
 	public CollectionOptions maxDocuments(long maxDocuments) {
-		return new CollectionOptions(size, maxDocuments, capped, collation);
+		return new CollectionOptions(size, maxDocuments, capped, collation, validationOptions);
 	}
 
 	/**
@@ -112,7 +123,7 @@ public class CollectionOptions {
 	 * @since 2.0
 	 */
 	public CollectionOptions size(long size) {
-		return new CollectionOptions(size, maxDocuments, capped, collation);
+		return new CollectionOptions(size, maxDocuments, capped, collation, validationOptions);
 	}
 
 	/**
@@ -123,7 +134,127 @@ public class CollectionOptions {
 	 * @since 2.0
 	 */
 	public CollectionOptions collation(@Nullable Collation collation) {
-		return new CollectionOptions(size, maxDocuments, capped, collation);
+		return new CollectionOptions(size, maxDocuments, capped, collation, validationOptions);
+	}
+
+	/**
+	 * Create new {@link CollectionOptions} with already given settings and {@code validationOptions} set to given
+	 * {@link MongoJsonSchema}.
+	 *
+	 * @param schema can be {@literal null}.
+	 * @return new {@link CollectionOptions}.
+	 * @since 2.1
+	 */
+	public CollectionOptions schema(@Nullable MongoJsonSchema schema) {
+		return validator(Validator.schema(schema));
+	}
+
+	/**
+	 * Create new {@link CollectionOptions} with already given settings and {@code validationOptions} set to given
+	 * {@link Validator}.
+	 *
+	 * @param validator can be {@literal null}.
+	 * @return new {@link CollectionOptions}.
+	 * @since 2.1
+	 */
+	public CollectionOptions validator(@Nullable Validator validator) {
+		return validation(validationOptions.validator(validator));
+	}
+
+	/**
+	 * Create new {@link CollectionOptions} with already given settings and {@code validationLevel} set to
+	 * {@link ValidationLevel#OFF}.
+	 *
+	 * @return new {@link CollectionOptions}.
+	 * @since 2.1
+	 */
+	public CollectionOptions disableValidation() {
+		return schemaValidationLevel(ValidationLevel.OFF);
+	}
+
+	/**
+	 * Create new {@link CollectionOptions} with already given settings and {@code validationLevel} set to
+	 * {@link ValidationLevel#STRICT}.
+	 *
+	 * @return new {@link CollectionOptions}.
+	 * @since 2.1
+	 */
+	public CollectionOptions strictValidation() {
+		return schemaValidationLevel(ValidationLevel.STRICT);
+	}
+
+	/**
+	 * Create new {@link CollectionOptions} with already given settings and {@code validationLevel} set to
+	 * {@link ValidationLevel#MODERATE}.
+	 *
+	 * @return new {@link CollectionOptions}.
+	 * @since 2.1
+	 */
+	public CollectionOptions moderateValidation() {
+		return schemaValidationLevel(ValidationLevel.MODERATE);
+	}
+
+	/**
+	 * Create new {@link CollectionOptions} with already given settings and {@code validationAction} set to
+	 * {@link ValidationAction#WARN}.
+	 *
+	 * @return new {@link CollectionOptions}.
+	 * @since 2.1
+	 */
+	public CollectionOptions warnOnValidationError() {
+		return schemaValidationAction(ValidationAction.WARN);
+	}
+
+	/**
+	 * Create new {@link CollectionOptions} with already given settings and {@code validationAction} set to
+	 * {@link ValidationAction#ERROR}.
+	 *
+	 * @return new {@link CollectionOptions}.
+	 * @since 2.1
+	 */
+	public CollectionOptions failOnValidationError() {
+		return schemaValidationAction(ValidationAction.ERROR);
+	}
+
+	/**
+	 * Create new {@link CollectionOptions} with already given settings and {@code validationLevel} set given
+	 * {@link ValidationLevel}.
+	 *
+	 * @param validationLevel must not be {@literal null}.
+	 * @return new {@link CollectionOptions}.
+	 * @since 2.1
+	 */
+	public CollectionOptions schemaValidationLevel(ValidationLevel validationLevel) {
+
+		Assert.notNull(validationLevel, "ValidationLevel must not be null!");
+		return validation(validationOptions.validationLevel(validationLevel));
+	}
+
+	/**
+	 * Create new {@link CollectionOptions} with already given settings and {@code validationAction} set given
+	 * {@link ValidationAction}.
+	 *
+	 * @param validationAction must not be {@literal null}.
+	 * @return new {@link CollectionOptions}.
+	 * @since 2.1
+	 */
+	public CollectionOptions schemaValidationAction(ValidationAction validationAction) {
+
+		Assert.notNull(validationAction, "ValidationAction must not be null!");
+		return validation(validationOptions.validationAction(validationAction));
+	}
+
+	/**
+	 * Create new {@link CollectionOptions} with the given {@link ValidationOptions}.
+	 *
+	 * @param validationOptions must not be {@literal null}. Use {@link ValidationOptions#none()} to remove validation.
+	 * @return new {@link CollectionOptions}.
+	 * @since 2.1
+	 */
+	public CollectionOptions validation(ValidationOptions validationOptions) {
+
+		Assert.notNull(validationOptions, "ValidationOptions must not be null!");
+		return new CollectionOptions(size, maxDocuments, capped, collation, validationOptions);
 	}
 
 	/**
@@ -163,4 +294,104 @@ public class CollectionOptions {
 	public Optional<Collation> getCollation() {
 		return Optional.ofNullable(collation);
 	}
+
+	/**
+	 * Get the {@link MongoJsonSchema} for the collection.
+	 *
+	 * @return {@link Optional#empty()} if not set.
+	 * @since 2.1
+	 */
+	public Optional<ValidationOptions> getValidationOptions() {
+		return validationOptions.isEmpty() ? Optional.empty() : Optional.of(validationOptions);
+	}
+
+	/**
+	 * Encapsulation of ValidationOptions options.
+	 *
+	 * @author Christoph Strobl
+	 * @author Andreas Zink
+	 * @since 2.1
+	 */
+	@RequiredArgsConstructor
+	public static class ValidationOptions {
+
+		private static final ValidationOptions NONE = new ValidationOptions(null, null, null);
+
+		private final @Nullable Validator validator;
+		private final @Nullable ValidationLevel validationLevel;
+		private final @Nullable ValidationAction validationAction;
+
+		/**
+		 * Create an empty {@link ValidationOptions}.
+		 *
+		 * @return never {@literal null}.
+		 */
+		public static ValidationOptions none() {
+			return NONE;
+		}
+
+		/**
+		 * Define the {@link Validator} to be used for document validation.
+		 *
+		 * @param validator can be {@literal null}.
+		 * @return new instance of {@link ValidationOptions}.
+		 */
+		public ValidationOptions validator(@Nullable Validator validator) {
+			return new ValidationOptions(validator, validationLevel, validationAction);
+		}
+
+		/**
+		 * Define the validation level to apply.
+		 *
+		 * @param validationLevel can be {@literal null}.
+		 * @return new instance of {@link ValidationOptions}.
+		 */
+		public ValidationOptions validationLevel(ValidationLevel validationLevel) {
+			return new ValidationOptions(validator, validationLevel, validationAction);
+		}
+
+		/**
+		 * Define the validation action to take.
+		 *
+		 * @param validationAction can be {@literal null}.
+		 * @return new instance of {@link ValidationOptions}.
+		 */
+		public ValidationOptions validationAction(ValidationAction validationAction) {
+			return new ValidationOptions(validator, validationLevel, validationAction);
+		}
+
+		/**
+		 * Get the {@link Validator} to use.
+		 *
+		 * @return never {@literal null}.
+		 */
+		public Optional<Validator> getValidator() {
+			return Optional.ofNullable(validator);
+		}
+
+		/**
+		 * Get the {@code validationLevel} to apply.
+		 *
+		 * @return {@link Optional#empty()} if not set.
+		 */
+		public Optional<ValidationLevel> getValidationLevel() {
+			return Optional.ofNullable(validationLevel);
+		}
+
+		/**
+		 * Get the {@code validationAction} to perform.
+		 *
+		 * @return @return {@link Optional#empty()} if not set.
+		 */
+		public Optional<ValidationAction> getValidationAction() {
+			return Optional.ofNullable(validationAction);
+		}
+
+		/**
+		 * @return {@literal true} if no arguments set.
+		 */
+		boolean isEmpty() {
+			return !Optionals.isAnyPresent(getValidator(), getValidationAction(), getValidationLevel());
+		}
+	}
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -21,7 +21,7 @@ import com.mongodb.client.FindIterable;
 
 /**
  * Simple callback interface to allow customization of a {@link FindIterable}.
- * 
+ *
  * @author Oliver Gierke
  * @author Christoph Strobl
  */
@@ -29,7 +29,7 @@ interface CursorPreparer {
 
 	/**
 	 * Prepare the given cursor (apply limits, skips and so on). Returns the prepared cursor.
-	 * 
+	 *
 	 * @param cursor
 	 */
 	FindIterable<Document> prepare(FindIterable<Document> cursor);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2017 the original author or authors.
+ * Copyright 2010-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -274,19 +274,10 @@ class DefaultBulkOperations implements BulkOperations {
 	public com.mongodb.bulk.BulkWriteResult execute() {
 
 		try {
-
-			MongoCollection<Document> collection = mongoOperations.getCollection(collectionName);
-			if (defaultWriteConcern != null) {
-				collection = collection.withWriteConcern(defaultWriteConcern);
-			}
-
-			return collection.bulkWrite(models.stream().map(this::mapWriteModel).collect(Collectors.toList()), bulkOptions);
-
-		} catch (BulkWriteException o_O) {
-
-			DataAccessException toThrow = exceptionTranslator.translateExceptionIfPossible(o_O);
-			throw toThrow == null ? o_O : toThrow;
-
+			
+			return mongoOperations.execute(collectionName, collection -> {
+				return collection.bulkWrite(models.stream().map(this::mapWriteModel).collect(Collectors.toList()), bulkOptions);
+			});
 		} finally {
 			this.bulkOptions = getBulkWriteOptions(bulkOperationContext.getBulkMode());
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 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 static org.springframework.data.mongodb.core.MongoTemplate.*;
-
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -50,18 +48,22 @@ public class DefaultIndexOperations implements IndexOperations {
 
 	private static final String PARTIAL_FILTER_EXPRESSION_KEY = "partialFilterExpression";
 
-	private final MongoDbFactory mongoDbFactory;
 	private final String collectionName;
 	private final QueryMapper mapper;
 	private final @Nullable Class<?> type;
 
+	private MongoOperations mongoOperations;
+
 	/**
 	 * 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}.
+	 * @deprecated since 2.1. Please use
+	 *             {@link DefaultIndexOperations#DefaultIndexOperations(MongoOperations, String, Class)}.
 	 */
+	@Deprecated
 	public DefaultIndexOperations(MongoDbFactory mongoDbFactory, String collectionName, QueryMapper queryMapper) {
 		this(mongoDbFactory, collectionName, queryMapper, null);
 	}
@@ -74,7 +76,10 @@ public class DefaultIndexOperations implements IndexOperations {
 	 * @param queryMapper must not be {@literal null}.
 	 * @param type Type used for mapping potential partial index filter expression. Can be {@literal null}.
 	 * @since 1.10
+	 * @deprecated since 2.1. Please use
+	 *             {@link DefaultIndexOperations#DefaultIndexOperations(MongoOperations, String, Class)}.
 	 */
+	@Deprecated
 	public DefaultIndexOperations(MongoDbFactory mongoDbFactory, String collectionName, QueryMapper queryMapper,
 			@Nullable Class<?> type) {
 
@@ -82,10 +87,29 @@ public class DefaultIndexOperations implements IndexOperations {
 		Assert.notNull(collectionName, "Collection name can not be null!");
 		Assert.notNull(queryMapper, "QueryMapper must not be null!");
 
-		this.mongoDbFactory = mongoDbFactory;
 		this.collectionName = collectionName;
 		this.mapper = queryMapper;
 		this.type = type;
+		this.mongoOperations = new MongoTemplate(mongoDbFactory);
+	}
+
+	/**
+	 * Creates a new {@link DefaultIndexOperations}.
+	 *
+	 * @param mongoOperations must not be {@literal null}.
+	 * @param collectionName must not be {@literal null} or empty.
+	 * @param type can be {@literal null}.
+	 * @since 2.1
+	 */
+	public DefaultIndexOperations(MongoOperations mongoOperations, String collectionName, @Nullable Class<?> type) {
+
+		Assert.notNull(mongoOperations, "MongoOperations must not be null!");
+		Assert.hasText(collectionName, "Collection name must not be null or empty!");
+
+		this.mongoOperations = mongoOperations;
+		this.mapper = new QueryMapper(mongoOperations.getConverter());
+		this.collectionName = collectionName;
+		this.type = type;
 	}
 
 	/*
@@ -187,11 +211,10 @@ public class DefaultIndexOperations implements IndexOperations {
 
 		Assert.notNull(callback, "CollectionCallback must not be null!");
 
-		try {
-			MongoCollection<Document> collection = mongoDbFactory.getDb().getCollection(collectionName);
-			return callback.doInCollection(collection);
-		} catch (RuntimeException e) {
-			throw potentiallyConvertRuntimeException(e, mongoDbFactory.getExceptionTranslator());
+		if (type != null) {
+			return mongoOperations.execute(type, callback);
 		}
+
+		return mongoOperations.execute(collectionName, callback);
 	}
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2017 the original author or authors.
+ * Copyright 2014-2018 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.
@@ -43,9 +43,10 @@ import com.mongodb.client.MongoDatabase;
 
 /**
  * Default implementation of {@link ScriptOperations} capable of saving and executing {@link ServerSideJavaScript}.
- * 
+ *
  * @author Christoph Strobl
  * @author Oliver Gierke
+ * @author Mark Paluch
  * @since 1.7
  */
 class DefaultScriptOperations implements ScriptOperations {
@@ -141,7 +142,7 @@ class DefaultScriptOperations implements ScriptOperations {
 
 		Assert.hasText(scriptName, "ScriptName must not be null or empty!");
 
-		return mongoOperations.exists(query(where("name").is(scriptName)), NamedMongoScript.class, SCRIPT_COLLECTION_NAME);
+		return mongoOperations.exists(query(where("_id").is(scriptName)), NamedMongoScript.class, SCRIPT_COLLECTION_NAME);
 	}
 
 	/*
@@ -190,7 +191,7 @@ class DefaultScriptOperations implements ScriptOperations {
 	 * Generate a valid name for the {@literal JavaScript}. MongoDB requires an id of type String for scripts. Calling
 	 * scripts having {@link ObjectId} as id fails. Therefore we create a random UUID without {@code -} (as this won't
 	 * work) an prefix the result with {@link #SCRIPT_NAME_PREFIX}.
-	 * 
+	 *
 	 * @return
 	 */
 	private static String generateScriptName() {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2016 the original author or authors.
+ * Copyright 2010-2018 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.
@@ -26,7 +26,7 @@ import com.mongodb.MongoException;
  * about exception handling. {@MongoException}s will be caught and translated by the calling MongoTemplate An
  * DocumentCallbackHandler is typically stateful: It keeps the result state within the object, to be available later for
  * later inspection.
- * 
+ *
  * @author Mark Pollack
  * @author Grame Rocher
  * @author Oliver Gierke

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.
@@ -19,11 +19,14 @@ import java.util.List;
 import java.util.Optional;
 import java.util.stream.Stream;
 
+import org.springframework.dao.DataAccessException;
 import org.springframework.data.geo.GeoResults;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.lang.Nullable;
 
+import com.mongodb.client.MongoCollection;
+
 /**
  * {@link ExecutableFindOperation} allows creation and execution of MongoDB find operations in a fluent API style.
  * <br />
@@ -202,7 +205,7 @@ public interface ExecutableFindOperation {
 	 * @author Christoph Strobl
 	 * @since 2.0
 	 */
-	interface FindWithProjection<T> extends FindWithQuery<T> {
+	interface FindWithProjection<T> extends FindWithQuery<T>, FindDistinct {
 
 		/**
 		 * Define the target type fields should be mapped to. <br />
@@ -214,6 +217,101 @@ public interface ExecutableFindOperation {
 		 * @throws IllegalArgumentException if resultType is {@literal null}.
 		 */
 		<R> FindWithQuery<R> as(Class<R> resultType);
+
+	}
+
+	/**
+	 * Distinct Find support.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface FindDistinct {
+
+		/**
+		 * Finds the distinct values for a specified {@literal field} across a single {@link MongoCollection} or view.
+		 *
+		 * @param field name of the field. Must not be {@literal null}.
+		 * @return new instance of {@link TerminatingDistinct}.
+		 * @throws IllegalArgumentException if field is {@literal null}.
+		 */
+		TerminatingDistinct<Object> distinct(String field);
+	}
+
+	/**
+	 * Result type override. Optional.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface DistinctWithProjection {
+
+		/**
+		 * Define the target type the result should be mapped to. <br />
+		 * Skip this step if you are anyway fine with the default conversion.
+		 * <dl>
+		 * <dt>{@link Object} (the default)</dt>
+		 * <dd>Result is mapped according to the {@link org.bson.BsonType} converting eg. {@link org.bson.BsonString} into
+		 * plain {@link String}, {@link org.bson.BsonInt64} to {@link Long}, etc. always picking the most concrete type with
+		 * respect to the domain types property.<br />
+		 * Any {@link org.bson.BsonType#DOCUMENT} is run through the {@link org.springframework.data.convert.EntityReader}
+		 * to obtain the domain type. <br />
+		 * Using {@link Object} also works for non strictly typed fields. Eg. a mixture different types like fields using
+		 * {@link String} in one {@link org.bson.Document} while {@link Long} in another.</dd>
+		 * <dt>Any Simple type like {@link String} or {@link Long}.</dt>
+		 * <dd>The result is mapped directly by the MongoDB Java driver and the {@link org.bson.codecs.CodeCodec Codecs} in
+		 * place. This works only for results where all documents considered for the operation use the very same type for
+		 * the field.</dd>
+		 * <dt>Any Domain type</dt>
+		 * <dd>Domain types can only be mapped if the if the result of the actual {@code distinct()} operation returns
+		 * {@link org.bson.BsonType#DOCUMENT}.</dd>
+		 * <dt>{@link org.bson.BsonValue}</dt>
+		 * <dd>Using {@link org.bson.BsonValue} allows retrieval of the raw driver specific format, which returns eg.
+		 * {@link org.bson.BsonString}.</dd>
+		 * </dl>
+		 *
+		 * @param resultType must not be {@literal null}.
+		 * @param <R> result type.
+		 * @return new instance of {@link TerminatingDistinct}.
+		 * @throws IllegalArgumentException if resultType is {@literal null}.
+		 */
+		<R> TerminatingDistinct<R> as(Class<R> resultType);
+	}
+
+	/**
+	 * Result restrictions. Optional.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface DistinctWithQuery<T> extends DistinctWithProjection {
+
+		/**
+		 * Set the filter query to be used.
+		 *
+		 * @param query must not be {@literal null}.
+		 * @return new instance of {@link TerminatingDistinct}.
+		 * @throws IllegalArgumentException if resultType is {@literal null}.
+		 */
+		TerminatingDistinct<T> matching(Query query);
+	}
+
+	/**
+	 * Terminating distinct find operations.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface TerminatingDistinct<T> extends DistinctWithQuery<T> {
+
+		/**
+		 * Get all matching distinct field values.
+		 *
+		 * @return empty {@link List} if not match found. Never {@literal null}.
+		 * @throws DataAccessException if eg. result cannot be converted correctly which may happen if the document contains
+		 *           {@link String} whereas the result type is specified as {@link Long}.
+		 */
+		List<T> all();
 	}
 
 	/**
@@ -222,5 +320,5 @@ public interface ExecutableFindOperation {
 	 * @author Christoph Strobl
 	 * @since 2.0
 	 */
-	interface ExecutableFind<T> extends FindWithCollection<T>, FindWithProjection<T> {}
+	interface ExecutableFind<T> extends FindWithCollection<T>, FindWithProjection<T>, FindDistinct {}
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.
@@ -205,6 +205,19 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 			return template.exists(query, domainType, getCollectionName());
 		}
 
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ExecutableFindOperation.FindDistinct#distinct(java.lang.String)
+		 */
+		@SuppressWarnings("unchecked")
+		@Override
+		public TerminatingDistinct<Object> distinct(String field) {
+
+			Assert.notNull(field, "Field must not be null!");
+
+			return new DistinctOperationSupport(this, field);
+		}
+
 		private List<T> doFind(@Nullable CursorPreparer preparer) {
 
 			Document queryObject = query.getQueryObject();
@@ -214,6 +227,12 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 					getCursorPreparer(query, preparer));
 		}
 
+		private List<T> doFindDistinct(String field) {
+
+			return template.findDistinct(query, field, getCollectionName(), domainType,
+					returnType == domainType ? (Class<T>) Object.class : returnType);
+		}
+
 		private CloseableIterator<T> doStream() {
 			return template.doStream(query, domainType, getCollectionName(), returnType);
 		}
@@ -261,4 +280,54 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 			return this;
 		}
 	}
+
+	/**
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	static class DistinctOperationSupport<T> implements TerminatingDistinct<T> {
+
+		private final String field;
+		private final ExecutableFindSupport<T> delegate;
+
+		public DistinctOperationSupport(ExecutableFindSupport<T> delegate, String field) {
+
+			this.delegate = delegate;
+			this.field = field;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ExecutableFindOperation.DistinctWithProjection#as(java.lang.Class)
+		 */
+		@Override
+		@SuppressWarnings("unchecked")
+		public <R> TerminatingDistinct<R> as(Class<R> resultType) {
+
+			Assert.notNull(resultType, "ResultType must not be null!");
+
+			return new DistinctOperationSupport<>((ExecutableFindSupport) delegate.as(resultType), field);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ExecutableFindOperation.DistinctWithQuery#matching(org.springframework.data.mongodb.core.query.Query)
+		 */
+		@Override
+		public TerminatingDistinct<T> matching(Query query) {
+
+			Assert.notNull(query, "Query must not be null!");
+
+			return new DistinctOperationSupport<>((ExecutableFindSupport<T>) delegate.matching(query), field);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ExecutableFindOperation.TerminatingDistinct#all()
+		 */
+		@Override
+		public List<T> all() {
+			return delegate.doFindDistinct(field);
+		}
+	}
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2017 the original author or authors.
+ * Copyright 2010-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -20,7 +20,7 @@ import org.springframework.util.Assert;
 
 /**
  * Value object to mitigate different representations of geo command execution results in MongoDB.
- * 
+ *
  * @author Oliver Gierke
  * @author Christoph Strobl
  * @soundtrack Fruitcake - Jeff Coffin (The Inside of the Outside)
@@ -34,7 +34,7 @@ class GeoCommandStatistics {
 
 	/**
 	 * Creates a new {@link GeoCommandStatistics} instance with the given source document.
-	 * 
+	 *
 	 * @param source must not be {@literal null}.
 	 */
 	private GeoCommandStatistics(Document source) {
@@ -45,7 +45,7 @@ class GeoCommandStatistics {
 
 	/**
 	 * Creates a new {@link GeoCommandStatistics} from the given command result extracting the statistics.
-	 * 
+	 *
 	 * @param commandResult must not be {@literal null}.
 	 * @return
 	 */
@@ -60,7 +60,7 @@ class GeoCommandStatistics {
 	/**
 	 * Returns the average distance reported by the command result. Mitigating a removal of the field in case the command
 	 * didn't return any result introduced in MongoDB 3.2 RC1.
-	 * 
+	 *
 	 * @return
 	 * @see <a href="https://jira.mongodb.org/browse/SERVER-21024">MongoDB Jira SERVER-21024</a>
 	 */

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 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.
@@ -29,7 +29,7 @@ import com.mongodb.WriteConcern;
  * <li>REMOVE has null document</li>
  * <li>INSERT_LIST has null entityType, document, and query</li>
  * </ul>
- * 
+ *
  * @author Mark Pollack
  * @author Oliver Gierke
  * @author Christoph Strobl
@@ -46,7 +46,7 @@ public class MongoAction {
 
 	/**
 	 * Create an instance of a {@link MongoAction}.
-	 * 
+	 *
 	 * @param defaultWriteConcern the default write concern. Can be {@literal null}.
 	 * @param mongoActionOperation action being taken against the collection. Must not be {@literal null}.
 	 * @param collectionName the collection name, must not be {@literal null} or empty.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2015 the original author or authors.
+ * Copyright 2011-2018 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.
@@ -18,7 +18,7 @@ package org.springframework.data.mongodb.core;
 /**
  * Enumeration for operations on a collection. Used with {@link MongoAction} to help determine the WriteConcern to use
  * for a given mutating operation
- * 
+ *
  * @author Mark Pollack
  * @author Oliver Gierke
  * @see MongoAction

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 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,7 +25,7 @@ import com.mongodb.client.MongoDatabase;
 
 /**
  * Mongo server administration exposed via JMX annotations
- * 
+ *
  * @author Mark Pollack
  * @author Thomas Darimont
  * @author Mark Paluch

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2014 the original author or authors.
+ * Copyright 2011-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2017 the original author or authors.
+ * Copyright 2013-2018 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.
@@ -23,7 +23,7 @@ import com.mongodb.WriteResult;
 
 /**
  * Mongo-specific {@link DataIntegrityViolationException}.
- * 
+ *
  * @author Oliver Gierke
  */
 public class MongoDataIntegrityViolationException extends DataIntegrityViolationException {
@@ -35,7 +35,7 @@ public class MongoDataIntegrityViolationException extends DataIntegrityViolation
 
 	/**
 	 * Creates a new {@link MongoDataIntegrityViolationException} using the given message and {@link WriteResult}.
-	 * 
+	 *
 	 * @param message the exception message
 	 * @param writeResult the {@link WriteResult} that causes the exception, must not be {@literal null}.
 	 * @param actionOperation the {@link MongoActionOperation} that caused the exception, must not be {@literal null}.
@@ -54,7 +54,7 @@ public class MongoDataIntegrityViolationException extends DataIntegrityViolation
 
 	/**
 	 * Returns the {@link WriteResult} that caused the exception.
-	 * 
+	 *
 	 * @return the writeResult
 	 */
 	public WriteResult getWriteResult() {
@@ -63,7 +63,7 @@ public class MongoDataIntegrityViolationException extends DataIntegrityViolation
 
 	/**
 	 * Returns the {@link MongoActionOperation} in which the current exception occured.
-	 * 
+	 *
 	 * @return the actionOperation
 	 */
 	public MongoActionOperation getActionOperation() {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2017 the original author or authors.
+ * Copyright 2010-2018 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.
@@ -29,6 +29,7 @@ import org.springframework.dao.InvalidDataAccessResourceUsageException;
 import org.springframework.dao.PermissionDeniedDataAccessException;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
 import org.springframework.data.mongodb.BulkOperationException;
+import org.springframework.data.mongodb.ClientSessionException;
 import org.springframework.data.mongodb.UncategorizedMongoDbException;
 import org.springframework.data.mongodb.util.MongoDbErrorCodes;
 import org.springframework.lang.Nullable;
@@ -44,7 +45,7 @@ import com.mongodb.bulk.BulkWriteError;
  * Simple {@link PersistenceExceptionTranslator} for Mongo. Convert the given runtime exception to an appropriate
  * exception from the {@code org.springframework.dao} hierarchy. Return {@literal null} if no translation is
  * appropriate: any other exception may have resulted from user code, and should not be translated.
- * 
+ *
  * @author Oliver Gierke
  * @author Michal Vich
  * @author Christoph Strobl
@@ -119,18 +120,28 @@ public class MongoExceptionTranslator implements PersistenceExceptionTranslator
 			int code = ((MongoException) ex).getCode();
 
 			if (MongoDbErrorCodes.isDuplicateKeyCode(code)) {
-				throw new DuplicateKeyException(ex.getMessage(), ex);
+				return new DuplicateKeyException(ex.getMessage(), ex);
 			} else if (MongoDbErrorCodes.isDataAccessResourceFailureCode(code)) {
-				throw new DataAccessResourceFailureException(ex.getMessage(), ex);
+				return new DataAccessResourceFailureException(ex.getMessage(), ex);
 			} else if (MongoDbErrorCodes.isInvalidDataAccessApiUsageCode(code) || code == 10003 || code == 12001
 					|| code == 12010 || code == 12011 || code == 12012) {
-				throw new InvalidDataAccessApiUsageException(ex.getMessage(), ex);
+				return new InvalidDataAccessApiUsageException(ex.getMessage(), ex);
 			} else if (MongoDbErrorCodes.isPermissionDeniedCode(code)) {
-				throw new PermissionDeniedDataAccessException(ex.getMessage(), ex);
+				return new PermissionDeniedDataAccessException(ex.getMessage(), ex);
 			}
 			return new UncategorizedMongoDbException(ex.getMessage(), ex);
 		}
 
+		// may interfere with OmitStackTraceInFastThrow (enabled by default).
+		// see https://jira.spring.io/browse/DATAMONGO-1905
+		if (ex instanceof IllegalStateException) {
+			for (StackTraceElement elm : ex.getStackTrace()) {
+				if (elm.getClassName().contains("ClientSession")) {
+					return new ClientSessionException(ex.getMessage(), ex);
+				}
+			}
+		}
+
 		// If we get here, we have an exception that resulted from user code,
 		// rather than the persistence provider, so we return null to indicate
 		// that translation should not occur.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 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.
@@ -18,6 +18,8 @@ package org.springframework.data.mongodb.core;
 import java.util.Collection;
 import java.util.List;
 import java.util.Set;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
 
 import org.bson.Document;
 import org.springframework.data.geo.GeoResults;
@@ -39,12 +41,15 @@ import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
 import org.springframework.data.util.CloseableIterator;
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 
+import com.mongodb.ClientSessionOptions;
 import com.mongodb.Cursor;
 import com.mongodb.ReadPreference;
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.result.DeleteResult;
 import com.mongodb.client.result.UpdateResult;
+import com.mongodb.session.ClientSession;
 
 /**
  * Interface that specifies a basic set of MongoDB operations. Implemented by {@link MongoTemplate}. Not often used but
@@ -72,11 +77,11 @@ public interface MongoOperations extends FluentMongoOperations {
 	String getCollectionName(Class<?> entityClass);
 
 	/**
-	 * Execute the a MongoDB command expressed as a JSON string. This will call the method JSON.parse that is part of the
-	 * MongoDB driver to convert the JSON string to a Document. Any errors that result from executing this command will be
+	 * Execute the a MongoDB command expressed as a JSON string. Parsing is delegated to {@link Document#parse(String)} to
+	 * obtain the {@link Document} holding the actual command. Any errors that result from executing this command will be
 	 * converted into Spring's DAO exception hierarchy.
 	 *
-	 * @param jsonCommand a MongoDB command expressed as a JSON string.
+	 * @param jsonCommand a MongoDB command expressed as a JSON string. Must not be {@literal null}.
 	 * @return a result object returned by the action.
 	 */
 	Document executeCommand(String jsonCommand);
@@ -151,6 +156,64 @@ public interface MongoOperations extends FluentMongoOperations {
 	@Nullable
 	<T> T execute(String collectionName, CollectionCallback<T> action);
 
+	/**
+	 * Obtain a {@link ClientSession session} bound instance of {@link SessionScoped} binding a new {@link ClientSession}
+	 * with given {@literal sessionOptions} to each and every command issued against MongoDB.
+	 *
+	 * @param sessionOptions must not be {@literal null}.
+	 * @return new instance of {@link SessionScoped}. Never {@literal null}.
+	 * @since 2.1
+	 */
+	SessionScoped withSession(ClientSessionOptions sessionOptions);
+
+	/**
+	 * Obtain a {@link ClientSession session} bound instance of {@link SessionScoped} binding the {@link ClientSession}
+	 * provided by the given {@link Supplier} to each and every command issued against MongoDB.
+	 * <p />
+	 * <strong>Note:</strong> It is up to the caller to manage the {@link ClientSession} lifecycle. Use the
+	 * {@link SessionScoped#execute(SessionCallback, Consumer)} hook to potentially close the {@link ClientSession}.
+	 *
+	 * @param sessionProvider must not be {@literal null}.
+	 * @since 2.1
+	 */
+	default SessionScoped withSession(Supplier<ClientSession> sessionProvider) {
+
+		Assert.notNull(sessionProvider, "SessionProvider must not be null!");
+
+		return new SessionScoped() {
+
+			private final Object lock = new Object();
+			private @Nullable ClientSession session = null;
+
+			@Override
+			public <T> T execute(SessionCallback<T> action, Consumer<ClientSession> onComplete) {
+
+				synchronized (lock) {
+					if (session == null) {
+						session = sessionProvider.get();
+					}
+				}
+
+				try {
+					return action.doInSession(MongoOperations.this.withSession(session));
+				} finally {
+					onComplete.accept(session);
+				}
+			}
+		};
+	}
+
+	/**
+	 * Obtain a {@link ClientSession} bound instance of {@link MongoOperations}.
+	 * <p />
+	 * <strong>Note:</strong> It is up to the caller to manage the {@link ClientSession} lifecycle.
+	 *
+	 * @param session must not be {@literal null}.
+	 * @return {@link ClientSession} bound instance of {@link MongoOperations}.
+	 * @since 2.1
+	 */
+	MongoOperations withSession(ClientSession session);
+
 	/**
 	 * Executes the given {@link Query} on the entity collection of the specified {@code entityType} backed by a Mongo DB
 	 * {@link Cursor}.
@@ -710,7 +773,66 @@ public interface MongoOperations extends FluentMongoOperations {
 	<T> T findById(Object id, Class<T> entityClass, String collectionName);
 
 	/**
-	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify<a/>
+	 * Finds the distinct values for a specified {@literal field} across a single {@link MongoCollection} or view and
+	 * returns the results in a {@link List}.
+	 *
+	 * @param field the name of the field to inspect for distinct values. Must not be {@literal null}.
+	 * @param entityClass the domain type used for determining the actual {@link MongoCollection}. Must not be
+	 *          {@literal null}.
+	 * @param resultClass the result type. Must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	default <T> List<T> findDistinct(String field, Class<?> entityClass, Class<T> resultClass) {
+		return findDistinct(new Query(), field, entityClass, resultClass);
+	}
+
+	/**
+	 * Finds the distinct values for a specified {@literal field} across a single {@link MongoCollection} or view and
+	 * returns the results in a {@link List}.
+	 *
+	 * @param query filter {@link Query} to restrict search. Must not be {@literal null}.
+	 * @param field the name of the field to inspect for distinct values. Must not be {@literal null}.
+	 * @param entityClass the domain type used for determining the actual {@link MongoCollection} and mapping the
+	 *          {@link Query} to the domain type fields. Must not be {@literal null}.
+	 * @param resultClass the result type. Must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	<T> List<T> findDistinct(Query query, String field, Class<?> entityClass, Class<T> resultClass);
+
+	/**
+	 * Finds the distinct values for a specified {@literal field} across a single {@link MongoCollection} or view and
+	 * returns the results in a {@link List}.
+	 *
+	 * @param query filter {@link Query} to restrict search. Must not be {@literal null}.
+	 * @param field the name of the field to inspect for distinct values. Must not be {@literal null}.
+	 * @param collectionName the explicit name of the actual {@link MongoCollection}. Must not be {@literal null}.
+	 * @param entityClass the domain type used for mapping the {@link Query} to the domain type fields.
+	 * @param resultClass the result type. Must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	<T> List<T> findDistinct(Query query, String field, String collectionName, Class<?> entityClass,
+			Class<T> resultClass);
+
+	/**
+	 * Finds the distinct values for a specified {@literal field} across a single {@link MongoCollection} or view and
+	 * returns the results in a {@link List}.
+	 *
+	 * @param query filter {@link Query} to restrict search. Must not be {@literal null}.
+	 * @param field the name of the field to inspect for distinct values. Must not be {@literal null}.
+	 * @param collection the explicit name of the actual {@link MongoCollection}. Must not be {@literal null}.
+	 * @param resultClass the result type. Must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	default <T> List<T> findDistinct(Query query, String field, String collection, Class<T> resultClass) {
+		return findDistinct(query, field, collection, Object.class, resultClass);
+	}
+
+	/**
+	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify <a/>
 	 * to apply provided {@link Update} on documents matching {@link Criteria} of given {@link Query}.
 	 *
 	 * @param query the {@link Query} class that specifies the {@link Criteria} used to find a record and also an optional
@@ -723,7 +845,7 @@ public interface MongoOperations extends FluentMongoOperations {
 	<T> T findAndModify(Query query, Update update, Class<T> entityClass);
 
 	/**
-	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify<a/>
+	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify <a/>
 	 * to apply provided {@link Update} on documents matching {@link Criteria} of given {@link Query}.
 	 *
 	 * @param query the {@link Query} class that specifies the {@link Criteria} used to find a record and also an optional
@@ -737,7 +859,7 @@ public interface MongoOperations extends FluentMongoOperations {
 	<T> T findAndModify(Query query, Update update, Class<T> entityClass, String collectionName);
 
 	/**
-	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify<a/>
+	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify <a/>
 	 * to apply provided {@link Update} on documents matching {@link Criteria} of given {@link Query} taking
 	 * {@link FindAndModifyOptions} into account.
 	 *
@@ -754,7 +876,7 @@ public interface MongoOperations extends FluentMongoOperations {
 	<T> T findAndModify(Query query, Update update, FindAndModifyOptions options, Class<T> entityClass);
 
 	/**
-	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify<a/>
+	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify <a/>
 	 * to apply provided {@link Update} on documents matching {@link Criteria} of given {@link Query} taking
 	 * {@link FindAndModifyOptions} into account.
 	 *
@@ -851,8 +973,8 @@ public interface MongoOperations extends FluentMongoOperations {
 	 * If you object has an "Id' property, it will be set with the generated Id from MongoDB. If your Id property is a
 	 * String then MongoDB ObjectId will be used to populate that string. Otherwise, the conversion from ObjectId to your
 	 * property type will be handled by Spring's BeanWrapper class that leverages Type Conversion API. See
-	 * <a href="http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation" >
-	 * Spring's Type Conversion"</a> for more details.
+	 * <a href="http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation" > Spring's Type
+	 * Conversion"</a> for more details.
 	 * <p/>
 	 * <p/>
 	 * Insert is used to initially store the object into the database. To update an existing object use the save method.
@@ -908,8 +1030,8 @@ public interface MongoOperations extends FluentMongoOperations {
 	 * If you object has an "Id' property, it will be set with the generated Id from MongoDB. If your Id property is a
 	 * String then MongoDB ObjectId will be used to populate that string. Otherwise, the conversion from ObjectId to your
 	 * property type will be handled by Spring's BeanWrapper class that leverages Type Conversion API. See
-	 * <a href="http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation" >
-	 * Spring's Type Conversion"</a> for more details.
+	 * <a href="http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation" > Spring's Type
+	 * Conversion"</a> for more details.
 	 *
 	 * @param objectToSave the object to store in the collection. Must not be {@literal null}.
 	 */
@@ -925,8 +1047,8 @@ public interface MongoOperations extends FluentMongoOperations {
 	 * If you object has an "Id' property, it will be set with the generated Id from MongoDB. If your Id property is a
 	 * String then MongoDB ObjectId will be used to populate that string. Otherwise, the conversion from ObjectId to your
 	 * property type will be handled by Spring's BeanWrapper class that leverages Type Conversion API. See <a
-	 * http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation">Spring's
-	 * Type Conversion"</a> for more details.
+	 * http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation">Spring's Type
+	 * Conversion"</a> for more details.
 	 *
 	 * @param objectToSave the object to store in the collection. Must not be {@literal null}.
 	 * @param collectionName name of the collection to store the object in. Must not be {@literal null}.
@@ -1083,6 +1205,7 @@ public interface MongoOperations extends FluentMongoOperations {
 	 * @param query the query document that specifies the criteria used to remove a record.
 	 * @param entityClass class that determines the collection to use.
 	 * @return the {@link DeleteResult} which lets you access the results of the previous delete.
+	 * @throws IllegalArgumentException when {@literal query} or {@literal entityClass} is {@literal null}.
 	 */
 	DeleteResult remove(Query query, Class<?> entityClass);
 
@@ -1094,6 +1217,8 @@ public interface MongoOperations extends FluentMongoOperations {
 	 * @param entityClass class of the pojo to be operated on. Can be {@literal null}.
 	 * @param collectionName name of the collection where the objects will removed, must not be {@literal null} or empty.
 	 * @return the {@link DeleteResult} which lets you access the results of the previous delete.
+	 * @throws IllegalArgumentException when {@literal query}, {@literal entityClass} or {@literal collectionName} is
+	 *           {@literal null}.
 	 */
 	DeleteResult remove(Query query, Class<?> entityClass, String collectionName);
 
@@ -1106,6 +1231,7 @@ public interface MongoOperations extends FluentMongoOperations {
 	 * @param query the query document that specifies the criteria used to remove a record.
 	 * @param collectionName name of the collection where the objects will removed, must not be {@literal null} or empty.
 	 * @return the {@link DeleteResult} which lets you access the results of the previous delete.
+	 * @throws IllegalArgumentException when {@literal query} or {@literal collectionName} is {@literal null}.
 	 */
 	DeleteResult remove(Query query, String collectionName);
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.
@@ -156,7 +156,7 @@ public interface ReactiveFindOperation {
 	/**
 	 * Result type override (optional).
 	 */
-	interface FindWithProjection<T> extends FindWithQuery<T> {
+	interface FindWithProjection<T> extends FindWithQuery<T>, FindDistinct {
 
 		/**
 		 * Define the target type fields should be mapped to. <br />
@@ -170,8 +170,101 @@ public interface ReactiveFindOperation {
 		<R> FindWithQuery<R> as(Class<R> resultType);
 	}
 
+	/**
+	 * Distinct Find support.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface FindDistinct {
+
+		/**
+		 * Finds the distinct values for a specified {@literal field} across a single
+		 * {@link com.mongodb.reactivestreams.client.MongoCollection} or view.
+		 *
+		 * @param field name of the field. Must not be {@literal null}.
+		 * @return new instance of {@link TerminatingDistinct}.
+		 * @throws IllegalArgumentException if field is {@literal null}.
+		 */
+		TerminatingDistinct<Object> distinct(String field);
+	}
+
+	/**
+	 * Result type override. Optional.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface DistinctWithProjection {
+
+		/**
+		 * Define the target type the result should be mapped to. <br />
+		 * Skip this step if you are anyway fine with the default conversion.
+		 * <dl>
+		 * <dt>{@link Object} (the default)</dt>
+		 * <dd>Result is mapped according to the {@link org.bson.BsonType} converting eg. {@link org.bson.BsonString} into
+		 * plain {@link String}, {@link org.bson.BsonInt64} to {@link Long}, etc. always picking the most concrete type with
+		 * respect to the domain types property.<br />
+		 * Any {@link org.bson.BsonType#DOCUMENT} is run through the {@link org.springframework.data.convert.EntityReader}
+		 * to obtain the domain type. <br />
+		 * Using {@link Object} also works for non strictly typed fields. Eg. a mixture different types like fields using
+		 * {@link String} in one {@link org.bson.Document} while {@link Long} in another.</dd>
+		 * <dt>Any Simple type like {@link String}, {@link Long}, ...</dt>
+		 * <dd>The result is mapped directly by the MongoDB Java driver and the {@link org.bson.codecs.CodeCodec Codecs} in
+		 * place. This works only for results where all documents considered for the operation use the very same type for
+		 * the field.</dd>
+		 * <dt>Any Domain type</dt>
+		 * <dd>Domain types can only be mapped if the if the result of the actual {@code distinct()} operation returns
+		 * {@link org.bson.BsonType#DOCUMENT}.</dd>
+		 * <dt>{@link org.bson.BsonValue}</dt>
+		 * <dd>Using {@link org.bson.BsonValue} allows retrieval of the raw driver specific format, which returns eg.
+		 * {@link org.bson.BsonString}.</dd>
+		 * </dl>
+		 *
+		 * @param resultType must not be {@literal null}.
+		 * @param <R> result type.
+		 * @return new instance of {@link TerminatingDistinct}.
+		 * @throws IllegalArgumentException if resultType is {@literal null}.
+		 */
+		<R> TerminatingDistinct<R> as(Class<R> resultType);
+	}
+
+	/**
+	 * Result restrictions. Optional.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface DistinctWithQuery<T> extends DistinctWithProjection {
+
+		/**
+		 * Set the filter query to be used.
+		 *
+		 * @param query must not be {@literal null}.
+		 * @return new instance of {@link TerminatingDistinct}.
+		 * @throws IllegalArgumentException if resultType is {@literal null}.
+		 */
+		TerminatingDistinct<T> matching(Query query);
+	}
+
+	/**
+	 * Terminating distinct find operations.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface TerminatingDistinct<T> extends DistinctWithQuery<T> {
+
+		/**
+		 * Get all matching distinct field values.
+		 *
+		 * @return empty {@link Flux} if not match found. Never {@literal null}.
+		 */
+		Flux<T> all();
+	}
+
 	/**
 	 * {@link ReactiveFind} provides methods for constructing lookup operations in a fluent way.
 	 */
-	interface ReactiveFind<T> extends FindWithCollection<T>, FindWithProjection<T> {}
+	interface ReactiveFind<T> extends FindWithCollection<T>, FindWithProjection<T>, FindDistinct {}
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.
@@ -19,7 +19,6 @@ import lombok.AccessLevel;
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import lombok.experimental.FieldDefaults;
-import org.springframework.lang.Nullable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -28,6 +27,7 @@ import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.SerializationUtils;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -196,6 +196,18 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 			return template.exists(query, domainType, getCollectionName());
 		}
 
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ReactiveFindOperation.FindDistinct#distinct(java.lang.String)
+		 */
+		@Override
+		public TerminatingDistinct<Object> distinct(String field) {
+
+			Assert.notNull(field, "Field must not be null!");
+
+			return new DistinctOperationSupport<>(this, field);
+		}
+
 		private Flux<T> doFind(@Nullable FindPublisherPreparer preparer) {
 
 			Document queryObject = query.getQueryObject();
@@ -205,6 +217,13 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 					preparer != null ? preparer : getCursorPreparer(query));
 		}
 
+		@SuppressWarnings("unchecked")
+		private Flux<T> doFindDistinct(String field) {
+
+			return template.findDistinct(query, field, getCollectionName(), domainType,
+					returnType == domainType ? (Class<T>) Object.class : returnType);
+		}
+
 		private FindPublisherPreparer getCursorPreparer(Query query) {
 			return template.new QueryFindPublisherPreparer(query, domainType);
 		}
@@ -216,5 +235,55 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 		private String asString() {
 			return SerializationUtils.serializeToJsonSafely(query);
 		}
+
+		/**
+		 * @author Christoph Strobl
+		 * @since 2.1
+		 */
+		static class DistinctOperationSupport<T> implements TerminatingDistinct<T> {
+
+			private final String field;
+			private final ReactiveFindSupport delegate;
+
+			public DistinctOperationSupport(ReactiveFindSupport delegate, String field) {
+
+				this.delegate = delegate;
+				this.field = field;
+			}
+
+			/*
+			 * (non-Javadoc)
+			 * @see org.springframework.data.mongodb.core.ReactiveFindOperation.DistinctWithProjection#as(java.lang.Class)
+			 */
+			@Override
+			public <R> TerminatingDistinct<R> as(Class<R> resultType) {
+
+				Assert.notNull(resultType, "ResultType must not be null!");
+
+				return new DistinctOperationSupport<>((ReactiveFindSupport) delegate.as(resultType), field);
+			}
+
+			/*
+			 * (non-Javadoc)
+			 * @see org.springframework.data.mongodb.core.ReactiveFindOperation.DistinctWithQuery#matching(org.springframework.data.mongodb.core.query.Query)
+			 */
+			@Override
+			@SuppressWarnings("unchecked")
+			public TerminatingDistinct<T> matching(Query query) {
+
+				Assert.notNull(query, "Query must not be null!");
+
+				return new DistinctOperationSupport<>((ReactiveFindSupport<T>) delegate.matching(query), field);
+			}
+
+			/*
+			 * (non-Javadoc)
+			 * @see org.springframework.data.mongodb.core..ReactiveFindOperation.TerminatingDistinct#all()
+			 */
+			@Override
+			public Flux<T> all() {
+				return delegate.doFindDistinct(field);
+			}
+		}
 	}
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.