DATAMONGO-2265 - Add initial ReactiveMongoTransactionManager.
Support declarative reactive transaction via the Transactional annotation via a MongoDB specific ReactiveTransactionManager implementation.
@Bean
ReactiveMongoTransactionManager transactionManager(ReactiveDatabaseFactory factory) {
return new ReactiveMongoTransactionManager(factory);
}
@Component
public class StateService {
@Transactional
Mono<UpdateResult> someBusinessFunction(Step step) {
return template.insert(step)
.then(process(step))
.then(template.update(Step.class).apply(Update.set("state", …));
};
});
Original Pull Request: #745
This commit is contained in:
Mark Paluch
committed by
Christoph Strobl
Christoph Strobl
parent
c15ab4c946
commit
5c10a5821b
@@ -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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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");
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
|
||||
/**
|
||||
|
||||
@@ -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.
|
||||
*
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
@@ -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"));
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user