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

DATAMONGO-1970 - Polishing.

Browse Source 
ReactiveMongoOperations.withSession(…) no longer commits transactions if a transaction is active. ReactiveSessionScoped obtained through inTransaction() solely manages transactions and participates in ongoing transactions if a given ClientSession has already an active transaction. Remove ReactiveSessionScoped.executeSingle methods to align with ReactiveMongoOperations.

Add tests. Switch reactive tests to .as(StepVerifier:create) form. Extend documentation.

Original pull request: #560.
This commit is contained in:
Mark Paluch
2018-05-14 11:22:53 +02:00
parent f296a499e5
commit a66b87118e
7 changed files with 260 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoContext.java
View File

@@ -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<ClientSession>} from Reactor {@link reactor.util.context.Context}
* Gets the {@code Mono<ClientSession>} 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<ClientSession> getSession() {
public 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}
* 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<ClientSession> session) {
return context.put(SESSION_KEY, session);
public static Context setSession(Context context, Publisher<ClientSession> 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));
}
}

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

@@ -198,7 +198,7 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
* 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
@@ -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.
* <p/>
* 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.
*
* <p/>
* 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<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);
}
};
}
ReactiveSessionScoped inTransaction(Publisher<ClientSession> sessionProvider);
/**
* Create an uncapped collection with a name based on the provided entity class.

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

@@ -461,6 +461,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
public <T> Flux<T> execute(String collectionName, ReactiveCollectionCallback<T> 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 <T> Flux<T> flatMap(ReactiveSessionCallback<T> action, Consumer<ClientSession> doFinally) {
public <T> Flux<T> execute(ReactiveSessionCallback<T> action, Consumer<ClientSession> 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<ClientSession> sessionProvider) {
Mono<ClientSession> cachedSession = Mono.from(sessionProvider).cache();
return new ReactiveSessionScoped() {
@Override
public <T> Flux<T> execute(ReactiveSessionCallback<T> action, Consumer<ClientSession> 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 <T> Flux<T> withSession(ReactiveSessionCallback<T> action, ClientSession session) {
return Flux.from(action.doInSession(new ReactiveSessionBoundMongoTemplate(session, ReactiveMongoTemplate.this))) //
.subscriberContext(ctx -> ReactiveMongoContext.setSession(ctx, Mono.just(session)));
}
/*

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

@@ -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 <T> 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 <T> Flux<T> execute(ReactiveSessionCallback<T> 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 <T> 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 <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));
}
<T> Flux<T> execute(ReactiveSessionCallback<T> action, Consumer<ClientSession> doFinally);
}

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

@@ -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<ClientSession> {

159
spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTransactionTests.java
View File

@@ -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<ClientSession> 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();
}
}

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

@@ -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<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.
<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<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`.
<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.
====