diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseFactory.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseFactory.java
index 2d6a0d49d..17b0f7d92 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseFactory.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseFactory.java
@@ -19,6 +19,7 @@ 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;
@@ -88,4 +89,16 @@ public interface ReactiveMongoDatabaseFactory extends CodecRegistryProvider {
 	 * @since 2.1
 	 */
 	ReactiveMongoDatabaseFactory withSession(ClientSession session);
+
+	/**
+	 * Returns if the given {@link ReactiveMongoDatabaseFactory} is bound to a
+	 * {@link com.mongodb.reactivestreams.client.ClientSession} that has an
+	 * {@link com.mongodb.reactivestreams.client.ClientSession#hasActiveTransaction() active transaction}.
+	 *
+	 * @return {@literal true} if there's an active transaction, {@literal false} otherwise.
+	 * @since 2.2
+	 */
+	default boolean isTransactionActive() {
+		return false;
+	}
 }
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseUtils.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseUtils.java
new file mode 100644
index 000000000..60035bde5
--- /dev/null
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseUtils.java
@@ -0,0 +1,279 @@
+/*
+ * 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 reactor.core.publisher.Mono;
+import reactor.util.context.Context;
+
+import org.springframework.lang.Nullable;
+import org.springframework.transaction.NoTransactionException;
+import org.springframework.transaction.reactive.ReactiveResourceSynchronization;
+import org.springframework.transaction.reactive.TransactionSynchronization;
+import org.springframework.transaction.reactive.TransactionSynchronizationManager;
+import org.springframework.transaction.support.ResourceHolderSynchronization;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+import com.mongodb.ClientSessionOptions;
+import com.mongodb.reactivestreams.client.ClientSession;
+import com.mongodb.reactivestreams.client.MongoCollection;
+import com.mongodb.reactivestreams.client.MongoDatabase;
+
+/**
+ * Helper class for managing a {@link MongoDatabase} instances via {@link ReactiveMongoDatabaseFactory}. Used for
+ * obtaining {@link ClientSession session bound} resources, such as {@link MongoDatabase} and {@link MongoCollection}
+ * suitable for transactional usage.
+ * <p />
+ * <strong>Note:</strong> Intended for internal usage only.
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ */
+public class ReactiveMongoDatabaseUtils {
+
+	/**
+	 * Check if the {@link ReactiveMongoDatabaseFactory} is actually bound to a
+	 * {@link com.mongodb.reactivestreams.client.ClientSession} that has an active transaction, or if a
+	 * {@link org.springframework.transaction.reactive.TransactionSynchronization} has been registered for the
+	 * {@link ReactiveMongoDatabaseFactory resource} and if the associated
+	 * {@link com.mongodb.reactivestreams.client.ClientSession} has an
+	 * {@link com.mongodb.reactivestreams.client.ClientSession#hasActiveTransaction() active transaction}.
+	 *
+	 * @param databaseFactory the resource to check transactions for. Must not be {@literal null}.
+	 * @return {@literal true} if the factory has an ongoing transaction.
+	 */
+	public static Mono<Boolean> isTransactionActive(ReactiveMongoDatabaseFactory databaseFactory) {
+
+		if (databaseFactory.isTransactionActive()) {
+			return Mono.just(true);
+		}
+
+		return TransactionSynchronizationManager.currentTransaction().map(it -> {
+
+			ReactiveMongoResourceHolder holder = (ReactiveMongoResourceHolder) it.getResource(databaseFactory);
+			return holder != null && holder.hasActiveTransaction();
+		}).onErrorResume(NoTransactionException.class, e -> Mono.just(false));
+	}
+
+	/**
+	 * Obtain the default {@link MongoDatabase database} form the given {@link ReactiveMongoDatabaseFactory factory} using
+	 * {@link SessionSynchronization#ON_ACTUAL_TRANSACTION native session synchronization}.
+	 * <p />
+	 * Registers a {@link MongoSessionSynchronization MongoDB specific transaction synchronization} within the subscriber
+	 * {@link Context} if {@link TransactionSynchronizationManager#isSynchronizationActive() synchronization is active}.
+	 *
+	 * @param factory the {@link ReactiveMongoDatabaseFactory} to get the {@link MongoDatabase} from.
+	 * @return the {@link MongoDatabase} that is potentially associated with a transactional {@link ClientSession}.
+	 */
+	public static Mono<MongoDatabase> getDatabase(ReactiveMongoDatabaseFactory factory) {
+		return doGetMongoDatabase(null, factory, SessionSynchronization.ON_ACTUAL_TRANSACTION);
+	}
+
+	/**
+	 * Obtain the default {@link MongoDatabase database} form the given {@link ReactiveMongoDatabaseFactory factory}.
+	 * <p />
+	 * Registers a {@link MongoSessionSynchronization MongoDB specific transaction synchronization} within the subscriber
+	 * {@link Context} if {@link TransactionSynchronizationManager#isSynchronizationActive() synchronization is active}.
+	 *
+	 * @param factory the {@link ReactiveMongoDatabaseFactory} to get the {@link MongoDatabase} from.
+	 * @param sessionSynchronization the synchronization to use. Must not be {@literal null}.
+	 * @return the {@link MongoDatabase} that is potentially associated with a transactional {@link ClientSession}.
+	 */
+	public static Mono<MongoDatabase> getDatabase(ReactiveMongoDatabaseFactory factory,
+			SessionSynchronization sessionSynchronization) {
+		return doGetMongoDatabase(null, factory, sessionSynchronization);
+	}
+
+	/**
+	 * Obtain the {@link MongoDatabase database} with given name form the given {@link ReactiveMongoDatabaseFactory
+	 * factory} using {@link SessionSynchronization#ON_ACTUAL_TRANSACTION native session synchronization}.
+	 * <p />
+	 * Registers a {@link MongoSessionSynchronization MongoDB specific transaction synchronization} within the subscriber
+	 * {@link Context} if {@link TransactionSynchronizationManager#isSynchronizationActive() synchronization is active}.
+	 *
+	 * @param dbName the name of the {@link MongoDatabase} to get.
+	 * @param factory the {@link ReactiveMongoDatabaseFactory} to get the {@link MongoDatabase} from.
+	 * @return the {@link MongoDatabase} that is potentially associated with a transactional {@link ClientSession}.
+	 */
+	public static Mono<MongoDatabase> getDatabase(String dbName, ReactiveMongoDatabaseFactory factory) {
+		return doGetMongoDatabase(dbName, factory, SessionSynchronization.ON_ACTUAL_TRANSACTION);
+	}
+
+	/**
+	 * Obtain the {@link MongoDatabase database} with given name form the given {@link ReactiveMongoDatabaseFactory
+	 * factory}.
+	 * <p />
+	 * Registers a {@link MongoSessionSynchronization MongoDB specific transaction synchronization} within the subscriber
+	 * {@link Context} if {@link TransactionSynchronizationManager#isSynchronizationActive() synchronization is active}.
+	 *
+	 * @param dbName the name of the {@link MongoDatabase} to get.
+	 * @param factory the {@link ReactiveMongoDatabaseFactory} to get the {@link MongoDatabase} from.
+	 * @param sessionSynchronization the synchronization to use. Must not be {@literal null}.
+	 * @return the {@link MongoDatabase} that is potentially associated with a transactional {@link ClientSession}.
+	 */
+	public static Mono<MongoDatabase> getDatabase(String dbName, ReactiveMongoDatabaseFactory factory,
+			SessionSynchronization sessionSynchronization) {
+		return doGetMongoDatabase(dbName, factory, sessionSynchronization);
+	}
+
+	private static Mono<MongoDatabase> doGetMongoDatabase(@Nullable String dbName, ReactiveMongoDatabaseFactory factory,
+			SessionSynchronization sessionSynchronization) {
+
+		Assert.notNull(factory, "Factory must not be null!");
+
+		return TransactionSynchronizationManager.currentTransaction()
+				.filter(TransactionSynchronizationManager::isSynchronizationActive).flatMap(synchronizationManager -> {
+
+					Mono<ClientSession> session = doGetSession(synchronizationManager, factory, sessionSynchronization);
+
+					return session.map(it -> {
+
+						ReactiveMongoDatabaseFactory factoryToUse = factory.withSession(it);
+						return StringUtils.hasText(dbName) ? factoryToUse.getMongoDatabase(dbName)
+								: factoryToUse.getMongoDatabase();
+					});
+
+				}).onErrorResume(NoTransactionException.class, e -> Mono.fromSupplier(() -> {
+					return StringUtils.hasText(dbName) ? factory.getMongoDatabase(dbName) : factory.getMongoDatabase();
+				}));
+	}
+
+	private static Mono<ClientSession> doGetSession(TransactionSynchronizationManager synchronizationManager,
+			ReactiveMongoDatabaseFactory dbFactory, SessionSynchronization sessionSynchronization) {
+
+		final ReactiveMongoResourceHolder registeredHolder = (ReactiveMongoResourceHolder) synchronizationManager
+				.getResource(dbFactory);
+
+		// check for native MongoDB transaction
+		if (registeredHolder != null
+				&& (registeredHolder.hasSession() || registeredHolder.isSynchronizedWithTransaction())) {
+
+			return createClientSession(dbFactory).map(session -> {
+
+				if (!registeredHolder.hasSession()) {
+					registeredHolder.setSession(session);
+				}
+
+				return registeredHolder.getSession();
+			});
+		}
+
+		if (SessionSynchronization.ON_ACTUAL_TRANSACTION.equals(sessionSynchronization)) {
+			return Mono.empty();
+		}
+
+		// init a non native MongoDB transaction by registering a MongoSessionSynchronization
+		return createClientSession(dbFactory).map(session -> {
+
+			ReactiveMongoResourceHolder newHolder = new ReactiveMongoResourceHolder(session, dbFactory);
+			newHolder.getRequiredSession().startTransaction();
+
+			synchronizationManager
+					.registerSynchronization(new MongoSessionSynchronization(synchronizationManager, newHolder, dbFactory));
+			newHolder.setSynchronizedWithTransaction(true);
+			synchronizationManager.bindResource(dbFactory, newHolder);
+
+			return newHolder.getSession();
+		});
+	}
+
+	private static Mono<ClientSession> createClientSession(ReactiveMongoDatabaseFactory dbFactory) {
+		return dbFactory.getSession(ClientSessionOptions.builder().causallyConsistent(true).build());
+	}
+
+	/**
+	 * MongoDB specific {@link ResourceHolderSynchronization} for resource cleanup at the end of a transaction when
+	 * participating in a non-native MongoDB transaction, such as a R2CBC transaction.
+	 *
+	 * @author Mark Paluch
+	 * @since 2.2
+	 */
+	private static class MongoSessionSynchronization
+			extends ReactiveResourceSynchronization<ReactiveMongoResourceHolder, Object> {
+
+		private final ReactiveMongoResourceHolder resourceHolder;
+
+		MongoSessionSynchronization(TransactionSynchronizationManager synchronizationManager,
+				ReactiveMongoResourceHolder resourceHolder, ReactiveMongoDatabaseFactory dbFactory) {
+
+			super(resourceHolder, dbFactory, synchronizationManager);
+			this.resourceHolder = resourceHolder;
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.transaction.reactive.ReactiveResourceSynchronization#shouldReleaseBeforeCompletion()
+		 */
+		@Override
+		protected boolean shouldReleaseBeforeCompletion() {
+			return false;
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.transaction.reactive.ReactiveResourceSynchronization#processResourceAfterCommit(java.lang.Object)
+		 */
+		@Override
+		protected Mono<Void> processResourceAfterCommit(ReactiveMongoResourceHolder resourceHolder) {
+
+			if (isTransactionActive(resourceHolder)) {
+				return Mono.from(resourceHolder.getRequiredSession().commitTransaction());
+			}
+
+			return Mono.empty();
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.transaction.reactive.ReactiveResourceSynchronization#afterCompletion(int)
+		 */
+		@Override
+		public Mono<Void> afterCompletion(int status) {
+
+			return Mono.defer(() -> {
+
+				if (status == TransactionSynchronization.STATUS_ROLLED_BACK && isTransactionActive(this.resourceHolder)) {
+					return Mono.from(resourceHolder.getRequiredSession().abortTransaction()).then(super.afterCompletion(status));
+				}
+
+				return super.afterCompletion(status);
+			});
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.transaction.reactive.ReactiveResourceSynchronization#releaseResource(java.lang.Object, java.lang.Object)
+		 */
+		@Override
+		protected Mono<Void> releaseResource(ReactiveMongoResourceHolder resourceHolder, Object resourceKey) {
+
+			return Mono.fromRunnable(() -> {
+				if (resourceHolder.hasActiveSession()) {
+					resourceHolder.getRequiredSession().close();
+				}
+			});
+		}
+
+		private boolean isTransactionActive(ReactiveMongoResourceHolder resourceHolder) {
+
+			if (!resourceHolder.hasSession()) {
+				return false;
+			}
+
+			return resourceHolder.getRequiredSession().hasActiveTransaction();
+		}
+	}
+}
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoResourceHolder.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoResourceHolder.java
new file mode 100644
index 000000000..49bc77570
--- /dev/null
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoResourceHolder.java
@@ -0,0 +1,137 @@
+/*
+ * 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.data.mongodb.core.ReactiveMongoTemplate;
+import org.springframework.lang.Nullable;
+import org.springframework.transaction.support.ResourceHolderSupport;
+
+import com.mongodb.reactivestreams.client.ClientSession;
+
+/**
+ * MongoDB specific resource holder, wrapping a {@link ClientSession}. {@link MongoTransactionManager} binds instances
+ * of this class to the subscriber context.
+ * <p />
+ * <strong>Note:</strong> Intended for internal usage only.
+ *
+ * @author Mark Paluch
+ * @author Christoph Strobl
+ * @since 2.2
+ * @see ReactiveMongoTransactionManager
+ * @see ReactiveMongoTemplate
+ */
+class ReactiveMongoResourceHolder extends ResourceHolderSupport {
+
+	private @Nullable ClientSession session;
+	private ReactiveMongoDatabaseFactory databaseFactory;
+
+	/**
+	 * Create a new {@link ReactiveMongoResourceHolder} for a given {@link ClientSession session}.
+	 *
+	 * @param session the associated {@link ClientSession}. Can be {@literal null}.
+	 * @param databaseFactory the associated {@link MongoDbFactory}. must not be {@literal null}.
+	 */
+	ReactiveMongoResourceHolder(@Nullable ClientSession session, ReactiveMongoDatabaseFactory databaseFactory) {
+
+		this.session = session;
+		this.databaseFactory = databaseFactory;
+	}
+
+	/**
+	 * @return the associated {@link ClientSession}. Can be {@literal null}.
+	 */
+	@Nullable
+	ClientSession getSession() {
+		return session;
+	}
+
+	/**
+	 * @return the required associated {@link ClientSession}.
+	 * @throws IllegalStateException if no session is associated.
+	 */
+	ClientSession getRequiredSession() {
+
+		ClientSession session = getSession();
+
+		if (session == null) {
+			throw new IllegalStateException("No ClientSession associated");
+		}
+		return session;
+	}
+
+	/**
+	 * @return the associated {@link ReactiveMongoDatabaseFactory}.
+	 */
+	public ReactiveMongoDatabaseFactory getDatabaseFactory() {
+		return databaseFactory;
+	}
+
+	/**
+	 * Set the {@link ClientSession} to guard.
+	 *
+	 * @param session can be {@literal null}.
+	 */
+	public void setSession(@Nullable ClientSession session) {
+		this.session = session;
+	}
+
+	/**
+	 * @return {@literal true} if session is not {@literal null}.
+	 */
+	boolean hasSession() {
+		return session != null;
+	}
+
+	/**
+	 * @return {@literal true} if the session is active and has not been closed.
+	 */
+	boolean hasActiveSession() {
+
+		if (!hasSession()) {
+			return false;
+		}
+
+		return hasServerSession() && !getRequiredSession().getServerSession().isClosed();
+	}
+
+	/**
+	 * @return {@literal true} if the session has an active transaction.
+	 * @see #hasActiveSession()
+	 */
+	boolean hasActiveTransaction() {
+
+		if (!hasActiveSession()) {
+			return false;
+		}
+
+		return getRequiredSession().hasActiveTransaction();
+	}
+
+	/**
+	 * @return {@literal true} if the {@link ClientSession} has a {@link com.mongodb.session.ServerSession} associated
+	 *         that is accessible via {@link ClientSession#getServerSession()}.
+	 */
+	boolean hasServerSession() {
+
+		try {
+			return getRequiredSession().getServerSession() != null;
+		} catch (IllegalStateException serverSessionClosed) {
+			// ignore
+		}
+
+		return false;
+	}
+}
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoTransactionManager.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoTransactionManager.java
new file mode 100644
index 000000000..3da00bb48
--- /dev/null
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoTransactionManager.java
@@ -0,0 +1,530 @@
+/*
+ * 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 reactor.core.publisher.Mono;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.lang.Nullable;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionException;
+import org.springframework.transaction.TransactionSystemException;
+import org.springframework.transaction.reactive.AbstractReactiveTransactionManager;
+import org.springframework.transaction.reactive.GenericReactiveTransaction;
+import org.springframework.transaction.reactive.TransactionSynchronizationManager;
+import org.springframework.transaction.support.SmartTransactionObject;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+import com.mongodb.ClientSessionOptions;
+import com.mongodb.MongoException;
+import com.mongodb.TransactionOptions;
+import com.mongodb.reactivestreams.client.ClientSession;
+
+/**
+ * A {@link org.springframework.transaction.ReactiveTransactionManager} implementation that manages
+ * {@link com.mongodb.reactivestreams.client.ClientSession} based transactions for a single
+ * {@link org.springframework.data.mongodb.ReactiveMongoDatabaseFactory}.
+ * <p />
+ * Binds a {@link ClientSession} from the specified
+ * {@link org.springframework.data.mongodb.ReactiveMongoDatabaseFactory} to the subscriber
+ * {@link reactor.util.context.Context}.
+ * <p />
+ * {@link org.springframework.transaction.TransactionDefinition#isReadOnly() Readonly} transactions operate on a
+ * {@link ClientSession} and enable causal consistency, and also {@link ClientSession#startTransaction() start},
+ * {@link com.mongodb.reactivestreams.client.ClientSession#commitTransaction() commit} or
+ * {@link ClientSession#abortTransaction() abort} a transaction.
+ * <p />
+ * Application code is required to retrieve the {@link com.mongodb.reactivestreams.client.MongoDatabase} via
+ * {@link org.springframework.data.mongodb.ReactiveMongoDatabaseUtils#getDatabase(ReactiveMongoDatabaseFactory)} instead
+ * of a standard {@link org.springframework.data.mongodb.ReactiveMongoDatabaseFactory#getMongoDatabase()} call. Spring
+ * classes such as {@link org.springframework.data.mongodb.core.ReactiveMongoTemplate} use this strategy implicitly.
+ * <p />
+ * By default failure of a {@literal commit} operation raises a {@link TransactionSystemException}. You can override
+ * {@link #doCommit(TransactionSynchronizationManager, ReactiveMongoTransactionObject)} to implement the
+ * <a href="https://docs.mongodb.com/manual/core/transactions/#retry-commit-operation">Retry Commit Operation</a>
+ * behavior as outlined in the MongoDB reference manual.
+ *
+ * @author Christoph Strobl
+ * @author Mark Paluch
+ * @since 2.2
+ * @see <a href="https://www.mongodb.com/transactions">MongoDB Transaction Documentation</a>
+ * @see ReactiveMongoDatabaseUtils#getDatabase(ReactiveMongoDatabaseFactory, SessionSynchronization)
+ */
+public class ReactiveMongoTransactionManager extends AbstractReactiveTransactionManager implements InitializingBean {
+
+	private @Nullable ReactiveMongoDatabaseFactory databaseFactory;
+	private @Nullable TransactionOptions options;
+
+	/**
+	 * Create a new {@link ReactiveMongoTransactionManager} for bean-style usage.
+	 * <p />
+	 * <strong>Note:</strong>The {@link org.springframework.data.mongodb.ReactiveMongoDatabaseFactory db factory} has to
+	 * be {@link #setDatabaseFactory(ReactiveMongoDatabaseFactory)} set} before using the instance. Use this constructor
+	 * to prepare a {@link ReactiveMongoTransactionManager} via a {@link org.springframework.beans.factory.BeanFactory}.
+	 * <p />
+	 * Optionally it is possible to set default {@link TransactionOptions transaction options} defining
+	 * {@link com.mongodb.ReadConcern} and {@link com.mongodb.WriteConcern}.
+	 *
+	 * @see #setDatabaseFactory(ReactiveMongoDatabaseFactory)
+	 */
+	public ReactiveMongoTransactionManager() {}
+
+	/**
+	 * Create a new {@link ReactiveMongoTransactionManager} obtaining sessions from the given
+	 * {@link ReactiveMongoDatabaseFactory}.
+	 *
+	 * @param databaseFactory must not be {@literal null}.
+	 */
+	public ReactiveMongoTransactionManager(ReactiveMongoDatabaseFactory databaseFactory) {
+		this(databaseFactory, null);
+	}
+
+	/**
+	 * Create a new {@link ReactiveMongoTransactionManager} obtaining sessions from the given
+	 * {@link ReactiveMongoDatabaseFactory} applying the given {@link TransactionOptions options}, if present, when
+	 * starting a new transaction.
+	 *
+	 * @param databaseFactory must not be {@literal null}.
+	 * @param options can be {@literal null}.
+	 */
+	public ReactiveMongoTransactionManager(ReactiveMongoDatabaseFactory databaseFactory,
+			@Nullable TransactionOptions options) {
+
+		Assert.notNull(databaseFactory, "DbFactory must not be null!");
+
+		this.databaseFactory = databaseFactory;
+		this.options = options;
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doGetTransaction(org.springframework.transaction.reactive.TransactionSynchronizationManager)
+	 */
+	@Override
+	protected Object doGetTransaction(TransactionSynchronizationManager synchronizationManager)
+			throws TransactionException {
+
+		ReactiveMongoResourceHolder resourceHolder = (ReactiveMongoResourceHolder) synchronizationManager
+				.getResource(getRequiredDatabaseFactory());
+		return new ReactiveMongoTransactionObject(resourceHolder);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#isExistingTransaction(java.lang.Object)
+	 */
+	@Override
+	protected boolean isExistingTransaction(Object transaction) throws TransactionException {
+		return extractMongoTransaction(transaction).hasResourceHolder();
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doBegin(org.springframework.transaction.reactive.TransactionSynchronizationManager, java.lang.Object, org.springframework.transaction.TransactionDefinition)
+	 */
+	@Override
+	protected Mono<Void> doBegin(TransactionSynchronizationManager synchronizationManager, Object transaction,
+			TransactionDefinition definition) throws TransactionException {
+
+		return Mono.defer(() -> {
+
+			ReactiveMongoTransactionObject mongoTransactionObject = extractMongoTransaction(transaction);
+
+			Mono<ReactiveMongoResourceHolder> holder = newResourceHolder(definition,
+					ClientSessionOptions.builder().causallyConsistent(true).build());
+
+			return holder.doOnNext(resourceHolder -> {
+
+				mongoTransactionObject.setResourceHolder(resourceHolder);
+
+				if (logger.isDebugEnabled()) {
+					logger.debug(
+							String.format("About to start transaction for session %s.", debugString(resourceHolder.getSession())));
+				}
+
+			}).doOnNext(resourceHolder -> {
+
+				mongoTransactionObject.startTransaction(options);
+
+				if (logger.isDebugEnabled()) {
+					logger.debug(String.format("Started transaction for session %s.", debugString(resourceHolder.getSession())));
+				}
+
+			}).onErrorMap(
+					ex -> new TransactionSystemException(String.format("Could not start Mongo transaction for session %s.",
+							debugString(mongoTransactionObject.getSession())), ex))
+					.doOnSuccess(resourceHolder -> {
+
+						synchronizationManager.bindResource(getRequiredDatabaseFactory(), resourceHolder);
+					}).then();
+		});
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doSuspend(org.springframework.transaction.reactive.TransactionSynchronizationManager, java.lang.Object)
+	 */
+	@Override
+	protected Mono<Object> doSuspend(TransactionSynchronizationManager synchronizationManager, Object transaction)
+			throws TransactionException {
+
+		return Mono.fromSupplier(() -> {
+
+			ReactiveMongoTransactionObject mongoTransactionObject = extractMongoTransaction(transaction);
+			mongoTransactionObject.setResourceHolder(null);
+
+			return synchronizationManager.unbindResource(getRequiredDatabaseFactory());
+		});
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doResume(org.springframework.transaction.reactive.TransactionSynchronizationManager, java.lang.Object, java.lang.Object)
+	 */
+	@Override
+	protected Mono<Void> doResume(TransactionSynchronizationManager synchronizationManager, @Nullable Object transaction,
+			Object suspendedResources) {
+		return Mono
+				.fromRunnable(() -> synchronizationManager.bindResource(getRequiredDatabaseFactory(), suspendedResources));
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doCommit(org.springframework.transaction.reactive.TransactionSynchronizationManager, org.springframework.transaction.reactive.GenericReactiveTransaction)
+	 */
+	@Override
+	protected final Mono<Void> doCommit(TransactionSynchronizationManager synchronizationManager,
+			GenericReactiveTransaction status) throws TransactionException {
+
+		return Mono.defer(() -> {
+
+			ReactiveMongoTransactionObject mongoTransactionObject = extractMongoTransaction(status);
+
+			if (logger.isDebugEnabled()) {
+				logger.debug(String.format("About to commit transaction for session %s.",
+						debugString(mongoTransactionObject.getSession())));
+			}
+
+			return doCommit(synchronizationManager, mongoTransactionObject).onErrorMap(ex -> {
+				return new TransactionSystemException(String.format("Could not commit Mongo transaction for session %s.",
+						debugString(mongoTransactionObject.getSession())), ex);
+			});
+		});
+	}
+
+	/**
+	 * Customization hook to perform an actual commit of the given transaction.<br />
+	 * If a commit operation encounters an error, the MongoDB driver throws a {@link MongoException} holding
+	 * {@literal error labels}. <br />
+	 * By default those labels are ignored, nevertheless one might check for
+	 * {@link MongoException#UNKNOWN_TRANSACTION_COMMIT_RESULT_LABEL transient commit errors labels} and retry the the
+	 * commit.
+	 *
+	 * @param synchronizationManager reactive synchronization manager.
+	 * @param transactionObject never {@literal null}.
+	 */
+	protected Mono<Void> doCommit(TransactionSynchronizationManager synchronizationManager,
+			ReactiveMongoTransactionObject transactionObject) {
+		return transactionObject.commitTransaction();
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doRollback(org.springframework.transaction.reactive.TransactionSynchronizationManager, org.springframework.transaction.reactive.GenericReactiveTransaction)
+	 */
+	@Override
+	protected Mono<Void> doRollback(TransactionSynchronizationManager synchronizationManager,
+			GenericReactiveTransaction status) {
+
+		return Mono.defer(() -> {
+
+			ReactiveMongoTransactionObject mongoTransactionObject = extractMongoTransaction(status);
+
+			if (logger.isDebugEnabled()) {
+				logger.debug(String.format("About to abort transaction for session %s.",
+						debugString(mongoTransactionObject.getSession())));
+			}
+
+			return mongoTransactionObject.abortTransaction().onErrorResume(MongoException.class, ex -> {
+				return Mono
+						.error(new TransactionSystemException(String.format("Could not abort Mongo transaction for session %s.",
+								debugString(mongoTransactionObject.getSession())), ex));
+			});
+		});
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doSetRollbackOnly(org.springframework.transaction.reactive.TransactionSynchronizationManager, org.springframework.transaction.reactive.GenericReactiveTransaction)
+	 */
+	@Override
+	protected Mono<Void> doSetRollbackOnly(TransactionSynchronizationManager synchronizationManager,
+			GenericReactiveTransaction status) throws TransactionException {
+
+		return Mono.fromRunnable(() -> {
+			ReactiveMongoTransactionObject transactionObject = extractMongoTransaction(status);
+			transactionObject.getRequiredResourceHolder().setRollbackOnly();
+		});
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doCleanupAfterCompletion(org.springframework.transaction.reactive.TransactionSynchronizationManager, java.lang.Object)
+	 */
+	@Override
+	protected Mono<Void> doCleanupAfterCompletion(TransactionSynchronizationManager synchronizationManager,
+			Object transaction) {
+
+		Assert.isInstanceOf(ReactiveMongoTransactionObject.class, transaction,
+				() -> String.format("Expected to find a %s but it turned out to be %s.", ReactiveMongoTransactionObject.class,
+						transaction.getClass()));
+
+		return Mono.fromRunnable(() -> {
+			ReactiveMongoTransactionObject mongoTransactionObject = (ReactiveMongoTransactionObject) transaction;
+
+			// Remove the connection holder from the thread.
+			synchronizationManager.unbindResource(getRequiredDatabaseFactory());
+			mongoTransactionObject.getRequiredResourceHolder().clear();
+
+			if (logger.isDebugEnabled()) {
+				logger.debug(String.format("About to release Session %s after transaction.",
+						debugString(mongoTransactionObject.getSession())));
+			}
+
+			mongoTransactionObject.closeSession();
+		});
+	}
+
+	/**
+	 * Set the {@link ReactiveMongoDatabaseFactory} that this instance should manage transactions for.
+	 *
+	 * @param databaseFactory must not be {@literal null}.
+	 */
+	public void setDatabaseFactory(ReactiveMongoDatabaseFactory databaseFactory) {
+
+		Assert.notNull(databaseFactory, "DbFactory must not be null!");
+		this.databaseFactory = databaseFactory;
+	}
+
+	/**
+	 * Set the {@link TransactionOptions} to be applied when starting transactions.
+	 *
+	 * @param options can be {@literal null}.
+	 */
+	public void setOptions(@Nullable TransactionOptions options) {
+		this.options = options;
+	}
+
+	/**
+	 * Get the {@link ReactiveMongoDatabaseFactory} that this instance manages transactions for.
+	 *
+	 * @return can be {@literal null}.
+	 */
+	@Nullable
+	public ReactiveMongoDatabaseFactory getDatabaseFactory() {
+		return databaseFactory;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
+	 */
+	@Override
+	public void afterPropertiesSet() {
+		getRequiredDatabaseFactory();
+	}
+
+	private Mono<ReactiveMongoResourceHolder> newResourceHolder(TransactionDefinition definition,
+			ClientSessionOptions options) {
+
+		ReactiveMongoDatabaseFactory dbFactory = getRequiredDatabaseFactory();
+
+		return dbFactory.getSession(options).map(session -> new ReactiveMongoResourceHolder(session, dbFactory));
+	}
+
+	/**
+	 * @throws IllegalStateException if {@link #databaseFactory} is {@literal null}.
+	 */
+	private ReactiveMongoDatabaseFactory getRequiredDatabaseFactory() {
+
+		Assert.state(databaseFactory != null,
+				"MongoTransactionManager operates upon a ReactiveMongoDatabaseFactory. Did you forget to provide one? It's required.");
+
+		return databaseFactory;
+	}
+
+	private static ReactiveMongoTransactionObject extractMongoTransaction(Object transaction) {
+
+		Assert.isInstanceOf(ReactiveMongoTransactionObject.class, transaction,
+				() -> String.format("Expected to find a %s but it turned out to be %s.", ReactiveMongoTransactionObject.class,
+						transaction.getClass()));
+
+		return (ReactiveMongoTransactionObject) transaction;
+	}
+
+	private static ReactiveMongoTransactionObject extractMongoTransaction(GenericReactiveTransaction status) {
+
+		Assert.isInstanceOf(ReactiveMongoTransactionObject.class, status.getTransaction(),
+				() -> String.format("Expected to find a %s but it turned out to be %s.", ReactiveMongoTransactionObject.class,
+						status.getTransaction().getClass()));
+
+		return (ReactiveMongoTransactionObject) status.getTransaction();
+	}
+
+	private static String debugString(@Nullable ClientSession session) {
+
+		if (session == null) {
+			return "null";
+		}
+
+		String debugString = String.format("[%s@%s ", ClassUtils.getShortName(session.getClass()),
+				Integer.toHexString(session.hashCode()));
+
+		try {
+			if (session.getServerSession() != null) {
+				debugString += String.format("id = %s, ", session.getServerSession().getIdentifier());
+				debugString += String.format("causallyConsistent = %s, ", session.isCausallyConsistent());
+				debugString += String.format("txActive = %s, ", session.hasActiveTransaction());
+				debugString += String.format("txNumber = %d, ", session.getServerSession().getTransactionNumber());
+				debugString += String.format("closed = %d, ", session.getServerSession().isClosed());
+				debugString += String.format("clusterTime = %s", session.getClusterTime());
+			} else {
+				debugString += "id = n/a";
+				debugString += String.format("causallyConsistent = %s, ", session.isCausallyConsistent());
+				debugString += String.format("txActive = %s, ", session.hasActiveTransaction());
+				debugString += String.format("clusterTime = %s", session.getClusterTime());
+			}
+		} catch (RuntimeException e) {
+			debugString += String.format("error = %s", e.getMessage());
+		}
+
+		debugString += "]";
+
+		return debugString;
+	}
+
+	/**
+	 * MongoDB specific transaction object, representing a {@link MongoResourceHolder}. Used as transaction object by
+	 * {@link ReactiveMongoTransactionManager}.
+	 *
+	 * @author Christoph Strobl
+	 * @author Mark Paluch
+	 * @since 2.2
+	 * @see ReactiveMongoResourceHolder
+	 */
+	protected static class ReactiveMongoTransactionObject implements SmartTransactionObject {
+
+		private @Nullable ReactiveMongoResourceHolder resourceHolder;
+
+		ReactiveMongoTransactionObject(@Nullable ReactiveMongoResourceHolder resourceHolder) {
+			this.resourceHolder = resourceHolder;
+		}
+
+		/**
+		 * Set the {@link MongoResourceHolder}.
+		 *
+		 * @param resourceHolder can be {@literal null}.
+		 */
+		void setResourceHolder(@Nullable ReactiveMongoResourceHolder resourceHolder) {
+			this.resourceHolder = resourceHolder;
+		}
+
+		/**
+		 * @return {@literal true} if a {@link MongoResourceHolder} is set.
+		 */
+		final boolean hasResourceHolder() {
+			return resourceHolder != null;
+		}
+
+		/**
+		 * Start a MongoDB transaction optionally given {@link TransactionOptions}.
+		 *
+		 * @param options can be {@literal null}
+		 */
+		void startTransaction(@Nullable TransactionOptions options) {
+
+			ClientSession session = getRequiredSession();
+			if (options != null) {
+				session.startTransaction(options);
+			} else {
+				session.startTransaction();
+			}
+		}
+
+		/**
+		 * Commit the transaction.
+		 */
+		public Mono<Void> commitTransaction() {
+			return Mono.from(getRequiredSession().commitTransaction());
+		}
+
+		/**
+		 * Rollback (abort) the transaction.
+		 */
+		public Mono<Void> abortTransaction() {
+			return Mono.from(getRequiredSession().abortTransaction());
+		}
+
+		/**
+		 * Close a {@link ClientSession} without regard to its transactional state.
+		 */
+		void closeSession() {
+
+			ClientSession session = getRequiredSession();
+			if (session.getServerSession() != null && !session.getServerSession().isClosed()) {
+				session.close();
+			}
+		}
+
+		@Nullable
+		public ClientSession getSession() {
+			return resourceHolder != null ? resourceHolder.getSession() : null;
+		}
+
+		private ReactiveMongoResourceHolder getRequiredResourceHolder() {
+
+			Assert.state(resourceHolder != null, "ReactiveMongoResourceHolder is required but not present. o_O");
+			return resourceHolder;
+		}
+
+		private ClientSession getRequiredSession() {
+
+			ClientSession session = getSession();
+			Assert.state(session != null, "A Session is required but it turned out to be null.");
+			return session;
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.transaction.support.SmartTransactionObject#isRollbackOnly()
+		 */
+		@Override
+		public boolean isRollbackOnly() {
+			return this.resourceHolder != null && this.resourceHolder.isRollbackOnly();
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.transaction.support.SmartTransactionObject#flush()
+		 */
+		@Override
+		public void flush() {
+			throw new UnsupportedOperationException("flush() not supported");
+		}
+
+	}
+}
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java
index de27e8297..eb99b0eb6 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java
@@ -41,6 +41,7 @@ import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
 import org.springframework.lang.Nullable;
+import org.springframework.transaction.reactive.TransactionalOperator;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 
@@ -217,7 +218,9 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
 	 * {@link ClientSession#abortTransaction() rolled back} upon errors.
 	 *
 	 * @return new instance of {@link ReactiveSessionScoped}. Never {@literal null}.
+	 * @deprecated since 2.2. Use {@code @Transactional} or {@link TransactionalOperator}.
 	 */
+	@Deprecated
 	ReactiveSessionScoped inTransaction();
 
 	/**
@@ -232,7 +235,9 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
 	 * @param sessionProvider must not be {@literal null}.
 	 * @return new instance of {@link ReactiveSessionScoped}. Never {@literal null}.
 	 * @since 2.1
+	 * @deprecated since 2.2. Use {@code @Transactional} or {@link TransactionalOperator}.
 	 */
+	@Deprecated
 	ReactiveSessionScoped inTransaction(Publisher<ClientSession> sessionProvider);
 
 	/**
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java
index 1eb878ab9..d0c0592b5 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java
@@ -63,6 +63,8 @@ import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.context.MappingContextEvent;
 import org.springframework.data.mongodb.MongoDbFactory;
 import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
+import org.springframework.data.mongodb.ReactiveMongoDatabaseUtils;
+import org.springframework.data.mongodb.SessionSynchronization;
 import org.springframework.data.mongodb.core.EntityOperations.AdaptibleEntity;
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
@@ -198,6 +200,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 	private @Nullable ApplicationEventPublisher eventPublisher;
 	private @Nullable ReactiveMongoPersistentEntityIndexCreator indexCreator;
 
+	private SessionSynchronization sessionSynchronization = SessionSynchronization.ON_ACTUAL_TRANSACTION;
+
 	/**
 	 * Constructor used for a basic template configuration.
 	 *
@@ -287,6 +291,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		this.mappingContext = that.mappingContext;
 		this.operations = that.operations;
 		this.propertyOperations = that.propertyOperations;
+		this.sessionSynchronization = that.sessionSynchronization;
 	}
 
 	private void onCheckForIndexes(MongoPersistentEntity<?> entity, Consumer<Throwable> subscriptionExceptionHandler) {
@@ -498,6 +503,17 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		};
 	}
 
+	/**
+	 * Define if {@link ReactiveMongoTemplate} should participate in transactions. Default is set to
+	 * {@link SessionSynchronization#ON_ACTUAL_TRANSACTION}.<br />
+	 * <strong>NOTE:</strong> MongoDB transactions require at least MongoDB 4.0.
+	 *
+	 * @since 2.2
+	 */
+	public void setSessionSynchronization(SessionSynchronization sessionSynchronization) {
+		this.sessionSynchronization = sessionSynchronization;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#inTransaction()
@@ -575,7 +591,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 
 		Assert.notNull(callback, "ReactiveDatabaseCallback must not be null!");
 
-		return Flux.defer(() -> callback.doInDB(prepareDatabase(doGetDatabase()))).onErrorMap(translateException());
+		return Mono.defer(this::doGetDatabase).flatMapMany(database -> callback.doInDB(prepareDatabase(database)))
+				.onErrorMap(translateException());
 	}
 
 	/**
@@ -589,7 +606,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 
 		Assert.notNull(callback, "ReactiveDatabaseCallback must not be null!");
 
-		return Mono.defer(() -> Mono.from(callback.doInDB(prepareDatabase(doGetDatabase()))))
+		return Mono.defer(this::doGetDatabase).flatMap(database -> Mono.from(callback.doInDB(prepareDatabase(database))))
 				.onErrorMap(translateException());
 	}
 
@@ -605,8 +622,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		Assert.hasText(collectionName, "Collection name must not be null or empty!");
 		Assert.notNull(callback, "ReactiveDatabaseCallback must not be null!");
 
-		Mono<MongoCollection<Document>> collectionPublisher = Mono
-				.fromCallable(() -> getAndPrepareCollection(doGetDatabase(), collectionName));
+		Mono<MongoCollection<Document>> collectionPublisher = doGetDatabase()
+				.map(database -> getAndPrepareCollection(database, collectionName));
 
 		return collectionPublisher.flatMapMany(callback::doInCollection).onErrorMap(translateException());
 	}
@@ -624,8 +641,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		Assert.hasText(collectionName, "Collection name must not be null or empty!");
 		Assert.notNull(callback, "ReactiveCollectionCallback must not be null!");
 
-		Mono<MongoCollection<Document>> collectionPublisher = Mono
-				.fromCallable(() -> getAndPrepareCollection(doGetDatabase(), collectionName));
+		Mono<MongoCollection<Document>> collectionPublisher = doGetDatabase()
+				.map(database -> getAndPrepareCollection(database, collectionName));
 
 		return collectionPublisher.flatMap(collection -> Mono.from(callback.doInCollection(collection)))
 				.onErrorMap(translateException());
@@ -680,7 +697,25 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 	 * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#getCollection(java.lang.String)
 	 */
 	public MongoCollection<Document> getCollection(String collectionName) {
-		return execute((MongoDatabaseCallback<MongoCollection<Document>>) db -> db.getCollection(collectionName));
+
+		Assert.notNull(collectionName, "Collection name must not be null!");
+
+		try {
+			return this.mongoDatabaseFactory.getMongoDatabase().getCollection(collectionName);
+		} catch (RuntimeException e) {
+			throw potentiallyConvertRuntimeException(e, exceptionTranslator);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#getCollection(java.lang.String)
+	 */
+	public Mono<MongoCollection<Document>> getCollection2(final String collectionName) {
+
+		Assert.notNull(collectionName, "Collection name must not be null!");
+
+		return doGetDatabase().map(it -> it.getCollection(collectionName));
 	}
 
 	/*
@@ -732,11 +767,11 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 	}
 
 	public MongoDatabase getMongoDatabase() {
-		return doGetDatabase();
+		return mongoDatabaseFactory.getMongoDatabase();
 	}
 
-	protected MongoDatabase doGetDatabase() {
-		return mongoDatabaseFactory.getMongoDatabase();
+	protected Mono<MongoDatabase> doGetDatabase() {
+		return ReactiveMongoDatabaseUtils.getDatabase(mongoDatabaseFactory, sessionSynchronization);
 	}
 
 	/*
@@ -2590,18 +2625,6 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		});
 	}
 
-	private <T> T execute(MongoDatabaseCallback<T> action) {
-
-		Assert.notNull(action, "MongoDatabaseCallback must not be null!");
-
-		try {
-			MongoDatabase db = this.doGetDatabase();
-			return action.doInDatabase(db);
-		} catch (RuntimeException e) {
-			throw potentiallyConvertRuntimeException(e, exceptionTranslator);
-		}
-	}
-
 	/**
 	 * Exception translation {@link Function} intended for {@link Flux#onErrorMap(Function)} usage.
 	 *
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactory.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactory.java
index 15dcaa1a5..f7effcf38 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactory.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactory.java
@@ -211,6 +211,15 @@ public class SimpleReactiveMongoDatabaseFactory implements DisposableBean, React
 			return delegate.withSession(session);
 		}
 
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.ReactiveMongoDatabaseFactory#isTransactionActive()
+		 */
+		@Override
+		public boolean isTransactionActive() {
+			return session != null && session.hasActiveTransaction();
+		}
+
 		private MongoDatabase decorateDatabase(MongoDatabase database) {
 			return createProxyInstance(session, database, MongoDatabase.class);
 		}
diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/ReactiveMongoDatabaseUtilsUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/ReactiveMongoDatabaseUtilsUnitTests.java
new file mode 100644
index 000000000..c23fb17bf
--- /dev/null
+++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/ReactiveMongoDatabaseUtilsUnitTests.java
@@ -0,0 +1,135 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 static org.assertj.core.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import org.springframework.transaction.reactive.TransactionSynchronizationManager;
+import org.springframework.transaction.reactive.TransactionalOperator;
+import org.springframework.transaction.support.DefaultTransactionDefinition;
+
+import com.mongodb.reactivestreams.client.ClientSession;
+import com.mongodb.reactivestreams.client.MongoDatabase;
+import com.mongodb.session.ServerSession;
+
+/**
+ * Unit tests for {@link ReactiveMongoDatabaseUtils}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class ReactiveMongoDatabaseUtilsUnitTests {
+
+	@Mock ClientSession session;
+	@Mock ServerSession serverSession;
+	@Mock ReactiveMongoDatabaseFactory databaseFactory;
+	@Mock MongoDatabase db;
+
+	@Before
+	public void setUp() {
+
+		when(databaseFactory.getSession(any())).thenReturn(Mono.just(session));
+		when(databaseFactory.getMongoDatabase()).thenReturn(db);
+
+		when(session.getServerSession()).thenReturn(serverSession);
+		when(session.hasActiveTransaction()).thenReturn(true);
+	}
+
+	@Test // DATAMONGO-2265
+	public void isTransactionActiveShouldDetectTxViaFactory() {
+
+		when(databaseFactory.isTransactionActive()).thenReturn(true);
+
+		ReactiveMongoDatabaseUtils.isTransactionActive(databaseFactory) //
+				.as(StepVerifier::create) //
+				.expectNext(true).verifyComplete();
+	}
+
+	@Test // DATAMONGO-2265
+	public void isTransactionActiveShouldReturnFalseIfNoTxActive() {
+
+		when(databaseFactory.isTransactionActive()).thenReturn(false);
+
+		ReactiveMongoDatabaseUtils.isTransactionActive(databaseFactory) //
+				.as(StepVerifier::create) //
+				.expectNext(false).verifyComplete();
+	}
+
+	@Test // DATAMONGO-2265
+	public void isTransactionActiveShouldLookupTxForActiveTransactionSynchronizationViaTxManager() {
+
+		when(databaseFactory.isTransactionActive()).thenReturn(false);
+		when(session.commitTransaction()).thenReturn(Mono.empty());
+
+		ReactiveMongoTransactionManager txManager = new ReactiveMongoTransactionManager(databaseFactory);
+		TransactionalOperator operator = TransactionalOperator.create(txManager, new DefaultTransactionDefinition());
+
+		operator.execute(tx -> {
+
+			return ReactiveMongoDatabaseUtils.isTransactionActive(databaseFactory);
+		}).as(StepVerifier::create).expectNext(true).verifyComplete();
+	}
+
+	@Test // DATAMONGO-2265
+	public void shouldNotStartSessionWhenNoTransactionOngoing() {
+
+		ReactiveMongoDatabaseUtils.getDatabase(databaseFactory, SessionSynchronization.ON_ACTUAL_TRANSACTION) //
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+
+		verify(databaseFactory, never()).getSession(any());
+		verify(databaseFactory, never()).withSession(any(ClientSession.class));
+	}
+
+	@Test // DATAMONGO-2265
+	public void shouldParticipateInOngoingMongoTransactionWhenSessionSychronizationIsNative() {
+
+		ReactiveMongoTransactionManager txManager = new ReactiveMongoTransactionManager(databaseFactory);
+		when(session.abortTransaction()).thenReturn(Mono.empty());
+
+		TransactionalOperator operator = TransactionalOperator.create(txManager, new DefaultTransactionDefinition());
+
+		operator.execute(tx -> {
+
+			return TransactionSynchronizationManager.currentTransaction().doOnNext(synchronizationManager -> {
+
+				assertThat(synchronizationManager.isSynchronizationActive()).isTrue();
+				assertThat(tx.isNewTransaction()).isTrue();
+
+				assertThat(synchronizationManager.hasResource(databaseFactory)).isTrue();
+
+			}).then(Mono.fromRunnable(tx::setRollbackOnly));
+		}).as(StepVerifier::create).verifyComplete();
+
+		verify(session).startTransaction();
+		verify(session).abortTransaction();
+
+		// TODO: Bug in doCleanupAfterCompletion
+		// verify(session).close();
+	}
+}
diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/ReactiveMongoTransactionManagerUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/ReactiveMongoTransactionManagerUnitTests.java
new file mode 100644
index 000000000..a5d16933a
--- /dev/null
+++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/ReactiveMongoTransactionManagerUnitTests.java
@@ -0,0 +1,271 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.reactive.TransactionalOperator;
+import org.springframework.transaction.support.DefaultTransactionDefinition;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+import com.mongodb.reactivestreams.client.ClientSession;
+import com.mongodb.reactivestreams.client.MongoDatabase;
+import com.mongodb.session.ServerSession;
+
+/**
+ * Unit tests for {@link ReactiveMongoTransactionManager}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class ReactiveMongoTransactionManagerUnitTests {
+
+	@Mock ClientSession session;
+	@Mock ClientSession session2;
+	@Mock ServerSession serverSession;
+	@Mock ReactiveMongoDatabaseFactory databaseFactory;
+	@Mock ReactiveMongoDatabaseFactory databaseFactory2;
+	@Mock MongoDatabase db;
+	@Mock MongoDatabase db2;
+
+	@Before
+	public void setUp() {
+
+		when(databaseFactory.getSession(any())).thenReturn(Mono.just(session), Mono.just(session2));
+
+		when(databaseFactory.withSession(session)).thenReturn(databaseFactory);
+		when(databaseFactory.withSession(session2)).thenReturn(databaseFactory2);
+
+		when(databaseFactory.getMongoDatabase()).thenReturn(db);
+		when(databaseFactory2.getMongoDatabase()).thenReturn(db2);
+	}
+
+	@After
+	public void verifyTransactionSynchronizationManager() {
+
+		assertTrue(TransactionSynchronizationManager.getResourceMap().isEmpty());
+		assertFalse(TransactionSynchronizationManager.isSynchronizationActive());
+	}
+
+	@Test // DATAMONGO-2265
+	public void triggerCommitCorrectly() {
+
+		ReactiveMongoTransactionManager txManager = new ReactiveMongoTransactionManager(databaseFactory);
+		ReactiveMongoTemplate template = new ReactiveMongoTemplate(databaseFactory);
+		when(session.commitTransaction()).thenReturn(Mono.empty());
+
+		TransactionalOperator operator = TransactionalOperator.create(txManager, new DefaultTransactionDefinition());
+
+		template.execute(db -> {
+			db.drop();
+			return Mono.empty();
+
+		}).as(operator::transactional) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		verify(databaseFactory).withSession(eq(session));
+
+		verify(session).startTransaction();
+		verify(session).commitTransaction();
+
+		// TODO: Bug in doCleanupAfterCompletion
+		// verify(session).close();
+	}
+
+	@Test // DATAMONGO-2265
+	public void participateInOnGoingTransactionWithCommit() {
+
+		ReactiveMongoTransactionManager txManager = new ReactiveMongoTransactionManager(databaseFactory);
+		ReactiveMongoTemplate template = new ReactiveMongoTemplate(databaseFactory);
+		when(session.commitTransaction()).thenReturn(Mono.empty());
+
+		TransactionalOperator operator = TransactionalOperator.create(txManager, new DefaultTransactionDefinition());
+
+		template.execute(db -> {
+			db.drop();
+			return Mono.empty();
+		}).as(StepVerifier::create).verifyComplete();
+
+		template.execute(db -> {
+			db.drop();
+			return Mono.empty();
+		}).as(operator::transactional) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		verify(databaseFactory, times(1)).withSession(eq(session));
+
+		verify(session).startTransaction();
+		verify(session).commitTransaction();
+
+		// TODO: Bug in doCleanupAfterCompletion
+		// verify(session).close();
+	}
+
+	@Test // DATAMONGO-2265
+	public void participateInOnGoingTransactionWithRollbackOnly() {
+
+		ReactiveMongoTransactionManager txManager = new ReactiveMongoTransactionManager(databaseFactory);
+		ReactiveMongoTemplate template = new ReactiveMongoTemplate(databaseFactory);
+		when(session.abortTransaction()).thenReturn(Mono.empty());
+
+		TransactionalOperator operator = TransactionalOperator.create(txManager, new DefaultTransactionDefinition());
+
+		operator.execute(tx -> {
+
+			return template.execute(db -> {
+				db.drop();
+				tx.setRollbackOnly();
+				return Mono.empty();
+			});
+		}).as(StepVerifier::create).verifyComplete();
+
+		verify(databaseFactory, times(1)).withSession(eq(session));
+
+		verify(session).startTransaction();
+		verify(session).abortTransaction();
+
+		// TODO: Bug in doCleanupAfterCompletion
+		// verify(session).close();
+	}
+
+	@Test // DATAMONGO-2265
+	public void suspendTransactionWhilePropagationNotSupported() {
+
+		ReactiveMongoTransactionManager txManager = new ReactiveMongoTransactionManager(databaseFactory);
+		ReactiveMongoTemplate template = new ReactiveMongoTemplate(databaseFactory);
+		when(session.commitTransaction()).thenReturn(Mono.empty());
+
+		TransactionalOperator outer = TransactionalOperator.create(txManager, new DefaultTransactionDefinition());
+
+		DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
+		definition.setPropagationBehavior(TransactionDefinition.PROPAGATION_NOT_SUPPORTED);
+		TransactionalOperator inner = TransactionalOperator.create(txManager, definition);
+
+		outer.execute(tx1 -> {
+
+			return template.execute(db -> {
+
+				db.drop();
+
+				return inner.execute(tx2 -> {
+					return template.execute(db2 -> {
+						db2.drop();
+						return Mono.empty();
+					});
+				});
+			});
+		}).as(StepVerifier::create).verifyComplete();
+
+		verify(session).startTransaction();
+		verify(session2, never()).startTransaction();
+
+		verify(databaseFactory, times(1)).withSession(eq(session));
+		verify(databaseFactory, never()).withSession(eq(session2));
+
+		// Bug in TransactionalOperator, should be 2
+		verify(db, times(1)).drop();
+
+		// TODO: Bug in doCleanupAfterCompletion
+		// verify(session).close();
+		verify(session2, never()).close();
+	}
+
+	@Test // DATAMONGO-2265
+	public void suspendTransactionWhilePropagationRequiresNew() {
+
+		ReactiveMongoTransactionManager txManager = new ReactiveMongoTransactionManager(databaseFactory);
+		ReactiveMongoTemplate template = new ReactiveMongoTemplate(databaseFactory);
+		when(session.commitTransaction()).thenReturn(Mono.empty());
+		when(session2.commitTransaction()).thenReturn(Mono.empty());
+
+		TransactionalOperator outer = TransactionalOperator.create(txManager, new DefaultTransactionDefinition());
+
+		DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
+		definition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
+		TransactionalOperator inner = TransactionalOperator.create(txManager, definition);
+
+		outer.execute(tx1 -> {
+
+			return template.execute(db -> {
+
+				db.drop();
+
+				return inner.execute(tx2 -> {
+					return template.execute(db2 -> {
+						db2.drop();
+						return Mono.empty();
+					});
+				});
+			});
+		}).as(StepVerifier::create).verifyComplete();
+
+		verify(session).startTransaction();
+		verify(session2).startTransaction();
+
+		verify(databaseFactory, times(1)).withSession(eq(session));
+		verify(databaseFactory).withSession(eq(session2));
+
+		verify(db).drop();
+		verify(db2).drop();
+
+		// TODO: Bug in doCleanupAfterCompletion
+		// verify(session).close();
+		// verify(session2).close();
+	}
+
+	@Test // DATAMONGO-2265
+	public void readonlyShouldInitiateASessionStartAndCommitTransaction() {
+
+		ReactiveMongoTransactionManager txManager = new ReactiveMongoTransactionManager(databaseFactory);
+		ReactiveMongoTemplate template = new ReactiveMongoTemplate(databaseFactory);
+		when(session.commitTransaction()).thenReturn(Mono.empty());
+
+		DefaultTransactionDefinition readonlyTxDefinition = new DefaultTransactionDefinition();
+		readonlyTxDefinition.setReadOnly(true);
+		TransactionalOperator operator = TransactionalOperator.create(txManager, readonlyTxDefinition);
+
+		template.execute(db -> {
+			db.drop();
+			return Mono.empty();
+
+		}).as(operator::transactional) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		verify(databaseFactory).withSession(eq(session));
+
+		verify(session).startTransaction();
+		verify(session).commitTransaction();
+
+		// TODO: Bug in doCleanupAfterCompletion
+		// verify(session).close();
+	}
+}
diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/support/ReactiveQuerydslMongoPredicateExecutorTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/support/ReactiveQuerydslMongoPredicateExecutorTests.java
index 47ca7c549..f67045c01 100644
--- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/support/ReactiveQuerydslMongoPredicateExecutorTests.java
+++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/support/ReactiveQuerydslMongoPredicateExecutorTests.java
@@ -17,6 +17,7 @@ package org.springframework.data.mongodb.repository.support;
 
 import org.springframework.data.mongodb.core.query.BasicQuery;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import java.util.Arrays;
@@ -27,6 +28,7 @@ import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
@@ -303,14 +305,14 @@ public class ReactiveQuerydslMongoPredicateExecutorTests {
 				.verifyComplete();
 	}
 
-	@Test // DATAMONGO-2182
+	@Test // DATAMONGO-2182, DATAMONGO-2265
 	public void translatesExceptionsCorrectly() {
 
 		ReactiveMongoOperations ops = new ReactiveMongoTemplate(dbFactory) {
 
 			@Override
-			protected MongoDatabase doGetDatabase() {
-				throw new MongoException(18, "Authentication Failed");
+			protected Mono<MongoDatabase> doGetDatabase() {
+				return Mono.error(new MongoException(18, "Authentication Failed"));
 			}
 		};
 
diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/rxtx/ReactiveTransactionIntegrationTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/rxtx/ReactiveTransactionIntegrationTests.java
new file mode 100644
index 000000000..e192a03a2
--- /dev/null
+++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/rxtx/ReactiveTransactionIntegrationTests.java
@@ -0,0 +1,128 @@
+/*
+ * 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.rxtx;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.RequiredArgsConstructor;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import org.bson.types.ObjectId;
+import org.junit.Test;
+
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
+import org.springframework.data.mongodb.ReactiveMongoTransactionManager;
+import org.springframework.data.mongodb.config.AbstractReactiveMongoConfiguration;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.ReactiveMongoOperations;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.reactive.TransactionalOperator;
+import org.springframework.transaction.support.DefaultTransactionDefinition;
+
+import com.mongodb.reactivestreams.client.MongoClient;
+import com.mongodb.reactivestreams.client.MongoClients;
+
+/**
+ * Integration tests for reactive transaction management.
+ *
+ * @author Mark Paluch
+ */
+public class ReactiveTransactionIntegrationTests {
+
+	@Configuration
+	static class TestMongoConfig extends AbstractReactiveMongoConfiguration {
+
+		@Override
+		public MongoClient reactiveMongoClient() {
+			return MongoClients.create("mongodb://localhost");
+		}
+
+		@Override
+		protected String getDatabaseName() {
+			return "test";
+		}
+
+		@Bean
+		public ReactiveMongoTransactionManager transactionManager(ReactiveMongoDatabaseFactory factory) {
+			return new ReactiveMongoTransactionManager(factory);
+		}
+	}
+
+	@Service
+	@RequiredArgsConstructor
+	static class PersonService {
+
+		final ReactiveMongoOperations operations;
+		final ReactiveMongoTransactionManager manager;
+
+		public Mono<Person> savePerson(Person person) {
+
+			TransactionalOperator transactionalOperator = TransactionalOperator.create(manager,
+					new DefaultTransactionDefinition());
+
+			return operations.save(person).<Person> flatMap(it -> {
+				return Mono.error(new RuntimeException("poof!"));
+			}).as(transactionalOperator::transactional);
+		}
+	}
+
+	@Test
+	public void shouldRollbackAfterException() {
+
+		AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(TestMongoConfig.class,
+				PersonService.class);
+
+		ReactiveMongoOperations operations = context.getBean(ReactiveMongoOperations.class);
+
+		MongoTemplate template = new MongoTemplate(new com.mongodb.MongoClient("localhost"), "test");
+
+		template.dropCollection(Person.class);
+		template.dropCollection(EventLog.class);
+
+		template.createCollection(Person.class);
+		template.createCollection(EventLog.class);
+
+		PersonService personService = context.getBean(PersonService.class);
+
+		personService.savePerson(new Person(null, "Walter", "White")) //
+				.as(StepVerifier::create) //
+				.verifyError(RuntimeException.class);
+
+		operations.count(new Query(), Person.class).as(StepVerifier::create).expectNext(0L).verifyComplete();
+	}
+
+	@Data
+	@AllArgsConstructor
+	static class Person {
+
+		ObjectId id;
+		String firstname, lastname;
+	}
+
+	@Data
+	@AllArgsConstructor
+	static class EventLog {
+
+		ObjectId id;
+		String action;
+	}
+
+}
diff --git a/src/main/asciidoc/reference/client-session-transactions.adoc b/src/main/asciidoc/reference/client-session-transactions.adoc
index 7961ae014..a79fcd235 100644
--- a/src/main/asciidoc/reference/client-session-transactions.adoc
+++ b/src/main/asciidoc/reference/client-session-transactions.adoc
@@ -220,7 +220,7 @@ NOTE: `@Transactional(readOnly = true)` advises `MongoTransactionManager` to als
 Same as with the reactive `ClientSession` support, the `ReactiveMongoTemplate` offers dedicated methods for operating
 within a transaction without having to worry about the commit/abort actions depending on the operations outcome.
 
-NOTE: Reactive use of `ClientSession` and transactions is limited to Template API usage. There's currently no session or transaction integration with reactive repositories.
+NOTE: Unless you specify a `ReactiveMongoTransactionManager` within your application context, transaction support is *DISABLED*. You can use `setSessionSynchronization(ALWAYS)` to participate in ongoing non-native MongoDB transactions.
 
 Using the plain MongoDB reactive driver API a `delete` within a transactional flow may look like this.
 
@@ -254,49 +254,78 @@ Mono<DeleteResult> result = Mono
 The culprit of the above operation is in keeping the main flows `DeleteResult` instead of the transaction outcome
 published via either `commitTransaction()` or `abortTransaction()`, which leads to a rather complicated setup.
 
-`MongoOperations.inTransaction()` allows you to utilize the callback from for the <<mongo.sessions.reactive,
-reactive session support>> to actually preserve the flows outcome but also perform commit and abort actions
-accordingly. This allows you to express the above flow simply as the following:
+== Transactions with `TransactionalOperator`
 
-.`ReactiveMongoTemplate` Transactions
+Spring Data MongoDB transactions support a `TransactionalOperator`. The following example shows how to create and use a `TransactionalOperator`:
+
+.Transactions with `TransactionalOperator`
 ====
 [source,java]
 ----
-Mono<DeleteResult> result = template.inTransaction()                                      <1>
+template.setSessionSynchronization(ALWAYS);                                          <1>
 
-    .execute(action -> action.remove(query(where("id").is("step-1")), Step.class));       <2>
+// ...
+
+TransactionalOperator rxtx = TransactionalOperator.create(anyTxManager,
+                                   new DefaultTransactionDefinition());              <2>
+
+
+Step step = // ...;
+template.insert(step);
+
+Mono<Void> process(step)
+					.then(template.update(Step.class).apply(Update.set("state", …))
+					.as(rxtx::transactional)                                         <3>
+					.then();
 ----
-<1> Initiate the transaction.
-<2> Operate within the `ClientSession`. Each `execute(…)` unit of work callback initiates a new transaction in the scope of the same `ClientSession`.
+<1> Enable transaction synchronization for Transactional participation.
+<2> Create the `TransactionalOperator` using the provided `ReactiveTransactionManager`.
+<3> `TransactionalOperator.transactional(…)` provides transaction management for all upstream operations.
 ====
 
-NOTE: In case you need access to the `ClientSession` within the flow, you can use `ReactiveMongoContext.getSession()`
- to obtain in from the Reactor `Context`.
+== Transactions with `ReactiveMongoTransactionManager`
 
-Everything happening inside the transactional callback is executed within a managed transaction. Errors within the
-reactive flow of `execute(…)` that are not propagated to outside of the callback do not affect the operations within the transaction.
+`ReactiveMongoTransactionManager` is the gateway to the well known Spring transaction support.
+It lets applications use  https://docs.spring.io/spring/docs/{springVersion}/spring-framework-reference/html/transaction.html[the managed transaction features of Spring].
+The `ReactiveMongoTransactionManager` binds a `ClientSession` to the subscriber `Context`.
+`ReactiveMongoTemplate` detects the session and operates on these resources which are associated with the transaction accordingly.
+`ReactiveMongoTemplate` can also participate in other, ongoing transactions.
+The following example shows how to create and use transactions with a `ReactiveMongoTransactionManager`:
 
+.Transactions with `ReactiveMongoTransactionManager`
 ====
 [source,java]
 ----
-template.inTransaction()                                                            <1>
+@Configuration
+static class Config extends AbstractMongoConfiguration {
 
-    .execute(action -> action.find(query(where("state").is("active")), Step.class)
-        .flatMap(step -> action.update(Step.class)
-            .matching(query(where("id").is(step.id)))
-            .apply(update("state", "paused"))
-            .all()))                                                                <2>
+    @Bean
+    ReactiveMongoTransactionManager transactionManager(ReactiveDatabaseFactory factory) {  <1>
+        return new ReactiveMongoTransactionManager(factory);
+    }
+
+    // ...
+}
+
+@Component
+public class StateService {
+
+    @Transactional
+    Mono<UpdateResult> someBusinessFunction(Step step) {                                  <2>
+
+        return template.insert(step)
+                   .then(process(step))
+                   .then(template.update(Step.class).apply(Update.set("state", …));
+    };
+});
 
-    .flatMap(updated -> {
-        // Exception could happen here                                              <3>
-    });
 ----
-<1> Initiate the managed transaction.
-<2> Operate within the `ClientSession`. The transaction is committed after this is done or rolled back if an
-error occurs here.
-<3> An error outside the transaction flow has no affect on the previous transactional execution.
+<1> Register `ReactiveMongoTransactionManager` in the application context.
+<2> Mark methods as transactional.
 ====
 
+NOTE: `@Transactional(readOnly = true)` advises `ReactiveMongoTransactionManager` to also start a transaction that adds the `ClientSession` to outgoing requests.
+
 [[mongo.transactions.behavior]]
 == Special behavior inside transactions