From 31eeeaf0f0b28acb105136191b1a1146760f9627 Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Thu, 8 Aug 2019 07:58:54 +0200 Subject: [PATCH] #517 - Add example for declarative reactive transactions in MongoDB. Original pull request: #518. --- mongodb/transactions/README.md | 50 ++++++-- .../ReactiveManagedTransitionService.java | 84 +++++++++++++ ...ReactiveManagedTransitionServiceTests.java | 117 ++++++++++++++++++ 3 files changed, 243 insertions(+), 8 deletions(-) create mode 100644 mongodb/transactions/src/main/java/example/springdata/mongodb/reactive/ReactiveManagedTransitionService.java create mode 100644 mongodb/transactions/src/test/java/example/springdata/mongodb/reactive/ReactiveManagedTransitionServiceTests.java diff --git a/mongodb/transactions/README.md b/mongodb/transactions/README.md index f6aff1d2..cd5bfc38 100644 --- a/mongodb/transactions/README.md +++ b/mongodb/transactions/README.md @@ -15,7 +15,7 @@ be patient. ## Sync Transactions `MongoTransactionManager` is the gateway to the well known Spring transaction support. It lets applications use -[the managed transaction features of Spring](http://docs.spring.io/spring/docs/{springVersion}/spring-framework-reference/html/transaction.html). +[the managed transaction features of Spring](http://docs.spring.io/spring/docs/{springVersion}/spring-framework-reference/data-access.html#transaction). The `MongoTransactionManager` binds a `ClientSession` to the thread. `MongoTemplate` detects the session and operates on these resources which are associated with the transaction accordingly. `MongoTemplate` can also participate in other, ongoing transactions. @@ -51,17 +51,16 @@ public class TransitionService { } ``` -## Reactive transactions +## Programmatic Reactive transactions -`ReactiveMongoTemplate` offers dedicated methods for operating within a transaction without having to worry about the -commit/abort actions depending on the operations outcome. There's currently no session or transaction integration -with reactive repositories - we apologize for that! +`ReactiveMongoTemplate` offers dedicated methods (like `inTransaction()`) for operating within a transaction without having to worry about the +commit/abort actions depending on the operations outcome. **NOTE:** Please note that you cannot preform meta operations, like collection creation within a transaction. ```java -@Component -public class RactiveTransitionService { +@Service +public class ReactiveTransitionService { public Mono run(Integer id) { @@ -76,4 +75,39 @@ public class RactiveTransitionService { }).next().map(Process::getId); } } -``` \ No newline at end of file +``` + +## Declarative Reactive transactions + +`ReactiveMongoTransactionManager` is the gateway to the reactive Spring transaction support. It lets applications use +[the managed transaction features of Spring](http://docs.spring.io/spring/docs/{springVersion}/spring-framework-reference/data-access.html#transaction). +The `ReactiveMongoTransactionManager` adds the `ClientSession` to the `reactor.util.context.Context`. `ReactiveMongoTemplate` detects the session and operates +on these resources which are associated with the transaction accordingly. + +```java +@EnableTransactionManagement +class Config extends AbstractReactiveMongoConfiguration { + + @Bean + ReactiveTransactionManager transactionManager(ReactiveMongoDatabaseFactory factory) { + return new ReactiveMongoTransactionManager(factory); + } + + // ... +} + + +@Service +class ReactiveManagedTransitionService { + + @Transactional + public Mono run(Integer id) { + + return lookup(id) + .flatMap(process -> start(template, process)) + .flatMap(it -> verify(it)) // + .flatMap(process -> finish(template, process)) + .map(Process::getId); + } +} +``` diff --git a/mongodb/transactions/src/main/java/example/springdata/mongodb/reactive/ReactiveManagedTransitionService.java b/mongodb/transactions/src/main/java/example/springdata/mongodb/reactive/ReactiveManagedTransitionService.java new file mode 100644 index 00000000..674732eb --- /dev/null +++ b/mongodb/transactions/src/main/java/example/springdata/mongodb/reactive/ReactiveManagedTransitionService.java @@ -0,0 +1,84 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package example.springdata.mongodb.reactive; + +import example.springdata.mongodb.Process; +import example.springdata.mongodb.State; +import lombok.RequiredArgsConstructor; +import reactor.core.publisher.Mono; + +import java.util.concurrent.atomic.AtomicInteger; + +import org.springframework.data.mongodb.core.ReactiveMongoOperations; +import org.springframework.data.mongodb.core.ReactiveMongoTemplate; +import org.springframework.data.mongodb.core.query.Criteria; +import org.springframework.data.mongodb.core.query.Query; +import org.springframework.data.mongodb.core.query.Update; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.util.Assert; + +/** + * @author Christoph Strobl + */ +@Service +@RequiredArgsConstructor +public class ReactiveManagedTransitionService { + + final ReactiveProcessRepository repository; + final ReactiveMongoTemplate template; + + final AtomicInteger counter = new AtomicInteger(0); + + public Mono newProcess() { + return repository.save(new Process(counter.incrementAndGet(), State.CREATED, 0)); + } + + @Transactional + public Mono run(Integer id) { + + return lookup(id) // + .flatMap(process -> start(template, process)) // + .flatMap(it -> verify(it)) // + .flatMap(process -> finish(template, process)) // + .map(Process::getId); + } + + private Mono finish(ReactiveMongoOperations operations, Process process) { + + return operations.update(Process.class).matching(Query.query(Criteria.where("id").is(process.getId()))) + .apply(Update.update("state", State.DONE).inc("transitionCount", 1)).first() // + .then(Mono.just(process)); + } + + Mono start(ReactiveMongoOperations operations, Process process) { + + return operations.update(Process.class).matching(Query.query(Criteria.where("id").is(process.getId()))) + .apply(Update.update("state", State.ACTIVE).inc("transitionCount", 1)).first() // + .then(Mono.just(process)); + } + + Mono lookup(Integer id) { + return repository.findById(id); + } + + Mono verify(Process process) { + + Assert.state(process.getId() % 3 != 0, "We're sorry but we needed to drop that one"); + return Mono.just(process); + } + +} diff --git a/mongodb/transactions/src/test/java/example/springdata/mongodb/reactive/ReactiveManagedTransitionServiceTests.java b/mongodb/transactions/src/test/java/example/springdata/mongodb/reactive/ReactiveManagedTransitionServiceTests.java new file mode 100644 index 00000000..bb0aaf02 --- /dev/null +++ b/mongodb/transactions/src/test/java/example/springdata/mongodb/reactive/ReactiveManagedTransitionServiceTests.java @@ -0,0 +1,117 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package example.springdata.mongodb.reactive; + +import static org.assertj.core.api.Assertions.*; + +import example.springdata.mongodb.Process; +import example.springdata.mongodb.State; +import example.springdata.mongodb.util.EmbeddedMongo; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +import org.bson.Document; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory; +import org.springframework.data.mongodb.ReactiveMongoTransactionManager; +import org.springframework.data.mongodb.config.AbstractReactiveMongoConfiguration; +import org.springframework.data.mongodb.repository.config.EnableReactiveMongoRepositories; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.transaction.ReactiveTransactionManager; +import org.springframework.transaction.annotation.EnableTransactionManagement; + +import com.mongodb.reactivestreams.client.MongoClient; +import com.mongodb.reactivestreams.client.MongoClients; + +/** + * Test showing MongoDB Transaction usage through a reactive API. + * + * @author Christoph Strobl + * @currentRead The Core - Peter V. Brett + */ +@RunWith(SpringRunner.class) +@ContextConfiguration +public class ReactiveManagedTransitionServiceTests { + + public static @ClassRule EmbeddedMongo replSet = EmbeddedMongo.replSet().configure(); + + @Autowired ReactiveManagedTransitionService managedTransitionService; + @Autowired MongoClient client; + + static final String DB_NAME = "spring-data-reactive-tx-examples"; + + @Configuration + @ComponentScan + @EnableReactiveMongoRepositories + @EnableTransactionManagement + static class Config extends AbstractReactiveMongoConfiguration { + + @Bean + ReactiveTransactionManager transactionManager(ReactiveMongoDatabaseFactory dbFactory) { + return new ReactiveMongoTransactionManager(dbFactory); + } + + @Bean + @Override + public MongoClient reactiveMongoClient() { + return MongoClients.create(replSet.getConnectionString()); + } + + @Override + protected String getDatabaseName() { + return DB_NAME; + } + } + + @Test + public void reactiveTxCommitRollback() { + + for (int i = 0; i < 10; i++) { + managedTransitionService.newProcess() // + .map(Process::getId) // + .flatMap(managedTransitionService::run) // + .onErrorReturn(-1).as(StepVerifier::create) // + .consumeNextWith(val -> {}) // + .verifyComplete(); + } + + Flux.from(client.getDatabase(DB_NAME).getCollection("processes").find(new Document())) // + .buffer(10) // + .as(StepVerifier::create) // + .consumeNextWith(list -> { + + for (Document document : list) { + + System.out.println("document: " + document); + + if (document.getInteger("_id") % 3 == 0) { + assertThat(document.getString("state")).isEqualTo(State.CREATED.toString()); + } else { + assertThat(document.getString("state")).isEqualTo(State.DONE.toString()); + } + } + + }) // + .verifyComplete(); + } +}