Spring/spring-data-mongodb
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAMONGO-1970 - Add support for MongoDB 4.0 transactions (reactive).

Browse Source 
We now support Mongo Transactions through the reactive Template API. However, there's no reactive repository transaction support yet.

Mono<DeleteResult> result = template.inTransaction()
                              .execute(action -> action.remove(query(where("id").is("step-1")), Step.class));

Original pull request: #560.
This commit is contained in:
Christoph Strobl
2018-05-11 10:38:50 +02:00
committed by Mark Paluch
parent 1acf00b039
commit f296a499e5
16 changed files with 551 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
pom.xml
View File

@@ -28,8 +28,8 @@
<project.type>multi</project.type>
<dist.id>spring-data-mongodb</dist.id>
<springdata.commons>2.1.0.BUILD-SNAPSHOT</springdata.commons>
<mongo>3.8.0-beta1</mongo>
<mongo.reactivestreams>1.8.0</mongo.reactivestreams>
<mongo>3.8.0-beta2</mongo>
<mongo.reactivestreams>1.9.0-beta1</mongo.reactivestreams>
<jmh.version>1.19</jmh.version>
</properties>

2
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/MongoTransactionManager.java
View File

@@ -360,7 +360,7 @@ public class MongoTransactionManager extends AbstractPlatformTransactionManager
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("statementId = %d, ", session.getServerSession().getStatementId());
debugString += String.format("closed = %d, ", session.getServerSession().isClosed());
debugString += String.format("clusterTime = %s", session.getClusterTime());
} else {
debugString += "id = n/a";

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

@@ -24,7 +24,7 @@ import org.springframework.dao.support.PersistenceExceptionTranslator;
import org.springframework.data.mongodb.core.MongoExceptionTranslator;
import com.mongodb.ClientSessionOptions;
import com.mongodb.session.ClientSession;
import com.mongodb.reactivestreams.client.ClientSession;
import com.mongodb.reactivestreams.client.MongoDatabase;
/**

56
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoContext.java Normal file
View File

@@ -0,0 +1,56 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.mongodb.core;
import reactor.core.publisher.Mono;
import reactor.util.context.Context;
import com.mongodb.reactivestreams.client.ClientSession;
/**
* {@link ReactiveMongoContext} utilizes and enriches the Reactor {@link Context} with information protentially required
* for e.g. {@link ClientSession} handling and transactions.
*
* @author Christoph Strobl
* @since 2.1
* @see Mono#subscriberContext()
* @see Context
*/
public class ReactiveMongoContext {
private static final Class<?> SESSION_KEY = ClientSession.class;
/**
* Gets the {@code Mono<ClientSession>} from Reactor {@link reactor.util.context.Context}
*
* @return the {@link Mono} emitting the client session.
*/
static Mono<ClientSession> getSession() {
return Mono.subscriberContext().filter(ctx -> ctx.hasKey(SESSION_KEY))
.flatMap(ctx -> ctx.<Mono<ClientSession>> get(SESSION_KEY));
}
/**
* Sets the {@link ClientSession} into the Reactor {@link reactor.util.context.Context}
*
* @return a new {@link Context}.
* @see Context#put(Object, Object)
*/
static Context setSession(Context context, Mono<ClientSession> session) {
return context.put(SESSION_KEY, session);
}
}

77
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java
View File

@@ -46,8 +46,8 @@ import com.mongodb.ClientSessionOptions;
import com.mongodb.ReadPreference;
import com.mongodb.client.result.DeleteResult;
import com.mongodb.client.result.UpdateResult;
import com.mongodb.reactivestreams.client.ClientSession;
import com.mongodb.reactivestreams.client.MongoCollection;
import com.mongodb.session.ClientSession;
/**
* Interface that specifies a basic set of MongoDB operations executed in a reactive way.
@@ -156,7 +156,7 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
* {@link ClientSession} when done.
*
* @param sessionProvider must not be {@literal null}.
* @return new instance of {@link SessionScoped}. Never {@literal null}.
* @return new instance of {@link ReactiveSessionScoped}. Never {@literal null}.
* @since 2.1
*/
default ReactiveSessionScoped withSession(Supplier<ClientSession> sessionProvider) {
@@ -169,52 +169,79 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
/**
* Obtain a {@link ClientSession session} bound instance of {@link SessionScoped} binding a new {@link ClientSession}
* with given {@literal sessionOptions} to each and every command issued against MongoDB.
* <p />
* <strong>Note:</strong> It is up to the caller to manage the {@link ClientSession} lifecycle. Use
* {@link ReactiveSessionScoped#execute(ReactiveSessionCallback, Consumer)} to provide a hook for processing the
* {@link ClientSession} when done.
*
* @param sessionOptions must not be {@literal null}.
* @return new instance of {@link SessionScoped}. Never {@literal null}.
* @return new instance of {@link ReactiveSessionScoped}. Never {@literal null}.
* @since 2.1
*/
ReactiveSessionScoped withSession(ClientSessionOptions sessionOptions);
/**
* Obtain a {@link ClientSession session} bound instance of {@link SessionScoped} binding the {@link ClientSession}
* provided by the given {@link Supplier} to each and every command issued against MongoDB.
* Obtain a {@link ClientSession session} bound instance of {@link ReactiveSessionScoped} binding the
* {@link ClientSession} provided by the given {@link Publisher} to each and every command issued against MongoDB.
* <p />
* <strong>Note:</strong> It is up to the caller to manage the {@link ClientSession} lifecycle. Use the
* {@literal onComplete} hook to potentially close the {@link ClientSession}.
* <strong>Note:</strong> It is up to the caller to manage the {@link ClientSession} lifecycle. Use
* {@link ReactiveSessionScoped#execute(ReactiveSessionCallback, Consumer)} to provide a hook for processing the
* {@link ClientSession} when done.
*
* @param sessionProvider must not be {@literal null}.
* @return new instance of {@link SessionScoped}. Never {@literal null}.
* @return new instance of {@link ReactiveSessionScoped}. Never {@literal null}.
* @since 2.1
*/
default ReactiveSessionScoped withSession(Publisher<ClientSession> sessionProvider) {
return new ReactiveSessionScoped() {
private final Mono<ClientSession> cachedSession = Mono.from(sessionProvider).cache();
@Override
public <T> Flux<T> execute(ReactiveSessionCallback<T> action, Consumer<ClientSession> doFinally) {
return cachedSession.flatMapMany(session -> {
return Flux.from(action.doInSession(ReactiveMongoOperations.this.withSession(session)))
.doFinally((signalType) -> doFinally.accept(session));
});
}
};
}
ReactiveSessionScoped withSession(Publisher<ClientSession> sessionProvider);
/**
* Obtain a {@link ClientSession} bound instance of {@link ReactiveMongoOperations}.
* <p />
* <strong>Note:</strong> It is up to the caller to manage the {@link ClientSession} lifecycle.
*
*
* @param session must not be {@literal null}.
* @return {@link ClientSession} bound instance of {@link ReactiveMongoOperations}.
* @since 2.1
*/
ReactiveMongoOperations withSession(ClientSession session);
/**
* Initiate a new {@link ClientSession} and obtain a {@link ClientSession session} bound instance of
* {@link ReactiveSessionScoped}. Starts the transaction and adds the {@link ClientSession} to each and every command
* issued against MongoDB.
*
* @return new instance of {@link ReactiveSessionScoped}. Never {@literal null}.
*/
ReactiveSessionScoped inTransaction();
/**
* Obtain a {@link ClientSession session} bound instance of {@link ReactiveSessionScoped}, start the transaction and
* bind the {@link ClientSession} provided by the given {@link Publisher} to each and every command issued against
* MongoDB.
*
* @param sessionProvider must not be {@literal null}.
* @return new instance of {@link ReactiveSessionScoped}. Never {@literal null}.
* @since 2.1
*/
default ReactiveSessionScoped inTransaction(Publisher<ClientSession> sessionProvider) {
return new ReactiveSessionScoped() {
@Override
public <T> Flux<T> flatMap(ReactiveSessionCallback<T> action, Consumer<ClientSession> doFinally) {
return ReactiveMongoOperations.this.withSession(Mono.from(sessionProvider).flatMap(session -> {
if (!session.hasActiveTransaction()) {
session.startTransaction();
}
return Mono.just(session);
})).execute(action, doFinally);
}
};
}
/**
* Create an uncapped collection with a name based on the provided entity class.
*

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

@@ -136,6 +136,7 @@ import com.mongodb.client.result.DeleteResult;
import com.mongodb.client.result.UpdateResult;
import com.mongodb.reactivestreams.client.AggregatePublisher;
import com.mongodb.reactivestreams.client.ChangeStreamPublisher;
import com.mongodb.reactivestreams.client.ClientSession;
import com.mongodb.reactivestreams.client.DistinctPublisher;
import com.mongodb.reactivestreams.client.FindPublisher;
import com.mongodb.reactivestreams.client.MapReducePublisher;
@@ -143,7 +144,6 @@ import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoCollection;
import com.mongodb.reactivestreams.client.MongoDatabase;
import com.mongodb.reactivestreams.client.Success;
import com.mongodb.session.ClientSession;
import com.mongodb.util.JSONParseException;
/**
@@ -476,17 +476,46 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
return new ReactiveSessionScoped() {
@Override
public <T> Flux<T> execute(ReactiveSessionCallback<T> action, Consumer<ClientSession> doFinally) {
public <T> Flux<T> flatMap(ReactiveSessionCallback<T> action, Consumer<ClientSession> doFinally) {
return cachedSession.flatMapMany(session -> {
return Flux
.from(action.doInSession(new ReactiveSessionBoundMongoTemplate(session, ReactiveMongoTemplate.this)))
.doFinally((signalType) -> doFinally.accept(session));
.from(action.doInSession(new ReactiveSessionBoundMongoTemplate(session, ReactiveMongoTemplate.this))) //
.subscriberContext(ctx -> ReactiveMongoContext.setSession(ctx, cachedSession)) //
.materialize() //
.flatMap(signal -> {
if (session.hasActiveTransaction()) {
if (signal.isOnComplete()) {
return Mono.from(session.commitTransaction()).thenReturn(signal);
}
if (signal.isOnError()) {
return Mono.from(session.abortTransaction()).thenReturn(signal);
}
}
return Mono.just(signal);
}) //
.<T> dematerialize() //
.doFinally(signalType -> {
doFinally.accept(session);
});
});
}
};
}
/*
* (non-Javadoc)
* @see org.springframework.data.mongodb.core.ReactiveMongoOperations#inTransaction()
*/
@Override
public ReactiveSessionScoped inTransaction() {
return inTransaction(
mongoDatabaseFactory.getSession(ClientSessionOptions.builder().causallyConsistent(true).build()));
}
/*
* (non-Javadoc)
* @see org.springframework.data.mongodb.core.ReactiveMongoOperations#withSession(com.mongodb.session.ClientSession)

6
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveSessionCallback.java
View File

@@ -19,12 +19,12 @@ import org.reactivestreams.Publisher;
import org.springframework.data.mongodb.core.query.Query;
/**
* Callback interface for executing operations within a {@link com.mongodb.session.ClientSession} using reactive
* infrastructure.
* Callback interface for executing operations within a {@link com.mongodb.reactivestreams.client.ClientSession} using
* reactive infrastructure.
*
* @author Christoph Strobl
* @since 2.1
* @see com.mongodb.session.ClientSession
* @see com.mongodb.reactivestreams.client.ClientSession
*/
@FunctionalInterface
public interface ReactiveSessionCallback<T> {

59
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveSessionScoped.java
View File

@@ -16,10 +16,13 @@
package org.springframework.data.mongodb.core;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.function.Consumer;
import com.mongodb.session.ClientSession;
import org.reactivestreams.Publisher;
import com.mongodb.reactivestreams.client.ClientSession;
/**
* Gateway interface to execute {@link ClientSession} bound operations against MongoDB via a
@@ -58,5 +61,57 @@ public interface ReactiveSessionScoped {
* @param <T> return type.
* @return a result object returned by the action. Can be {@literal null}.
*/
<T> Flux<T> execute(ReactiveSessionCallback<T> action, Consumer<ClientSession> doFinally);
default <T> Flux<T> execute(ReactiveSessionCallback<T> action, Consumer<ClientSession> doFinally) {
return Flux.from(flatMap(action, doFinally));
}
/**
* Executes the given {@link ReactiveSessionCallback} within the {@link com.mongodb.session.ClientSession} returning a
* plain {@link Publisher}. For more convenience use {@link #execute(ReactiveSessionCallback, Consumer)} or
* {@link #executeSingle(ReactiveSessionCallback, Consumer)} to get the Reactor types.
* <p/>
* It is up to the caller to make sure the {@link com.mongodb.session.ClientSession} is {@link ClientSession#close()
* closed} when done.
*
* @param action callback object that specifies the MongoDB action the callback action. Must not be {@literal null}.
* @param doFinally callback object that accepts {@link ClientSession} after invoking {@link ReactiveSessionCallback}.
* This {@link Consumer} is guaranteed to be notified in any case (successful and exceptional outcome of
* {@link ReactiveSessionCallback}).
* @param <T> return type.
* @return a result object returned by the action. Can be {@literal null}.
*/
<T> Publisher<T> flatMap(ReactiveSessionCallback<T> action, Consumer<ClientSession> doFinally);
/**
* Executes the given {@link ReactiveSessionCallback} within the {@link com.mongodb.session.ClientSession} returning
* single result emitted via {@link Mono}.
* <p/>
* It is up to the caller to make sure the {@link com.mongodb.session.ClientSession} is {@link ClientSession#close()
* closed} when done.
*
* @param action callback object that specifies the MongoDB action the callback action. Must not be {@literal null}.
* @param <T> return type.
* @return a result object returned by the action. Can be {@literal null}.
*/
default <T> Mono<T> executeSingle(ReactiveSessionCallback<T> action) {
return executeSingle(action, (session) -> {});
}
/**
* Executes the given {@link ReactiveSessionCallback} within the {@link com.mongodb.session.ClientSession} returning
* single result emitted via {@link Mono}.
* <p/>
* It is up to the caller to make sure the {@link com.mongodb.session.ClientSession} is {@link ClientSession#close()
* closed} when done.
*
* @param action callback object that specifies the MongoDB action the callback action. Must not be {@literal null}.
* @param doFinally callback object that accepts {@link ClientSession} after invoking {@link ReactiveSessionCallback}.
* This {@link Consumer} is guaranteed to be notified in any case (successful and exceptional outcome of
* {@link ReactiveSessionCallback}).
* @param <T> return type.
* @return a result object returned by the action. Can be {@literal null}.
*/
default <T> Mono<T> executeSingle(ReactiveSessionCallback<T> action, Consumer<ClientSession> doFinally) {
return Mono.from(flatMap(action, doFinally));
}
}

12
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactory.java
View File

@@ -30,11 +30,11 @@ import org.springframework.util.Assert;
import com.mongodb.ClientSessionOptions;
import com.mongodb.ConnectionString;
import com.mongodb.WriteConcern;
import com.mongodb.reactivestreams.client.ClientSession;
import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoClients;
import com.mongodb.reactivestreams.client.MongoCollection;
import com.mongodb.reactivestreams.client.MongoDatabase;
import com.mongodb.session.ClientSession;
/**
* Factory to create {@link MongoDatabase} instances from a {@link MongoClient} instance.
@@ -215,23 +215,23 @@ public class SimpleReactiveMongoDatabaseFactory implements DisposableBean, React
return createProxyInstance(session, database, MongoDatabase.class);
}
private MongoDatabase proxyDatabase(ClientSession session, MongoDatabase database) {
private MongoDatabase proxyDatabase(com.mongodb.session.ClientSession session, MongoDatabase database) {
return createProxyInstance(session, database, MongoDatabase.class);
}
private MongoCollection proxyCollection(ClientSession session, MongoCollection collection) {
private MongoCollection proxyCollection(com.mongodb.session.ClientSession session, MongoCollection collection) {
return createProxyInstance(session, collection, MongoCollection.class);
}
private <T> T createProxyInstance(ClientSession session, T target, Class<T> targetType) {
private <T> T createProxyInstance(com.mongodb.session.ClientSession session, T target, Class<T> targetType) {
ProxyFactory factory = new ProxyFactory();
factory.setTarget(target);
factory.setInterfaces(targetType);
factory.setOpaque(true);
factory.addAdvice(new SessionAwareMethodInterceptor<>(session, target, ClientSession.class, MongoDatabase.class, this::proxyDatabase,
MongoCollection.class, this::proxyCollection));
factory.addAdvice(new SessionAwareMethodInterceptor<>(session, target, ClientSession.class, MongoDatabase.class,
this::proxyDatabase, MongoCollection.class, this::proxyCollection));
return targetType.cast(factory.getProxy());
}

37
spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveClientSessionTests.java
View File

@@ -29,37 +29,43 @@ import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.TestRule;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.test.util.MongoTestUtils;
import org.springframework.data.mongodb.test.util.MongoVersionRule;
import org.springframework.data.mongodb.test.util.ReplicaSet;
import org.springframework.data.util.Version;
import com.mongodb.ClientSessionOptions;
import com.mongodb.reactivestreams.client.ClientSession;
import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoClients;
import com.mongodb.session.ClientSession;
import com.mongodb.reactivestreams.client.Success;
/**
* @author Christoph Strobl
* @author Mark Paluch
* @currentRead Beyond the Shadows - Brent Weeks
*/
public class ReactiveClientSessionTests {
public static @ClassRule MongoVersionRule REQUIRES_AT_LEAST_3_6_0 = MongoVersionRule.atLeast(Version.parse("3.6.0"));
public static @ClassRule TestRule replSet = ReplicaSet.required();
static final String DATABASE_NAME = "reflective-client-session-tests";
static final String COLLECTION_NAME = "test";
MongoClient client;
ReactiveMongoTemplate template;
@Before
public void setUp() {
client = MongoClients.create();
client = MongoTestUtils.reactiveReplSetClient();
template = new ReactiveMongoTemplate(client, "reflective-client-session-tests");
template = new ReactiveMongoTemplate(client, DATABASE_NAME);
StepVerifier.create(template.dropCollection("test")).verifyComplete();
StepVerifier.create(MongoTestUtils.createOrReplaceCollection(DATABASE_NAME, COLLECTION_NAME, client))
.expectNext(Success.SUCCESS).verifyComplete();
StepVerifier.create(template.insert(new Document("_id", "id-1").append("value", "spring"), "test"))
StepVerifier.create(template.insert(new Document("_id", "id-1").append("value", "spring"), COLLECTION_NAME))
.expectNextCount(1).verifyComplete();
}
@@ -71,7 +77,8 @@ public class ReactiveClientSessionTests {
assertThat(session.getOperationTime()).isNull();
StepVerifier.create(template.withSession(() -> session).execute(action -> action.findAll(Document.class, "test")))
StepVerifier
.create(template.withSession(() -> session).execute(action -> action.findAll(Document.class, COLLECTION_NAME)))
.expectNextCount(1).verifyComplete();
assertThat(session.getOperationTime()).isNotNull();
@@ -89,8 +96,8 @@ public class ReactiveClientSessionTests {
assertThat(session.getOperationTime()).isNull();
StepVerifier
.create(
template.withSession(() -> session).execute(action -> action.findOne(new Query(), Document.class, "test")))
.create(template.withSession(() -> session)
.execute(action -> action.findOne(new Query(), Document.class, COLLECTION_NAME)))
.expectNextCount(1).verifyComplete();
assertThat(session.getOperationTime()).isNotNull();
@@ -108,13 +115,21 @@ public class ReactiveClientSessionTests {
ReactiveSessionScoped sessionScoped = template.withSession(sessionSupplier);
sessionScoped.execute(action -> action.findOne(new Query(), Document.class, "test")).blockFirst();
sessionScoped.execute(action -> action.findOne(new Query(), Document.class, COLLECTION_NAME)).blockFirst();
assertThat(sessionSupplier.getInvocationCount()).isEqualTo(1);
sessionScoped.execute(action -> action.findOne(new Query(), Document.class, "test")).blockFirst();
sessionScoped.execute(action -> action.findOne(new Query(), Document.class, COLLECTION_NAME)).blockFirst();
assertThat(sessionSupplier.getInvocationCount()).isEqualTo(1);
}
@Test // DATAMONGO-1970
public void addsClientSessionToContext() {
template.withSession(client.startSession(ClientSessionOptions.builder().causallyConsistent(true).build()))
.execute(action -> ReactiveMongoContext.getSession()).as(StepVerifier::create)
.consumeNextWith(session -> assertThat(session).isNotNull()).verifyComplete();
}
static class CountingSessionSupplier implements Supplier<ClientSession> {
AtomicInteger invocationCount = new AtomicInteger(0);

174
spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTransactionTests.java Normal file
View File

@@ -0,0 +1,174 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.mongodb.core;
import static org.springframework.data.mongodb.core.query.Criteria.*;
import static org.springframework.data.mongodb.core.query.Query.*;
import org.springframework.data.mongodb.core.query.Update;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import java.util.Arrays;
import java.util.stream.Collectors;
import org.bson.Document;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.TestRule;
import org.reactivestreams.Publisher;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.test.util.MongoTestUtils;
import org.springframework.data.mongodb.test.util.MongoVersionRule;
import org.springframework.data.mongodb.test.util.ReplicaSet;
import org.springframework.data.util.Version;
import com.mongodb.ClientSessionOptions;
import com.mongodb.reactivestreams.client.ClientSession;
import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.Success;
/**
* @author Christoph Strobl
* @currentRead The Core - Peter V. Brett
*/
public class ReactiveMongoTemplateTransactionTests {
public static @ClassRule MongoVersionRule REQUIRES_AT_LEAST_3_7_5 = MongoVersionRule.atLeast(Version.parse("3.7.5"));
public static @ClassRule TestRule replSet = ReplicaSet.required();
static final String DATABASE_NAME = "reactive-template-tx-tests";
static final String COLLECTION_NAME = "test";
static final Document DOCUMENT = new Document("_id", "id-1").append("value", "spring");
static final Query ID_QUERY = query(where("_id").is("id-1"));
static final Person AHMANN = new Person("ahmann", 32);
static final Person ARLEN = new Person("arlen", 24);
static final Person LEESHA = new Person("leesha", 22);
static final Person RENNA = new Person("renna", 22);
MongoClient client;
ReactiveMongoTemplate template;
@Before
public void setUp() {
client = MongoTestUtils.reactiveReplSetClient();
template = new ReactiveMongoTemplate(client, DATABASE_NAME);
StepVerifier.create(MongoTestUtils.createOrReplaceCollection(DATABASE_NAME, COLLECTION_NAME, client))
.expectNext(Success.SUCCESS).verifyComplete();
StepVerifier.create(MongoTestUtils.createOrReplaceCollection(DATABASE_NAME, "person", client))
.expectNext(Success.SUCCESS).verifyComplete();
StepVerifier.create(template.insert(DOCUMENT, COLLECTION_NAME)).expectNextCount(1).verifyComplete();
template.insertAll(Arrays.asList(AHMANN, ARLEN, LEESHA, RENNA)).as(StepVerifier::create).expectNextCount(4)
.verifyComplete();
}
@Test // DATAMONGO-1970
public void reactiveTransactionWithExplicitTransactionStart() throws InterruptedException {
Publisher<ClientSession> sessionPublisher = client
.startSession(ClientSessionOptions.builder().causallyConsistent(true).build());
template.withSession(sessionPublisher)
.executeSingle(action -> ReactiveMongoContext.getSession().flatMap(session -> {
session.startTransaction();
return action.remove(ID_QUERY, Document.class, COLLECTION_NAME);
})).as(StepVerifier::create).expectNextCount(1).verifyComplete();
template.exists(ID_QUERY, COLLECTION_NAME) //
.as(StepVerifier::create).expectNext(false).verifyComplete();
}
@Test // DATAMONGO-1970
public void reactiveTransactionsCommitOnComplete() throws InterruptedException {
template.inTransaction().execute(action -> action.remove(ID_QUERY, Document.class, COLLECTION_NAME)) //
.as(StepVerifier::create).expectNextCount(1).verifyComplete();
template.exists(ID_QUERY, COLLECTION_NAME) //
.as(StepVerifier::create).expectNext(false).verifyComplete();
}
@Test // DATAMONGO-1970
public void reactiveTransactionsAbortOnError() throws InterruptedException {
template.inTransaction().execute(action -> {
return action.remove(ID_QUERY, Document.class, COLLECTION_NAME).flatMap(result -> Mono.fromSupplier(() -> {
throw new RuntimeException("¯\\_(ツ)_/¯");
}));
}).as(StepVerifier::create).expectError().verify();
template.exists(ID_QUERY, COLLECTION_NAME) //
.as(StepVerifier::create).expectNext(true).verifyComplete();
}
@Test // DATAMONGO-1970
public void changesNotVisibleOutsideTransaction() throws InterruptedException {
template.inTransaction().execute(action -> {
return action.remove(ID_QUERY, Document.class, COLLECTION_NAME).flatMap(val -> {
// once we use the collection directly we're no longer participating in the tx
return Mono.from(template.getCollection(COLLECTION_NAME).find(ID_QUERY.getQueryObject()));
});
}).as(StepVerifier::create).expectNext(DOCUMENT).verifyComplete();
template.exists(ID_QUERY, COLLECTION_NAME) //
.as(StepVerifier::create).expectNext(false).verifyComplete();
}
@Test // DATAMONGO-1970
public void takeDoesNotAbortTransaction() throws InterruptedException {
template.inTransaction().execute(action -> {
return action.find(query(where("age").exists(true)).with(Sort.by("age")), Person.class).take(3)
.flatMap(person -> {
return action.remove(person);
});
}).as(StepVerifier::create).expectNextCount(3).verifyComplete();
template.count(query(where("age").exists(true)), Person.class) //
.as(StepVerifier::create).expectNext(1L).verifyComplete();
}
@Test // DATAMONGO-1970
public void errorInFlowOutsideTransactionDoesNotAbortIt() throws InterruptedException {
template.inTransaction().execute(action -> {
return action.find(query(where("age").is(22)).with(Sort.by("age")), Person.class).buffer(2).flatMap(values -> {
return action.remove(query(where("id").in(values.stream().map(Person::getId).collect(Collectors.toList()))),
Person.class).then(Mono.just(values));
});
}).flatMap(deleted -> {
throw new RuntimeException("error outside the transaction does not influence it.");
}).as(StepVerifier::create).expectError().verify();
template.count(query(where("age").exists(true)), Person.class) //
.as(StepVerifier::create).expectNext(2L).verifyComplete();
}
}

2
spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveSessionBoundMongoTemplateUnitTests.java
View File

@@ -53,12 +53,12 @@ import com.mongodb.client.model.DeleteOptions;
import com.mongodb.client.model.FindOneAndUpdateOptions;
import com.mongodb.client.model.UpdateOptions;
import com.mongodb.reactivestreams.client.AggregatePublisher;
import com.mongodb.reactivestreams.client.ClientSession;
import com.mongodb.reactivestreams.client.DistinctPublisher;
import com.mongodb.reactivestreams.client.FindPublisher;
import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoCollection;
import com.mongodb.reactivestreams.client.MongoDatabase;
import com.mongodb.session.ClientSession;
/**
* Unit tests for {@link ReactiveSessionBoundMongoTemplate}.

2
spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactoryUnitTests.java
View File

@@ -31,9 +31,9 @@ import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
import org.springframework.test.util.ReflectionTestUtils;
import com.mongodb.reactivestreams.client.ClientSession;
import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoDatabase;
import com.mongodb.session.ClientSession;
/**
* Unit tests for {@link SimpleReactiveMongoDatabaseFactory}.

36
spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonRepositoryTransactionalTests.java
View File

@@ -54,8 +54,8 @@ import org.springframework.test.context.transaction.BeforeTransaction;
import org.springframework.transaction.annotation.Transactional;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.ReadPreference;
import com.mongodb.WriteConcern;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.model.Filters;
@@ -123,20 +123,30 @@ public class PersonRepositoryTransactionalTests {
}
@AfterTransaction
public void verifyDbState() {
public void verifyDbState() throws InterruptedException {
MongoCollection<Document> collection = client.getDatabase(DB_NAME)
Thread.sleep(100);
MongoCollection<Document> collection = client.getDatabase(DB_NAME) //
.withWriteConcern(WriteConcern.MAJORITY) //
.withReadPreference(ReadPreference.primary()) //
.getCollection(template.getCollectionName(Person.class));
assertionList.forEach(it -> {
try {
assertionList.forEach(it -> {
boolean isPresent = collection.find(Filters.eq("_id", new ObjectId(it.getId().toString()))).limit(1).iterator()
.hasNext();
boolean isPresent = collection.find(Filters.eq("_id", new ObjectId(it.getId().toString()))).iterator()
.hasNext();
assertThat(isPresent).isEqualTo(it.shouldBePresent())
.withFailMessage(String.format("After transaction entity %s should %s.", it.getPersistable(),
it.shouldBePresent() ? "be present" : "NOT be present"));
});
assertThat(isPresent) //
.withFailMessage(String.format("After transaction entity %s should %s.", it.getPersistable(),
it.shouldBePresent() ? "be present" : "NOT be present"))
.isEqualTo(it.shouldBePresent());
});
} finally {
assertionList.clear();
}
}
@Rollback(false)
@@ -144,7 +154,6 @@ public class PersonRepositoryTransactionalTests {
public void shouldHonorCommitForDerivedQuery() {
repository.removePersonByLastnameUsingAnnotatedQuery(durzo.getLastname());
repository.removePersonByLastnameUsingAnnotatedQuery(durzo.getLastname());
assertAfterTransaction(durzo).isNotPresent();
}
@@ -184,6 +193,11 @@ public class PersonRepositoryTransactionalTests {
public boolean isNew() {
return person.id != null;
}
@Override
public String toString() {
return getId() + " - " + person.toString();
}
});
assertionList.add(assertion);

46
spring-data-mongodb/src/test/java/org/springframework/data/mongodb/test/util/MongoTestUtils.java
View File

@@ -15,20 +15,25 @@
*/
package org.springframework.data.mongodb.test.util;
import reactor.core.publisher.Mono;
import org.bson.Document;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoClientURI;
import com.mongodb.ReadPreference;
import com.mongodb.WriteConcern;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import com.mongodb.reactivestreams.client.MongoClients;
import com.mongodb.reactivestreams.client.Success;
/**
* @author Christoph Strobl
*/
public class MongoTestUtils {
public static final String CONNECTION_STRING = "mongodb://localhost:27017/?replicaSet=rs0"; // &readPreference=primary&w=majority
/**
* Create a {@link com.mongodb.client.MongoCollection} if it does not exist, or drop and recreate it if it does.
*
@@ -37,7 +42,7 @@ public class MongoTestUtils {
* @param client must not be {@literal null}.
*/
public static MongoCollection<Document> createOrReplaceCollection(String dbName, String collectionName,
MongoClient client) {
com.mongodb.MongoClient client) {
MongoDatabase database = client.getDatabase(dbName);
@@ -59,14 +64,39 @@ public class MongoTestUtils {
}
/**
* Create a new {@link MongoClient} with defaults suitable for replica set usage.
* Create a {@link com.mongodb.client.MongoCollection} if it does not exist, or drop and recreate it if it does.
*
* @return new instance of {@link MongoClient}.
* @param dbName must not be {@literal null}.
* @param collectionName must not be {@literal null}.
* @param client must not be {@literal null}.
*/
public static MongoClient replSetClient() {
public static Mono<Success> createOrReplaceCollection(String dbName, String collectionName,
com.mongodb.reactivestreams.client.MongoClient client) {
return new MongoClient("localhost",
MongoClientOptions.builder().requiredReplicaSetName("rs0").build());
com.mongodb.reactivestreams.client.MongoDatabase database = client.getDatabase(dbName)
.withWriteConcern(WriteConcern.MAJORITY).withReadPreference(ReadPreference.primary());
return Mono.from(database.getCollection(collectionName).drop())
.then(Mono.from(database.createCollection(collectionName)));
}
/**
* Create a new {@link com.mongodb.MongoClient} with defaults suitable for replica set usage.
*
* @return new instance of {@link com.mongodb.MongoClient}.
*/
public static com.mongodb.MongoClient replSetClient() {
return new com.mongodb.MongoClient(new MongoClientURI(CONNECTION_STRING));
}
/**
* Create a new {@link com.mongodb.reactivestreams.client.MongoClient} with defaults suitable for replica set usage.
*
* @return new instance of {@link com.mongodb.reactivestreams.client.MongoClient}.
*/
public static com.mongodb.reactivestreams.client.MongoClient reactiveReplSetClient() {
return MongoClients.create(CONNECTION_STRING);
}
}

75
src/main/asciidoc/reference/client-session-transactions.adoc
View File

@@ -9,6 +9,9 @@ Both `MongoOperations` and `ReactiveMongoOperations` provide gateway methods for
NOTE: Methods such as `(Reactive)MongoOperations#getCollection` return native MongoDB Java Driver gateway objects (such as `MongoCollection`) that themselves offer dedicated methods for `ClientSession`. These methods are *NOT* session-proxied. You should provide the `ClientSession` where needed when interacting directly with a `MongoCollection` or `MongoDatabase` and not through one of the `#execute` callbacks on `MongoOperations`.
[[mongo.sessions.sync]]
== Synchronous `ClientSession` support.
The following example shows the usage of a session:
.`ClientSession` with `MongoOperations`
@@ -45,6 +48,9 @@ session.close() <4>
WARNING: When dealing with `DBRef` instances, especially lazily loaded ones, it is essential to *not* close the `ClientSession` before all data is loaded. Otherwise, lazy fetch fails.
[[mongo.sessions.reactive]]
== Reactive `ClientSession` support
The reactive counterpart uses the same building blocks as the imperative one, as the following example shows:
.ClientSession with `ReactiveMongoOperations`
@@ -207,3 +213,72 @@ public class StateService {
NOTE: `@Transactional(readOnly = true)` advises `MongoTransactionManager` to also start a transaction that adds the
`ClientSession` to outgoing requests.
== Reactive transactions
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.
Using the plain MongoDB reactive driver API a `delete within a transactional flow may look like this.
.Native driver support
====
[source,java]
----
Mono<DeleteResult> result = Mono
.from(client.startSession()) <1>
.flatMap(session -> {
session.startTransaction(); <2>
return Mono.from(collection.deleteMany(session, ...)) <3>
.onErrorResume(e -> Mono.from(session.abortTransaction()).then(Mono.error(e))) <4>
.flatMap(val -> Mono.from(session.commitTransaction()).then(Mono.just(val))) <5>
.doFinally(signal -> session.close()); <6>
});
----
<1> Ok, first we obvoiusly need to initiate the session.
<2> Once we've the `ClientSession` at hand, start the transaction.
<3> Operate within the transaction by passing on the `ClientSession` to the operation.
<4> If the operations errors, we need to abort the transaction and preserve the error.
<5> Or of course, commit the changes in case of success. Still preserving the operations result.
<6> Last, we need to make sure to close the session.
====`
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:
.ReactiveMongoTemplate transactions
====
[source,java]
----
Mono<DeleteResult> result = template.inTransaction() <1>
.execute(action -> action.remove(query(where("id").is("step-1")), Step.class)); <2>
----
<1> Initiate the transaction.
<2> Operate within the `ClientSession`.
====
NOTE: In case you need access to the `ClientSession` within the flow, you can use `ReactiveMongoContext.getSession()`
to obtain in from the Reactor `Context`.
Everything happening inside the transactional callback is executed within a transaction. Errors subsequent in the
reactive flow do not influence the operations within the transaction.
====
[source,java]
----
template.inTransaction() <1>
.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>
.flatMap(deleted -> {
// errors here <3>
}).subscribe();
----
<1> Initiate the transaction.
<2> Operate within the `ClientSession`. The transaction is committed after when 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.
====