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 index 9120af61e..edb9144a0 100644 --- 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 @@ -15,16 +15,19 @@ */ package org.springframework.data.mongodb.core; +import org.reactivestreams.Publisher; +import org.springframework.util.Assert; 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 + * {@link ReactiveMongoContext} utilizes and enriches the Reactor {@link Context} with information potentially required * for e.g. {@link ClientSession} handling and transactions. * * @author Christoph Strobl + * @author Mark Paluch * @since 2.1 * @see Mono#subscriberContext() * @see Context @@ -34,23 +37,32 @@ public class ReactiveMongoContext { private static final Class SESSION_KEY = ClientSession.class; /** - * Gets the {@code Mono} from Reactor {@link reactor.util.context.Context} + * Gets the {@code Mono} from Reactor {@link reactor.util.context.Context}. The resulting {@link Mono} + * emits the {@link ClientSession} if a session is associated with the current {@link reactor.util.context.Context + * subscriber context}. If the context does not contain a session, the resulting {@link Mono} terminates empty (i.e. + * without emitting a value). * - * @return the {@link Mono} emitting the client session. + * @return the {@link Mono} emitting the client session if present; otherwise the {@link Mono} terminates empty. */ - static Mono getSession() { + public 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} + * Sets the {@link ClientSession} into the Reactor {@link reactor.util.context.Context}. * + * @param context must not be {@literal null}. + * @param session must not be {@literal null}. * @return a new {@link Context}. * @see Context#put(Object, Object) */ - static Context setSession(Context context, Mono session) { - return context.put(SESSION_KEY, session); + public static Context setSession(Context context, Publisher session) { + + Assert.notNull(context, "Context must not be null!"); + Assert.notNull(session, "Session publisher must not be null!"); + + return context.put(SESSION_KEY, Mono.from(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 b75d93803..c16e18248 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 @@ -198,7 +198,7 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations { * 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 @@ -209,6 +209,10 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations { * 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. + *

+ * Each {@link ReactiveSessionScoped#execute(ReactiveSessionCallback) execution} initiates a new managed transaction + * that is {@link ClientSession#commitTransaction() committed} on success. Transactions are + * {@link ClientSession#abortTransaction() rolled back} upon errors. * * @return new instance of {@link ReactiveSessionScoped}. Never {@literal null}. */ @@ -218,29 +222,16 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations { * 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. - * + *

+ * Each {@link ReactiveSessionScoped#execute(ReactiveSessionCallback) execution} initiates a new managed transaction + * that is {@link ClientSession#commitTransaction() committed} on success. Transactions are + * {@link ClientSession#abortTransaction() rolled back} upon errors. + * * @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); - } - }; - } + ReactiveSessionScoped inTransaction(Publisher sessionProvider); /** * 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 7f21da2fa..0ed3e3b11 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 @@ -461,6 +461,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati public Flux execute(String collectionName, ReactiveCollectionCallback callback) { Assert.notNull(callback, "ReactiveCollectionCallback must not be null!"); + return createFlux(collectionName, callback); } @@ -476,14 +477,50 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return new ReactiveSessionScoped() { @Override - public Flux flatMap(ReactiveSessionCallback action, Consumer doFinally) { + public Flux execute(ReactiveSessionCallback action, Consumer doFinally) { return cachedSession.flatMapMany(session -> { - return Flux + return ReactiveMongoTemplate.this.withSession(action, session) // + .doFinally(signalType -> { + doFinally.accept(session); + }); + }); + } + }; + } - .from(action.doInSession(new ReactiveSessionBoundMongoTemplate(session, ReactiveMongoTemplate.this))) // - .subscriberContext(ctx -> ReactiveMongoContext.setSession(ctx, cachedSession)) // + /* + * (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#inTransaction(org.reactivestreams.Publisher) + */ + @Override + public ReactiveSessionScoped inTransaction(Publisher sessionProvider) { + + Mono cachedSession = Mono.from(sessionProvider).cache(); + + return new ReactiveSessionScoped() { + + @Override + public Flux execute(ReactiveSessionCallback action, Consumer doFinally) { + + return cachedSession.flatMapMany(session -> { + + if (!session.hasActiveTransaction()) { + session.startTransaction(); + } + + return ReactiveMongoTemplate.this.withSession(action, session) // .materialize() // .flatMap(signal -> { @@ -506,14 +543,10 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati }; } - /* - * (non-Javadoc) - * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#inTransaction() - */ - @Override - public ReactiveSessionScoped inTransaction() { - return inTransaction( - mongoDatabaseFactory.getSession(ClientSessionOptions.builder().causallyConsistent(true).build())); + private Flux withSession(ReactiveSessionCallback action, ClientSession session) { + + return Flux.from(action.doInSession(new ReactiveSessionBoundMongoTemplate(session, ReactiveMongoTemplate.this))) // + .subscriberContext(ctx -> ReactiveMongoContext.setSession(ctx, Mono.just(session))); } /* 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 f25be4ddc..3e1568276 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,12 +16,9 @@ package org.springframework.data.mongodb.core; import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; import java.util.function.Consumer; -import org.reactivestreams.Publisher; - import com.mongodb.reactivestreams.client.ClientSession; /** @@ -42,7 +39,7 @@ public interface ReactiveSessionScoped { * * @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}. + * @return a result object returned by the action, can be {@link Flux#empty()}. */ default Flux execute(ReactiveSessionCallback action) { return execute(action, (session) -> {}); @@ -59,59 +56,7 @@ public interface ReactiveSessionScoped { * 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}. + * @return a result object returned by the action, can be {@link Flux#empty()}. */ - 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)); - } + Flux execute(ReactiveSessionCallback action, Consumer doFinally); } 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 9aa4f489b..f78301265 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 @@ -62,11 +62,15 @@ public class ReactiveClientSessionTests { template = new ReactiveMongoTemplate(client, DATABASE_NAME); - StepVerifier.create(MongoTestUtils.createOrReplaceCollection(DATABASE_NAME, COLLECTION_NAME, client)) - .expectNext(Success.SUCCESS).verifyComplete(); + MongoTestUtils.createOrReplaceCollection(DATABASE_NAME, COLLECTION_NAME, client) // + .as(StepVerifier::create) // + .expectNext(Success.SUCCESS) // + .verifyComplete(); - StepVerifier.create(template.insert(new Document("_id", "id-1").append("value", "spring"), COLLECTION_NAME)) - .expectNextCount(1).verifyComplete(); + template.insert(new Document("_id", "id-1").append("value", "spring"), COLLECTION_NAME) // + .as(StepVerifier::create) // + .expectNextCount(1) // + .verifyComplete(); } @Test // DATAMONGO-1880 @@ -77,8 +81,9 @@ public class ReactiveClientSessionTests { assertThat(session.getOperationTime()).isNull(); - StepVerifier - .create(template.withSession(() -> session).execute(action -> action.findAll(Document.class, COLLECTION_NAME))) + template.withSession(() -> session) // + .execute(action -> action.findAll(Document.class, COLLECTION_NAME)) // + .as(StepVerifier::create) // .expectNextCount(1).verifyComplete(); assertThat(session.getOperationTime()).isNotNull(); @@ -95,10 +100,11 @@ public class ReactiveClientSessionTests { assertThat(session.getOperationTime()).isNull(); - StepVerifier - .create(template.withSession(() -> session) - .execute(action -> action.findOne(new Query(), Document.class, COLLECTION_NAME))) - .expectNextCount(1).verifyComplete(); + template.withSession(() -> session) + .execute(action -> action.findOne(new Query(), Document.class, COLLECTION_NAME)) // + .as(StepVerifier::create) // + .expectNextCount(1) // + .verifyComplete(); assertThat(session.getOperationTime()).isNotNull(); assertThat(session.getServerSession().isClosed()).isFalse(); @@ -126,8 +132,10 @@ public class ReactiveClientSessionTests { 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(); + .execute(action -> ReactiveMongoContext.getSession()) // + .as(StepVerifier::create) // + .expectNextCount(1) // + .verifyComplete(); } static class CountingSessionSupplier implements Supplier { 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 index ccf585fc8..603566c30 100644 --- 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 @@ -15,10 +15,10 @@ */ package org.springframework.data.mongodb.core; +import static org.assertj.core.api.Assertions.*; 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; @@ -44,7 +44,10 @@ import com.mongodb.reactivestreams.client.MongoClient; import com.mongodb.reactivestreams.client.Success; /** + * Integration tests for Mongo Transactions using {@link ReactiveMongoTemplate}. + * * @author Christoph Strobl + * @author Mark Paluch * @currentRead The Core - Peter V. Brett */ public class ReactiveMongoTemplateTransactionTests { @@ -72,26 +75,32 @@ public class ReactiveMongoTemplateTransactionTests { 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, COLLECTION_NAME, client)) // + .expectNext(Success.SUCCESS) // + .verifyComplete(); StepVerifier.create(MongoTestUtils.createOrReplaceCollection(DATABASE_NAME, "person", client)) - .expectNext(Success.SUCCESS).verifyComplete(); + .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) + template.insertAll(Arrays.asList(AHMANN, ARLEN, LEESHA, RENNA)) // + .as(StepVerifier::create) // + .expectNextCount(4) // .verifyComplete(); } @Test // DATAMONGO-1970 - public void reactiveTransactionWithExplicitTransactionStart() throws InterruptedException { + public void reactiveTransactionWithExplicitTransactionStart() { Publisher sessionPublisher = client .startSession(ClientSessionOptions.builder().causallyConsistent(true).build()); - template.withSession(sessionPublisher) - .executeSingle(action -> ReactiveMongoContext.getSession().flatMap(session -> { + ClientSession clientSession = Mono.from(sessionPublisher).block(); + + template.withSession(Mono.just(clientSession)) + .execute(action -> ReactiveMongoContext.getSession().flatMap(session -> { session.startTransaction(); return action.remove(ID_QUERY, Document.class, COLLECTION_NAME); @@ -99,34 +108,84 @@ public class ReactiveMongoTemplateTransactionTests { })).as(StepVerifier::create).expectNextCount(1).verifyComplete(); template.exists(ID_QUERY, COLLECTION_NAME) // - .as(StepVerifier::create).expectNext(false).verifyComplete(); - } + .as(StepVerifier::create) // + .expectNext(true) // + .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(); + assertThat(clientSession.hasActiveTransaction()).isTrue(); + StepVerifier.create(clientSession.commitTransaction()).verifyComplete(); template.exists(ID_QUERY, COLLECTION_NAME) // - .as(StepVerifier::create).expectNext(false).verifyComplete(); + .as(StepVerifier::create) // + .expectNext(false) // + .verifyComplete(); } @Test // DATAMONGO-1970 - public void reactiveTransactionsAbortOnError() throws InterruptedException { + public void reactiveTransactionsCommitOnComplete() { + + 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() { 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(); + }).as(StepVerifier::create) // + .expectError() // + .verify(); template.exists(ID_QUERY, COLLECTION_NAME) // - .as(StepVerifier::create).expectNext(true).verifyComplete(); + .as(StepVerifier::create) // + .expectNext(true) // + .verifyComplete(); } @Test // DATAMONGO-1970 - public void changesNotVisibleOutsideTransaction() throws InterruptedException { + public void withSessionDoesNotManageTransactions() { + + Mono.from(client.startSession()).flatMap(session -> { + + session.startTransaction(); + return template.withSession(session).remove(ID_QUERY, Document.class, COLLECTION_NAME); + }).as(StepVerifier::create).expectNextCount(1).verifyComplete(); + + template.exists(ID_QUERY, COLLECTION_NAME) // + .as(StepVerifier::create) // + .expectNext(true) // + .verifyComplete(); + } + + @Test // DATAMONGO-1970 + public void inTransactionCommitsProvidedTransactionalSession() { + + ClientSession session = Mono.from(client.startSession()).block(); + + session.startTransaction(); + + template.inTransaction(Mono.just(session)).execute(action -> { + return action.remove(ID_QUERY, Document.class, COLLECTION_NAME); + }) // + .as(StepVerifier::create) // + .expectNextCount(1) // + .verifyComplete(); + + assertThat(session.hasActiveTransaction()).isFalse(); + } + + @Test // DATAMONGO-1970 + public void changesNotVisibleOutsideTransaction() { template.inTransaction().execute(action -> { return action.remove(ID_QUERY, Document.class, COLLECTION_NAME).flatMap(val -> { @@ -137,25 +196,60 @@ public class ReactiveMongoTemplateTransactionTests { }).as(StepVerifier::create).expectNext(DOCUMENT).verifyComplete(); template.exists(ID_QUERY, COLLECTION_NAME) // - .as(StepVerifier::create).expectNext(false).verifyComplete(); + .as(StepVerifier::create) // + .expectNext(false) // + .verifyComplete(); } @Test // DATAMONGO-1970 - public void takeDoesNotAbortTransaction() throws InterruptedException { + public void executeCreatesNewTransaction() { + + ReactiveSessionScoped sessionScoped = template.inTransaction(); + + sessionScoped.execute(action -> { + 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(); + + sessionScoped.execute(action -> { + return action.insert(DOCUMENT, COLLECTION_NAME); + }) // + .as(StepVerifier::create) // + .expectNextCount(1) // + .verifyComplete(); + + template.exists(ID_QUERY, COLLECTION_NAME) // + .as(StepVerifier::create) // + .expectNext(true) // + .verifyComplete(); + } + + @Test // DATAMONGO-1970 + public void takeDoesNotAbortTransaction() { 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(); + .flatMap(action::remove); + }) // + .as(StepVerifier::create) // + .expectNextCount(3) // + .verifyComplete(); template.count(query(where("age").exists(true)), Person.class) // - .as(StepVerifier::create).expectNext(1L).verifyComplete(); + .as(StepVerifier::create) // + .expectNext(1L) // + .verifyComplete(); } @Test // DATAMONGO-1970 - public void errorInFlowOutsideTransactionDoesNotAbortIt() throws InterruptedException { + public void errorInFlowOutsideTransactionDoesNotAbortIt() { template.inTransaction().execute(action -> { @@ -166,9 +260,14 @@ public class ReactiveMongoTemplateTransactionTests { }); }).flatMap(deleted -> { throw new RuntimeException("error outside the transaction does not influence it."); - }).as(StepVerifier::create).expectError().verify(); + }) // + .as(StepVerifier::create) // + .expectError() // + .verify(); template.count(query(where("age").exists(true)), Person.class) // - .as(StepVerifier::create).expectNext(2L).verifyComplete(); + .as(StepVerifier::create) // + .expectNext(2L) // + .verifyComplete(); } } diff --git a/src/main/asciidoc/reference/client-session-transactions.adoc b/src/main/asciidoc/reference/client-session-transactions.adoc index 8b9018fac..4f09388f9 100644 --- a/src/main/asciidoc/reference/client-session-transactions.adoc +++ b/src/main/asciidoc/reference/client-session-transactions.adoc @@ -88,6 +88,8 @@ By using a `Publisher` that provides the actual session, you can defer session a Still, you need to close the session when done, so as to not pollute the server with stale sessions. Use the `doFinally` hook on `execute` to call `ClientSession#close()` when you no longer need the session. If you prefer having more control over the session itself, you can obtain the `ClientSession` through the driver and provide it through a `Supplier`. +NOTE: Reactive use of `ClientSession` is limited to Template API usage. There's currently no session integration with reactive repositories. + [[mongo.transactions]] = MongoDB Transactions @@ -219,7 +221,9 @@ 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. -Using the plain MongoDB reactive driver API a `delete within a transactional flow may look like this. +NOTE: Reactive use of `ClientSession` and transactions is limited to Template API usage. There's currently no session or transaction integration with reactive repositories. + +Using the plain MongoDB reactive driver API a `delete` within a transactional flow may look like this. .Native driver support ==== @@ -227,21 +231,26 @@ Using the plain MongoDB reactive driver API a `delete within a transactional flo ---- 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. +<1> First we obviously need to initiate the session. +<2> Once we have 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. +<4> If the operations completes exceptionally, 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. -====` +<6> Lastly, 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. @@ -250,35 +259,41 @@ published via either `commitTransaction()` or `abortTransaction()`, which leads 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 +.`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`. +<2> Operate within the `ClientSession`. Each `execute(…)` unit of work callback initiates a new transaction in the scope of the same `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. +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. ==== [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(); +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(updated -> { + // Exception could happen here <3> + }); ---- -<1> Initiate the transaction. -<2> Operate within the `ClientSession`. The transaction is committed after when this is done or rolled back if an +<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. ====