diff --git a/pom.xml b/pom.xml index ff6cdb7d9..7227da358 100644 --- a/pom.xml +++ b/pom.xml @@ -28,8 +28,8 @@ multi spring-data-mongodb 2.1.0.BUILD-SNAPSHOT - 3.8.0-beta1 - 1.8.0 + 3.8.0-beta2 + 1.9.0-beta1 1.19 diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/MongoTransactionManager.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/MongoTransactionManager.java index fc403adae..5ec4e10e7 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/MongoTransactionManager.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/MongoTransactionManager.java @@ -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"; diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseFactory.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseFactory.java index 1b6dbf73b..69d606c84 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseFactory.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseFactory.java @@ -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; /** diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoContext.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoContext.java new file mode 100644 index 000000000..9120af61e --- /dev/null +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoContext.java @@ -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} from Reactor {@link reactor.util.context.Context} + * + * @return the {@link Mono} emitting the client session. + */ + static Mono getSession() { + + return Mono.subscriberContext().filter(ctx -> ctx.hasKey(SESSION_KEY)) + .flatMap(ctx -> ctx.> 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 session) { + return context.put(SESSION_KEY, session); + } +} diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java index 1dca1b0d8..b75d93803 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java @@ -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 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. + *

+ * Note: 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. *

- * Note: It is up to the caller to manage the {@link ClientSession} lifecycle. Use the - * {@literal onComplete} hook to potentially close the {@link ClientSession}. + * Note: 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 sessionProvider) { - - return new ReactiveSessionScoped() { - - private final Mono cachedSession = Mono.from(sessionProvider).cache(); - - @Override - public Flux execute(ReactiveSessionCallback action, Consumer doFinally) { - - return cachedSession.flatMapMany(session -> { - return Flux.from(action.doInSession(ReactiveMongoOperations.this.withSession(session))) - .doFinally((signalType) -> doFinally.accept(session)); - }); - } - }; - } + ReactiveSessionScoped withSession(Publisher sessionProvider); /** * Obtain a {@link ClientSession} bound instance of {@link ReactiveMongoOperations}. *

* Note: 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 sessionProvider) { + + return new ReactiveSessionScoped() { + + @Override + public Flux flatMap(ReactiveSessionCallback action, Consumer 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. * diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java index b974d2440..7f21da2fa 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java @@ -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 Flux execute(ReactiveSessionCallback action, Consumer doFinally) { + public Flux flatMap(ReactiveSessionCallback action, Consumer 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); + }) // + . 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) diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveSessionCallback.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveSessionCallback.java index a6f46b033..253ae13cd 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveSessionCallback.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveSessionCallback.java @@ -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 { diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveSessionScoped.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveSessionScoped.java index 047dbe6ef..f25be4ddc 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveSessionScoped.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveSessionScoped.java @@ -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 return type. * @return a result object returned by the action. Can be {@literal null}. */ - Flux execute(ReactiveSessionCallback action, Consumer doFinally); + default Flux execute(ReactiveSessionCallback action, Consumer 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. + *

+ * 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 return type. + * @return a result object returned by the action. Can be {@literal null}. + */ + Publisher flatMap(ReactiveSessionCallback action, Consumer doFinally); + + /** + * Executes the given {@link ReactiveSessionCallback} within the {@link com.mongodb.session.ClientSession} returning + * single result emitted via {@link Mono}. + *

+ * 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 return type. + * @return a result object returned by the action. Can be {@literal null}. + */ + default Mono executeSingle(ReactiveSessionCallback action) { + return executeSingle(action, (session) -> {}); + } + + /** + * Executes the given {@link ReactiveSessionCallback} within the {@link com.mongodb.session.ClientSession} returning + * single result emitted via {@link Mono}. + *

+ * 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 return type. + * @return a result object returned by the action. Can be {@literal null}. + */ + default Mono executeSingle(ReactiveSessionCallback action, Consumer doFinally) { + return Mono.from(flatMap(action, doFinally)); + } } diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactory.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactory.java index 726fb4e9b..71d73a442 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactory.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactory.java @@ -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 createProxyInstance(ClientSession session, T target, Class targetType) { + private T createProxyInstance(com.mongodb.session.ClientSession session, T target, Class 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()); } diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveClientSessionTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveClientSessionTests.java index d409c07b2..9aa4f489b 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveClientSessionTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveClientSessionTests.java @@ -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 { AtomicInteger invocationCount = new AtomicInteger(0); diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTransactionTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTransactionTests.java new file mode 100644 index 000000000..ccf585fc8 --- /dev/null +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTransactionTests.java @@ -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 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(); + } +} diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveSessionBoundMongoTemplateUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveSessionBoundMongoTemplateUnitTests.java index 412eec50a..05c940d5c 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveSessionBoundMongoTemplateUnitTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveSessionBoundMongoTemplateUnitTests.java @@ -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}. diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactoryUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactoryUnitTests.java index df7dbe599..42e96f8eb 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactoryUnitTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactoryUnitTests.java @@ -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}. diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonRepositoryTransactionalTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonRepositoryTransactionalTests.java index 1883d1613..9807422e8 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonRepositoryTransactionalTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonRepositoryTransactionalTests.java @@ -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 collection = client.getDatabase(DB_NAME) + Thread.sleep(100); + + MongoCollection 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); diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/test/util/MongoTestUtils.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/test/util/MongoTestUtils.java index 023665b41..18fad010c 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/test/util/MongoTestUtils.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/test/util/MongoTestUtils.java @@ -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 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 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); } } diff --git a/src/main/asciidoc/reference/client-session-transactions.adoc b/src/main/asciidoc/reference/client-session-transactions.adoc index f2c11591c..8b9018fac 100644 --- a/src/main/asciidoc/reference/client-session-transactions.adoc +++ b/src/main/asciidoc/reference/client-session-transactions.adoc @@ -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 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 <> 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 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. +====