#361 - Add example for MongoDB 4.0 transactions.

README.md

@@ -31,6 +31,7 @@ We have separate folders for the samples of individual modules:
 * `reactive` - Example project to show reactive template and repository support.
 * `security` - Example project showing usage of Spring Security with MongoDB.
 * `text-search` - Example project showing usage of MongoDB text search feature.
+* `transactions` - Example project for synchronous and reactive MongoDB 4.0 transaction support.
 
 ## Spring Data REST
 

mongodb/pom.xml

@@ -27,6 +27,7 @@
 		<module>reactive</module>
 		<module>security</module>
 		<module>text-search</module>
+		<module>transactions</module>
 	</modules>
 
 	<dependencies>

mongodb/transactions/README.md

@@ -0,0 +1,79 @@
+# Spring Data MongoDB - Transaction Sample
+
+This project contains samples for upcoming MongoDB 4.0 transactions.
+
+## Running the Sample
+
+The sample uses multiple embedded MongoDB processes in a [MongoDB replica set](https://docs.mongodb.com/manual/replication/).
+It contains test for both the synchronous and the reactive transaction support in the `sync` / `reactive` packages.   
+
+You may run the examples directly from your IDE or use maven on the command line.
+
+**INFO:** The operations to download the required MongoDB binaries and spin up the cluster can take some time. Please 
+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 `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.
+
+```java
+@Configuration
+static class Config extends AbstractMongoConfiguration {
+
+	@Bean
+	MongoTransactionManager transactionManager(MongoDbFactory dbFactory) {
+		return new MongoTransactionManager(dbFactory);
+	}
+
+	// ...
+}
+
+@Component
+public class TransitionService {
+
+    @Transactional
+    public void run(Integer id) {
+    
+    	Process process = lookup(id);
+    
+    	if (!State.CREATED.equals(process.getState())) {
+    		return;
+    	}
+    
+    	start(process);
+    	verify(process);
+    	finish(process);
+    }
+}
+```
+
+## 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!
+
+**NOTE:** Please note that you cannot preform meta operations, like collection creation within a transaction.
+
+```java
+@Component
+public class RactiveTransitionService {
+	
+    public Mono<Integer> run(Integer id) {
+
+		return template.inTransaction().execute(action -> {
+
+			return lookup(id) //
+			        .filter(State.CREATED::equals)
+			        .flatMap(process -> start(action, process))
+			        .flatMap(this::verify)
+			        .flatMap(process -> finish(action, process));
+
+		}).next().map(Process::getId);
+	}
+}
+```

mongodb/transactions/pom.xml

@@ -0,0 +1,72 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.springframework.data.examples</groupId>
+        <artifactId>spring-data-mongodb-examples</artifactId>
+        <version>2.0.0.BUILD-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>spring-data-mongodb-transactions</artifactId>
+    <name>Spring Data MongoDB - Transactions</name>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.mongodb</groupId>
+                    <artifactId>mongodb-driver</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.data</groupId>
+            <artifactId>spring-data-mongodb</artifactId>
+            <version>2.1.0.BUILD-SNAPSHOT</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.data</groupId>
+            <artifactId>spring-data-commons</artifactId>
+            <version>2.1.0.BUILD-SNAPSHOT</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mongodb</groupId>
+            <artifactId>mongo-java-driver</artifactId>
+            <version>3.8.0-beta2</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mongodb</groupId>
+            <artifactId>mongodb-driver-async</artifactId>
+            <version>3.8.0-beta2</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mongodb</groupId>
+            <artifactId>mongodb-driver-reactivestreams</artifactId>
+            <version>1.9.0-beta1</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-core</artifactId>
+            <version>3.1.7.RELEASE</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <version>3.1.7.RELEASE</version>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+
+</project>

mongodb/transactions/src/main/java/example/springdata/mongodb/Process.java

@@ -0,0 +1,37 @@
+/*
+ * 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 example.springdata.mongodb;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+/**
+ * @author Christoph Strobl
+ * @currentRead The Core - Peter V. Brett
+ */
+@Data
+@AllArgsConstructor
+@Document("processes")
+public class Process {
+
+	@Id Integer id;
+	State state;
+	int transitionCount;
+
+}

mongodb/transactions/src/main/java/example/springdata/mongodb/State.java

@@ -0,0 +1,25 @@
+/*
+ * 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 example.springdata.mongodb;
+
+/**
+ * @author Christoph Strobl
+ * @currentRead The Core - Peter V. Brett
+ */
+public enum State {
+
+	CREATED, ACTIVE, DONE
+}

mongodb/transactions/src/main/java/example/springdata/mongodb/reactive/ReactiveProcessRepository.java

@@ -0,0 +1,28 @@
+/*
+ * 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 example.springdata.mongodb.reactive;
+
+import example.springdata.mongodb.Process;
+
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+
+/**
+ * @author Christoph Strobl
+ * @currentRead The Core - Peter V. Brett
+ */
+public interface ReactiveProcessRepository extends ReactiveCrudRepository<Process, Integer> {
+
+}

mongodb/transactions/src/main/java/example/springdata/mongodb/reactive/ReactiveTransitionService.java

@@ -0,0 +1,85 @@
+/*
+ * 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 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.util.Assert;
+
+/**
+ * @author Christoph Strobl
+ * @currentRead The Core - Peter V. Brett
+ */
+@Service
+@RequiredArgsConstructor
+public class ReactiveTransitionService {
+
+	final ReactiveProcessRepository repository;
+	final ReactiveMongoTemplate template;
+
+	final AtomicInteger counter = new AtomicInteger(0);
+
+	public Mono<Process> newProcess() {
+		return repository.save(new Process(counter.incrementAndGet(), State.CREATED, 0));
+	}
+
+	public Mono<Integer> run(Integer id) {
+
+		return template.inTransaction().execute(action -> {
+
+			return lookup(id) //
+					.flatMap(process -> start(action, process)) //
+					.flatMap(this::verify) //
+					.flatMap(process -> finish(action, process));
+
+		}).next().map(Process::getId);
+	}
+
+	private Mono<Process> 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<Process> 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<Process> lookup(Integer id) {
+		return repository.findById(id);
+	}
+
+	Mono<Process> verify(Process process) {
+
+		Assert.state(process.getId() % 3 != 0, "We're sorry but we needed to drop that one");
+		return Mono.just(process);
+	}
+}

mongodb/transactions/src/main/java/example/springdata/mongodb/sync/ProcessRepository.java

@@ -0,0 +1,28 @@
+/*
+ * 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 example.springdata.mongodb.sync;
+
+import example.springdata.mongodb.Process;
+
+import org.springframework.data.repository.CrudRepository;
+
+/**
+ * @author Christoph Strobl
+ * @currentRead The Core - Peter V. Brett
+ */
+interface ProcessRepository extends CrudRepository<Process, Integer> {
+
+}

mongodb/transactions/src/main/java/example/springdata/mongodb/sync/TransitionService.java

@@ -0,0 +1,82 @@
+/*
+ * 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 example.springdata.mongodb.sync;
+
+import example.springdata.mongodb.Process;
+import example.springdata.mongodb.State;
+import lombok.RequiredArgsConstructor;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.springframework.data.mongodb.core.MongoTemplate;
+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
+ * @currentRead The Core - Peter V. Brett
+ */
+@Service
+@RequiredArgsConstructor
+class TransitionService {
+
+	final ProcessRepository repository;
+	final MongoTemplate template;
+
+	final AtomicInteger counter = new AtomicInteger(0);
+
+	public Process newProcess() {
+		return repository.save(new Process(counter.incrementAndGet(), State.CREATED, 0));
+	}
+
+	@Transactional
+	public void run(Integer id) {
+
+		Process process = lookup(id);
+
+		if (!State.CREATED.equals(process.getState())) {
+			return;
+		}
+
+		start(process);
+		verify(process);
+		finish(process);
+	}
+
+	private void finish(Process process) {
+
+		template.update(Process.class).matching(Query.query(Criteria.where("id").is(process.getId())))
+				.apply(Update.update("state", State.DONE).inc("transitionCount", 1)).first();
+	}
+
+	void start(Process process) {
+
+		template.update(Process.class).matching(Query.query(Criteria.where("id").is(process.getId())))
+				.apply(Update.update("state", State.ACTIVE).inc("transitionCount", 1)).first();
+	}
+
+	Process lookup(Integer id) {
+		return repository.findById(id).get();
+	}
+
+	void verify(Process process) {
+		Assert.state(process.getId() % 3 != 0, "We're sorry but we needed to drop that one");
+	}
+}

mongodb/transactions/src/test/java/example/springdata/mongodb/reactive/ReactiveTransitionServiceTests.java

@@ -0,0 +1,105 @@
+/*
+ * 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 example.springdata.mongodb.reactive;
+
+import static org.assertj.core.api.Assertions.*;
+
+import example.springdata.mongodb.Process;
+import example.springdata.mongodb.State;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+import utils.EmbeddedMongo;
+
+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.config.AbstractReactiveMongoConfiguration;
+import org.springframework.data.mongodb.repository.config.EnableReactiveMongoRepositories;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import com.mongodb.reactivestreams.client.MongoClient;
+import com.mongodb.reactivestreams.client.MongoClients;
+
+/**
+ * @author Christoph Strobl
+ * @currentRead The Core - Peter V. Brett
+ */
+@RunWith(SpringRunner.class)
+@ContextConfiguration
+public class ReactiveTransitionServiceTests {
+
+	public static @ClassRule EmbeddedMongo replSet = EmbeddedMongo.replSet().configure();
+
+	@Autowired ReactiveTransitionService transitionService;
+	@Autowired MongoClient client;
+
+	static final String DB_NAME = "spring-data-reactive-tx-examples";
+
+	@Configuration
+	@ComponentScan
+	@EnableReactiveMongoRepositories
+	static class Config extends AbstractReactiveMongoConfiguration {
+
+		@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++) {
+			transitionService.newProcess() //
+					.map(Process::getId) //
+					.flatMap(transitionService::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();
+	}
+}

mongodb/transactions/src/test/java/example/springdata/mongodb/sync/TransitionServiceTests.java

@@ -0,0 +1,110 @@
+/*
+ * 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 example.springdata.mongodb.sync;
+
+import example.springdata.mongodb.Process;
+import example.springdata.mongodb.State;
+import utils.EmbeddedMongo;
+
+import java.util.function.Consumer;
+
+import org.assertj.core.api.Assertions;
+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.MongoDbFactory;
+import org.springframework.data.mongodb.MongoTransactionManager;
+import org.springframework.data.mongodb.config.AbstractMongoConfiguration;
+import org.springframework.data.mongodb.repository.config.EnableMongoRepositories;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
+
+import com.mongodb.MongoClient;
+import com.mongodb.client.model.Filters;
+import com.mongodb.client.model.Projections;
+
+/**
+ * @author Christoph Strobl
+ * @currentRead The Core - Peter V. Brett
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration
+public class TransitionServiceTests {
+
+	public static @ClassRule EmbeddedMongo replSet = EmbeddedMongo.replSet().configure();
+
+	static final String DB_NAME = "spring-data-tx-examples";
+
+	@Autowired TransitionService transitionService;
+	@Autowired com.mongodb.MongoClient client;
+
+	@Configuration
+	@ComponentScan
+	@EnableMongoRepositories
+	@EnableTransactionManagement
+	static class Config extends AbstractMongoConfiguration {
+
+		@Bean
+		PlatformTransactionManager transactionManager(MongoDbFactory dbFactory) {
+			return new MongoTransactionManager(dbFactory);
+		}
+
+		@Override
+		@Bean
+		public MongoClient mongoClient() {
+			return replSet.getMongoClient();
+		}
+
+		@Override
+		protected String getDatabaseName() {
+			return DB_NAME;
+		}
+	}
+
+	@Test
+	public void txCommitRollback() {
+
+		for (int i = 0; i < 10; i++) {
+
+			Process process = transitionService.newProcess();
+
+			try {
+
+				transitionService.run(process.getId());
+				Assertions.assertThat(stateInDb(process)).isEqualTo(State.DONE);
+			} catch (IllegalStateException e) {
+				Assertions.assertThat(stateInDb(process)).isEqualTo(State.CREATED);
+			}
+		}
+
+		client.getDatabase(DB_NAME).getCollection("processes").find(new Document())
+				.forEach((Consumer<? super Document>) System.out::println);
+	}
+
+	State stateInDb(Process process) {
+
+		return State.valueOf(client.getDatabase(DB_NAME).getCollection("processes").find(Filters.eq("_id", process.getId()))
+				.projection(Projections.include("state")).first().get("state", String.class));
+	}
+
+}

mongodb/transactions/src/test/java/utils/EmbeddedMongo.java

@@ -0,0 +1,340 @@
+/*
+ * 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 utils;
+
+import de.flapdoodle.embed.mongo.config.IMongoCmdOptions;
+import de.flapdoodle.embed.mongo.config.IMongodConfig;
+import de.flapdoodle.embed.mongo.config.IMongosConfig;
+import de.flapdoodle.embed.mongo.config.MongoCmdOptionsBuilder;
+import de.flapdoodle.embed.mongo.config.MongodConfigBuilder;
+import de.flapdoodle.embed.mongo.config.MongosConfigBuilder;
+import de.flapdoodle.embed.mongo.config.Net;
+import de.flapdoodle.embed.mongo.config.Storage;
+import de.flapdoodle.embed.mongo.distribution.Feature;
+import de.flapdoodle.embed.mongo.distribution.IFeatureAwareVersion;
+import de.flapdoodle.embed.mongo.distribution.Versions;
+import de.flapdoodle.embed.mongo.tests.MongosSystemForTestFactory;
+import de.flapdoodle.embed.process.distribution.GenericVersion;
+import de.flapdoodle.embed.process.runtime.Network;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.rules.ExternalResource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
+
+import com.mongodb.MongoClient;
+import com.mongodb.MongoClientURI;
+
+/**
+ * @author Christoph Strobl
+ */
+public class EmbeddedMongo extends ExternalResource {
+
+	private static final Logger LOGGER = LoggerFactory.getLogger(EmbeddedMongo.class);
+
+	private static final String LOCALHOST = "127.0.0.1";
+	private static final String DEFAULT_REPLICA_SET_NAME = "rs0";
+	private static final String DEFAULT_CONFIG_SERVER_REPLICA_SET_NAME = "rs-config";
+
+	private static final String STORAGE_ENGINE = "wiredTiger";
+
+	private static final IFeatureAwareVersion VERSION = Versions.withFeatures(new GenericVersion("3.7.9"),
+			Feature.ONLY_WITH_SSL, Feature.ONLY_64BIT, Feature.NO_HTTP_INTERFACE_ARG, Feature.STORAGE_ENGINE,
+			Feature.MONGOS_CONFIGDB_SET_STYLE, Feature.NO_CHUNKSIZE_ARG);
+
+	private final TestResource resource;
+
+	private EmbeddedMongo(TestResource resource) {
+		this.resource = resource;
+	}
+
+	public static Builder builder() {
+		return new Builder();
+	}
+
+	public static Builder replSet() {
+		return replSet(DEFAULT_REPLICA_SET_NAME);
+	}
+
+	public static Builder replSet(String replicaSetName) {
+		return new Builder().withReplicaSetName(replicaSetName);
+	}
+
+	public static class Builder {
+
+		IFeatureAwareVersion version;
+		String replicaSetName;
+		List<Integer> serverPorts;
+		List<Integer> configServerPorts;
+
+		Builder() {
+
+			version = VERSION;
+			replicaSetName = null;
+			serverPorts = Collections.emptyList();
+			configServerPorts = Collections.emptyList();
+		}
+
+		public Builder withVersion(IFeatureAwareVersion version) {
+
+			this.version = version;
+			return this;
+		}
+
+		public Builder withReplicaSetName(String replicaSetName) {
+
+			this.replicaSetName = replicaSetName;
+			return this;
+		}
+
+		public Builder withServerPorts(Integer... ports) {
+
+			this.serverPorts = Arrays.asList(ports);
+			return this;
+		}
+
+		public EmbeddedMongo configure() {
+
+			if (serverPorts.size() > 1 || StringUtils.hasText(replicaSetName)) {
+
+				String rsName = StringUtils.hasText(replicaSetName) ? replicaSetName : DEFAULT_REPLICA_SET_NAME;
+				return new EmbeddedMongo(new ReplSet(version, rsName, serverPorts.toArray(new Integer[serverPorts.size()])));
+			}
+
+			throw new UnsupportedOperationException("implement me");
+		}
+
+	}
+
+	@Override
+	protected void before() throws Throwable {
+		resource.start();
+	}
+
+	@Override
+	protected void after() {
+		resource.stop();
+	}
+
+	public MongoClient getMongoClient() {
+		return resource.mongoClient();
+	}
+
+	public String getConnectionString() {
+		return resource.connectionString();
+	}
+
+	private static Integer randomOrDefaultServerPort() {
+
+		try {
+			return Network.getFreeServerPort();
+		} catch (IOException e) {
+			return 27017;
+		}
+	}
+
+	interface TestResource {
+
+		void start();
+
+		void stop();
+
+		String connectionString();
+
+		default MongoClient mongoClient() {
+			return new MongoClient(new MongoClientURI(connectionString()));
+		}
+	}
+
+	static class ReplSet implements TestResource {
+
+		private static final String DEFAULT_SHARDING = "none";
+		private static final String DEFAULT_SHARD_KEY = "_class";
+
+		private final IFeatureAwareVersion serverVersion;
+		private final String configServerReplicaSetName;
+		private final String replicaSetName;
+		private final int mongosPort;
+		private final Integer[] serverPorts;
+		private final Integer[] configServerPorts;
+
+		private MongosSystemForTestFactory mongosTestFactory;
+
+		ReplSet(IFeatureAwareVersion serverVersion, String replicaSetName, Integer... serverPorts) {
+
+			this.serverVersion = serverVersion;
+			this.replicaSetName = replicaSetName;
+			this.serverPorts = defaultPortsIfRequired(serverPorts);
+			this.configServerPorts = defaultPortsIfRequired(null);
+			this.configServerReplicaSetName = DEFAULT_CONFIG_SERVER_REPLICA_SET_NAME;
+			this.mongosPort = randomOrDefaultServerPort();
+		}
+
+		Integer[] defaultPortsIfRequired(Integer[] ports) {
+
+			if (!ObjectUtils.isEmpty(ports)) {
+				return ports;
+			}
+
+			try {
+				return new Integer[] { Network.getFreeServerPort(), Network.getFreeServerPort(), Network.getFreeServerPort() };
+			} catch (IOException e) {
+				throw new RuntimeException(e);
+			}
+		}
+
+		@Override
+		public void start() {
+
+			if (mongosTestFactory != null) {
+				return;
+			}
+
+			doStart();
+		}
+
+		private void doStart() {
+
+			Map<String, List<IMongodConfig>> replicaSets = new LinkedHashMap<>();
+			replicaSets.put(configServerReplicaSetName, initConfigServers());
+			replicaSets.put(replicaSetName, initReplicaSet());
+
+			// create mongos
+			IMongosConfig mongosConfig = defaultMongosConfig(serverVersion, mongosPort, defaultCommandOptions(),
+					configServerReplicaSetName, configServerPorts[0]);
+
+			mongosTestFactory = new MongosSystemForTestFactory(mongosConfig, replicaSets, Collections.emptyList(),
+					DEFAULT_SHARDING, DEFAULT_SHARDING, DEFAULT_SHARD_KEY);
+
+			try {
+				LOGGER.info(String.format("Starting config servers at ports %s",
+						StringUtils.arrayToCommaDelimitedString(configServerPorts)));
+				LOGGER.info(String.format("Starting replica set '%s' servers at ports %s", replicaSetName,
+						StringUtils.arrayToCommaDelimitedString(serverPorts)));
+
+				mongosTestFactory.start();
+
+				LOGGER
+						.info(String.format("Replica set '%s' started. Connection String: %s", replicaSetName, connectionString()));
+			} catch (Throwable e) {
+				throw new RuntimeException(" Error while starting cluster. ", e);
+			}
+		}
+
+		private List<IMongodConfig> initReplicaSet() {
+			List<IMongodConfig> replicaSet1 = new ArrayList<>();
+
+			for (int port : serverPorts) {
+				replicaSet1.add(defaultMongodConfig(serverVersion, port, defaultCommandOptions(), false, true, replicaSetName));
+			}
+			return replicaSet1;
+		}
+
+		private List<IMongodConfig> initConfigServers() {
+			List<IMongodConfig> configServers = new ArrayList<>(configServerPorts.length);
+
+			for (Integer port : configServerPorts) {
+				configServers.add(
+						defaultMongodConfig(serverVersion, port, defaultCommandOptions(), true, false, configServerReplicaSetName));
+			}
+			return configServers;
+		}
+
+		@Override
+		public void stop() {
+
+			if (mongosTestFactory != null) {
+
+				LOGGER.info(String.format("Stopping replica set '%s' servers at ports %s", replicaSetName,
+						StringUtils.arrayToCommaDelimitedString(serverPorts)));
+
+				mongosTestFactory.stop();
+			}
+		}
+
+		@Override
+		public String connectionString() {
+			return "mongodb://localhost:" + serverPorts[0] + "/?replicaSet=" + replicaSetName;
+		}
+	}
+
+	private static IMongoCmdOptions defaultCommandOptions() {
+
+		return new MongoCmdOptionsBuilder() //
+				.useNoPrealloc(false) //
+				.useSmallFiles(false) //
+				.useNoJournal(false) //
+				.useStorageEngine(STORAGE_ENGINE) //
+				.verbose(false) //
+				.build();
+	}
+
+	private static IMongodConfig defaultMongodConfig(IFeatureAwareVersion version, int port, IMongoCmdOptions cmdOptions,
+			boolean configServer, boolean shardServer, String replicaSet) {
+
+		try {
+			MongodConfigBuilder builder = new MongodConfigBuilder() //
+					.version(version) //
+					.net(new Net(LOCALHOST, port, Network.localhostIsIPv6())) //
+					.configServer(configServer).cmdOptions(cmdOptions); //
+
+			if (StringUtils.hasText(replicaSet)) {
+
+				builder = builder //
+						.replication(new Storage(null, replicaSet, 0));
+
+				if (!configServer) {
+					builder = builder.shardServer(shardServer);
+				} else {
+					builder = builder.shardServer(false);
+				}
+			}
+
+			return builder.build();
+		} catch (IOException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	private static IMongosConfig defaultMongosConfig(IFeatureAwareVersion version, int port, IMongoCmdOptions cmdOptions,
+			String configServerReplicaSet, int configServerPort) {
+
+		try {
+			MongosConfigBuilder builder = new MongosConfigBuilder() //
+					.version(version) //
+					.net(new Net(LOCALHOST, port, Network.localhostIsIPv6())) //
+					.cmdOptions(cmdOptions);
+
+			if (StringUtils.hasText(configServerReplicaSet)) {
+				builder = builder.replicaSet(configServerReplicaSet) //
+						.configDB(LOCALHOST + ":" + configServerPort);
+			}
+
+			return builder.build();
+		} catch (IOException e) {
+			throw new RuntimeException(e);
+		}
+	}
+}

mongodb/transactions/src/test/resources/logback.xml

@@ -0,0 +1,17 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+
+    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d %5p %40.40c:%4L - %m%n</pattern>
+        </encoder>
+    </appender>
+
+    <logger name="utils.EmbeddedMongo" level="info"/>
+    <logger name="org.mongodb.driver.protocol" level="debug"/>
+
+    <root level="error">
+        <appender-ref ref="console"/>
+    </root>
+
+</configuration>