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

Upgrade to Spring 6 and Spring Boot 3.

Browse Source 
See #654
This commit is contained in:
Christoph Strobl
2022-06-01 11:20:04 +02:00
committed by Mark Paluch
parent e59e147d6f
commit d52d73f34c
2115 changed files with 8303 additions and 9562 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
mongodb/example/src/main/java/example/springdata/mongodb/advanced/ApplicationConfiguration.java
View File

@@ -15,8 +15,8 @@
*/
package example.springdata.mongodb.advanced;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import org.bson.Document;
import org.springframework.beans.factory.annotation.Autowired;

4
mongodb/pom.xml
View File

@@ -18,7 +18,7 @@
<modules>
<module>aggregation</module>
<module>change-streams</module>
<!-- <module>change-streams</module> -->
<module>example</module>
<module>fluent-api</module>
<module>geo-json</module>
@@ -30,7 +30,7 @@
<module>repository-metrics</module>
<module>security</module>
<module>text-search</module>
<module>transactions</module>
<!-- <module>transactions</module> -->
<module>schema-validation</module>
<module>querydsl</module>
<module>linking</module>

10
mongodb/reactive/pom.xml
View File

@@ -18,16 +18,6 @@
<artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
</dependency>
<dependency>
<groupId>io.reactivex.rxjava2</groupId>
<artifactId>rxjava</artifactId>
</dependency>
<dependency>
<groupId>io.reactivex</groupId>
<artifactId>rxjava-reactive-streams</artifactId>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>

77
mongodb/reactive/src/main/java/example/springdata/mongodb/people/RxJava2PersonRepository.java
View File

@@ -1,77 +0,0 @@
/*
* Copyright 2016-2021 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.people;
import io.reactivex.Flowable;
import io.reactivex.Maybe;
import io.reactivex.Single;
import org.springframework.data.mongodb.repository.Query;
import org.springframework.data.mongodb.repository.Tailable;
import org.springframework.data.repository.reactive.RxJava2CrudRepository;
/**
* Repository interface to manage {@link Person} instances.
*
* @author Mark Paluch
* @author Christoph Strobl
*/
public interface RxJava2PersonRepository extends RxJava2CrudRepository<Person, String> {
/**
* Derived query selecting by {@code lastname}.
*
* @param lastname
* @return
*/
Flowable<Person> findByLastname(String lastname);
/**
* String query selecting one entity.
*
* @param lastname
* @return
*/
@Query("{ 'firstname': ?0, 'lastname': ?1}")
Maybe<Person> findByFirstnameAndLastname(String firstname, String lastname);
/**
* Derived query selecting by {@code lastname}. {@code lastname} uses deferred resolution that does not require
* blocking to obtain the parameter value.
*
* @param lastname
* @return
*/
Flowable<Person> findByLastname(Single<String> lastname);
/**
* Derived query selecting by {@code firstname} and {@code lastname}. {@code firstname} uses deferred resolution which
* does not require blocking to obtain the parameter value.
*
* @param firstname
* @param lastname
* @return
*/
Maybe<Person> findByFirstnameAndLastname(Single<String> firstname, String lastname);
/**
* Use a tailable cursor to emit a stream of entities as new entities are written to the capped collection.
*
* @return
*/
@Tailable
Flowable<Person> findWithTailableCursorBy();
}

13
mongodb/reactive/src/test/java/example/springdata/mongodb/people/ReactiveMongoTemplateIntegrationTest.java
View File

@@ -20,7 +20,6 @@ import static org.assertj.core.api.Assertions.*;
import example.springdata.mongodb.util.MongoContainers;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import rx.RxReactiveStreams;
import java.util.Arrays;
@@ -90,16 +89,4 @@ class ReactiveMongoTemplateIntegrationTest {
count.as(StepVerifier::create).expectNext(6L).verifyComplete();
}
/**
* Note that the all object conversions are performed before the results are printed to the console.
*/
@Test
void convertReactorTypesToRxJava2() {
var flux = template.find(Query.query(Criteria.where("lastname").is("White")), Person.class);
long count = RxReactiveStreams.toObservable(flux).count().toSingle().toBlocking().value();
assertThat(count).isEqualTo(2);
}
}

212
mongodb/reactive/src/test/java/example/springdata/mongodb/people/RxJava2PersonRepositoryIntegrationTest.java
View File

@@ -1,212 +0,0 @@
/*
* Copyright 2016-2021 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.people;
import static org.assertj.core.api.Assertions.*;
import example.springdata.mongodb.util.MongoContainers;
import io.reactivex.Flowable;
import io.reactivex.Single;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.data.mongo.DataMongoTest;
import org.springframework.data.mongodb.core.CollectionOptions;
import org.springframework.data.mongodb.core.ReactiveMongoOperations;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.testcontainers.containers.MongoDBContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
/**
* Integration test for {@link RxJava2PersonRepository} using RxJava2 types. Note that {@link ReactiveMongoOperations}
* is only available using Project Reactor types as the native Template API implementation does not come in multiple
* reactive flavors.
*
* @author Mark Paluch
* @author Jens Schauder
* @author Christoph Strobl
*/
@Testcontainers
@DataMongoTest
class RxJava2PersonRepositoryIntegrationTest {
@Container //
private static MongoDBContainer mongoDBContainer = MongoContainers.getDefaultContainer();
@DynamicPropertySource
static void setProperties(DynamicPropertyRegistry registry) {
registry.add("spring.data.mongodb.uri", mongoDBContainer::getReplicaSetUrl);
}
@Autowired RxJava2PersonRepository repository;
@Autowired ReactiveMongoOperations operations;
@BeforeEach
void setUp() {
var recreateCollection = operations.collectionExists(Person.class) //
.flatMap(exists -> exists ? operations.dropCollection(Person.class) : Mono.just(exists)) //
.then(operations.createCollection(Person.class, CollectionOptions.empty() //
.size(1024 * 1024) //
.maxDocuments(100) //
.capped()));
StepVerifier.create(recreateCollection).expectNextCount(1).verifyComplete();
repository.saveAll(Flowable.just(new Person("Walter", "White", 50), //
new Person("Skyler", "White", 45), //
new Person("Saul", "Goodman", 42), //
new Person("Jesse", "Pinkman", 27))) //
.test() //
.awaitCount(4) //
.assertNoErrors() //
.awaitTerminalEvent();
}
/**
* This sample performs a count, inserts data and performs a count again using reactive operator chaining. It prints
* the two counts ({@code 4} and {@code 6}) to the console.
*/
@Test
void shouldInsertAndCountData() {
var people = Flowable.just(new Person("Hank", "Schrader", 43), //
new Person("Mike", "Ehrmantraut", 62));
repository.count() //
.doOnSuccess(System.out::println) //
.toFlowable() //
.switchMap(count -> repository.saveAll(people)) //
.lastElement() //
.toSingle() //
.flatMap(v -> repository.count()) //
.doOnSuccess(System.out::println) //
.test() //
.awaitCount(1) //
.assertValue(6L) //
.assertNoErrors() //
.awaitTerminalEvent();
}
/**
* Note that the all object conversions are performed before the results are printed to the console.
*/
@Test
void shouldPerformConversionBeforeResultProcessing() {
repository.findAll() //
.doOnNext(System.out::println) //
.test() //
.awaitCount(4) //
.assertNoErrors() //
.awaitTerminalEvent();
}
/**
* A tailable cursor streams data using {@link Flowable} as it arrives inside the capped collection.
*/
@Test
void shouldStreamDataWithTailableCursor() throws Exception {
Queue<Person> people = new ConcurrentLinkedQueue<>();
var subscription = repository.findWithTailableCursorBy() //
.doOnNext(System.out::println) //
.doOnNext(people::add) //
.doOnComplete(() -> System.out.println("Complete")) //
.doOnTerminate(() -> System.out.println("Terminated")) //
.subscribe();
Thread.sleep(100);
repository.save(new Person("Tuco", "Salamanca", 33)).test().awaitTerminalEvent();
Thread.sleep(100);
repository.save(new Person("Mike", "Ehrmantraut", 62)).test().awaitTerminalEvent();
Thread.sleep(100);
subscription.dispose();
repository.save(new Person("Gus", "Fring", 53)).test().awaitTerminalEvent();
Thread.sleep(100);
assertThat(people).hasSize(6);
}
/**
* Fetch data using query derivation.
*/
@Test
void shouldQueryDataWithQueryDerivation() {
repository.findByLastname("White") //
.test() //
.awaitCount(2) //
.assertNoErrors() //
.awaitTerminalEvent();
}
/**
* Fetch data using a string query.
*/
@Test
void shouldQueryDataWithStringQuery() {
repository.findByFirstnameAndLastname("Walter", "White") //
.test() //
.awaitCount(1) //
.assertNoErrors() //
.awaitTerminalEvent();
}
/**
* Fetch data using query derivation.
*/
@Test
void shouldQueryDataWithDeferredQueryDerivation() {
repository.findByLastname(Single.just("White")) //
.test() //
.awaitCount(2) //
.assertNoErrors() //
.awaitTerminalEvent();
}
/**
* Fetch data using query derivation and deferred parameter resolution.
*/
@Test
void shouldQueryDataWithMixedDeferredQueryDerivation() {
repository.findByFirstnameAndLastname(Single.just("Walter"), "White") //
.test() //
.awaitCount(1) //
.assertNoErrors() //
.awaitTerminalEvent();
}
}

4
mongodb/text-search/src/main/java/example/springdata/mongodb/textsearch/MongoTestConfiguration.java
View File

@@ -15,8 +15,8 @@
*/
package example.springdata.mongodb.textsearch;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;

6
mongodb/util/pom.xml
View File

@@ -24,12 +24,6 @@
<artifactId>mongodb-driver-sync</artifactId>
</dependency>
<dependency>
<groupId>de.flapdoodle.embed</groupId>
<artifactId>de.flapdoodle.embed.mongo</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>mongodb</artifactId>

450
mongodb/util/src/main/java/example/springdata/mongodb/util/EmbeddedMongo.java
View File

@@ -1,450 +0,0 @@
/*
* Copyright 2018-2021 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.util;
import de.flapdoodle.embed.mongo.Command;
import de.flapdoodle.embed.mongo.config.MongoCmdOptions;
import de.flapdoodle.embed.mongo.config.MongodConfig;
import de.flapdoodle.embed.mongo.config.MongosConfig;
import de.flapdoodle.embed.mongo.config.Net;
import de.flapdoodle.embed.mongo.config.Storage;
import de.flapdoodle.embed.mongo.distribution.IFeatureAwareVersion;
import de.flapdoodle.embed.mongo.distribution.Versions;
import de.flapdoodle.embed.mongo.packageresolver.Feature;
import de.flapdoodle.embed.process.config.process.ProcessOutput;
import de.flapdoodle.embed.process.distribution.Version;
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 java.util.function.Function;
import org.junit.AssumptionViolatedException;
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.client.MongoClient;
import com.mongodb.client.MongoClients;
/**
* {@link org.junit.rules.TestRule} for a MongoDB server resource that is started/stopped along the test lifecycle.
*
* @author Christoph Strobl
* @author Mark Paluch
*/
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(Version.of("3.7.9"), 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;
}
/**
* Create a new {@link Builder} to build {@link EmbeddedMongo}.
*
* @return
*/
public static Builder builder() {
return new Builder();
}
/**
* Create a new {@link Builder} that is initialized as replica set to build {@link EmbeddedMongo}.
*
* @return
*/
public static Builder replSet() {
return replSet(DEFAULT_REPLICA_SET_NAME);
}
public static Builder replSet(String replicaSetName) {
return new Builder().withReplicaSetName(replicaSetName);
}
/**
* {@link Builder} for {@link EmbeddedMongo}.
*/
public static class Builder {
IFeatureAwareVersion version;
String replicaSetName;
List<Integer> serverPorts;
List<Integer> configServerPorts;
boolean silent = true;
Builder() {
version = VERSION;
replicaSetName = null;
serverPorts = Collections.emptyList();
configServerPorts = Collections.emptyList();
}
/**
* Configure the MongoDB {@link IFeatureAwareVersion version}.
*
* @param version
* @return
*/
public Builder withVersion(IFeatureAwareVersion version) {
this.version = version;
return this;
}
/**
* Configure the replica set name.
*
* @param version
* @return
*/
public Builder withReplicaSetName(String replicaSetName) {
this.replicaSetName = replicaSetName;
return this;
}
/**
* Configure the server ports.
*
* @param version
* @return
*/
public Builder withServerPorts(Integer... ports) {
this.serverPorts = Arrays.asList(ports);
return this;
}
/**
* Configure whether to stay silent (stream only Mongo process errors to stdout) or to stream all process output to
* stdout. By default, only process errors are forwarded to stdout.
*
* @param silent
* @return
*/
public Builder withSilent(boolean silent) {
this.silent = silent;
return this;
}
public EmbeddedMongo configure() {
if (serverPorts.size() > 1 || StringUtils.hasText(replicaSetName)) {
var rsName = StringUtils.hasText(replicaSetName) ? replicaSetName : DEFAULT_REPLICA_SET_NAME;
return new EmbeddedMongo(
new ReplSet(version, rsName, silent, serverPorts.toArray(new Integer[serverPorts.size()])));
}
throw new UnsupportedOperationException("implement me");
}
}
@Override
protected void before() {
try {
resource.start();
} catch (RuntimeException e) {
LOGGER.error("Cannot start MongoDB", e);
throw new AssumptionViolatedException("Cannot start MongoDB. Skipping", e);
}
}
@Override
protected void after() {
try {
resource.stop();
} catch (RuntimeException e) {
LOGGER.error("Cannot stop MongoDB", e);
}
}
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 specifying a test resource which exposes lifecycle methods and connection coordinates.
*/
interface TestResource {
/**
* Start the resource.
*/
void start();
/**
* Stop the resource.
*/
void stop();
/**
* @return the connection string to configure a MongoDB client.
*/
String connectionString();
default MongoClient mongoClient() {
return MongoClients.create(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 final Function<Command, ProcessOutput> outputFunction;
private MongosSystemForTestFactory mongosTestFactory;
ReplSet(IFeatureAwareVersion serverVersion, String replicaSetName, boolean silent, 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();
if (silent) {
outputFunction = it -> ProcessOutput.silent();
} else {
outputFunction = it -> ProcessOutput.named(it.commandName(), LoggerFactory.getLogger(getClass()));
}
}
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<MongodConfig>> replicaSets = new LinkedHashMap<>();
replicaSets.put(configServerReplicaSetName, initConfigServers());
replicaSets.put(replicaSetName, initReplicaSet());
// create mongos
var mongosConfig = defaultMongosConfig(serverVersion, mongosPort, defaultCommandOptions(),
configServerReplicaSetName, configServerPorts[0]);
mongosTestFactory = new MongosSystemForTestFactory(mongosConfig, replicaSets, Collections.emptyList(),
DEFAULT_SHARDING, DEFAULT_SHARDING, DEFAULT_SHARD_KEY, outputFunction);
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<MongodConfig> initReplicaSet() {
List<MongodConfig> rs = new ArrayList<>();
for (int port : serverPorts) {
rs.add(defaultMongodConfig(serverVersion, port, defaultCommandOptions(), false, true, replicaSetName));
}
return rs;
}
private List<MongodConfig> initConfigServers() {
List<MongodConfig> configServers = new ArrayList<>(configServerPorts.length);
for (var 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;
}
}
/**
* @return Default {@link IMongoCmdOptions command options}.
*/
private static MongoCmdOptions defaultCommandOptions() {
return MongoCmdOptions.builder() //
.useNoPrealloc(false) //
.useSmallFiles(false) //
.useNoJournal(false) //
.storageEngine(STORAGE_ENGINE) //
.isVerbose(false) //
.build();
}
/**
* Create a default {@code mongod} config.
*
* @param version
* @param port
* @param cmdOptions
* @param configServer
* @param shardServer
* @param replicaSet
* @return
*/
private static MongodConfig defaultMongodConfig(IFeatureAwareVersion version, int port, MongoCmdOptions cmdOptions,
boolean configServer, boolean shardServer, String replicaSet) {
try {
var builder = MongodConfig.builder() //
.version(version) //
.putArgs("--quiet", null) //
.net(new Net(LOCALHOST, port, Network.localhostIsIPv6())) //
.isConfigServer(configServer).cmdOptions(cmdOptions); //
if (StringUtils.hasText(replicaSet)) {
builder = builder //
.replication(new Storage(null, replicaSet, 0));
if (!configServer) {
builder = builder.isShardServer(shardServer);
} else {
builder = builder.isShardServer(false);
}
}
return builder.build();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* Create a default {@code mongos} config.
*
* @param version
* @param port
* @param cmdOptions
* @param configServerReplicaSet
* @param configServerPort
* @return
*/
private static MongosConfig defaultMongosConfig(IFeatureAwareVersion version, int port, MongoCmdOptions cmdOptions,
String configServerReplicaSet, int configServerPort) {
try {
var builder = MongosConfig.builder() //
.version(version) //
.putArgs("--quiet", null) //
.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);
}
}
}

283
mongodb/util/src/main/java/example/springdata/mongodb/util/MongosSystemForTestFactory.java
View File

@@ -1,283 +0,0 @@
/*
* Copyright 2018-2021 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.util;
import de.flapdoodle.embed.mongo.Command;
import de.flapdoodle.embed.mongo.MongodProcess;
import de.flapdoodle.embed.mongo.MongodStarter;
import de.flapdoodle.embed.mongo.MongosExecutable;
import de.flapdoodle.embed.mongo.MongosProcess;
import de.flapdoodle.embed.mongo.MongosStarter;
import de.flapdoodle.embed.mongo.config.MongodConfig;
import de.flapdoodle.embed.mongo.config.MongosConfig;
import de.flapdoodle.embed.mongo.config.Net;
import de.flapdoodle.embed.process.config.RuntimeConfig;
import de.flapdoodle.embed.process.config.process.ProcessOutput;
import lombok.SneakyThrows;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.mongodb.BasicDBList;
import com.mongodb.BasicDBObject;
import com.mongodb.DBObject;
import com.mongodb.MongoClientSettings;
import com.mongodb.ServerAddress;
import com.mongodb.client.MongoClients;
class MongosSystemForTestFactory {
private final static Logger logger = LoggerFactory
.getLogger(MongosSystemForTestFactory.class);
public static final String ADMIN_DATABASE_NAME = "admin";
public static final String LOCAL_DATABASE_NAME = "local";
public static final String REPLICA_SET_NAME = "rep1";
public static final String OPLOG_COLLECTION = "oplog.rs";
private final MongosConfig config;
private final Map<String, List<MongodConfig>> replicaSets;
private final List<MongodConfig> configServers;
private final String shardDatabase;
private final String shardCollection;
private final String shardKey;
private final Function<Command, ProcessOutput> outputFunction;
private MongosExecutable mongosExecutable;
private MongosProcess mongosProcess;
private List<MongodProcess> mongodProcessList;
private List<MongodProcess> mongodConfigProcessList;
public MongosSystemForTestFactory(MongosConfig config, Map<String, List<MongodConfig>> replicaSets,
List<MongodConfig> configServers, String shardDatabase,
String shardCollection, String shardKey, Function<Command, ProcessOutput> outputFunction) {
this.config = config;
this.replicaSets = replicaSets;
this.configServers = configServers;
this.shardDatabase = shardDatabase;
this.shardCollection = shardCollection;
this.shardKey = shardKey;
this.outputFunction = outputFunction;
}
public void start() throws Throwable {
this.mongodProcessList = new ArrayList<>();
this.mongodConfigProcessList = new ArrayList<>();
for (var entry : replicaSets.entrySet()) {
initializeReplicaSet(entry);
}
for (var config : configServers) {
initializeConfigServer(config);
}
initializeMongos();
configureMongos();
}
private void initializeReplicaSet(Entry<String, List<MongodConfig>> entry)
throws Exception {
var replicaName = entry.getKey();
var mongoConfigList = entry.getValue();
if (mongoConfigList.size() < 3) {
throw new Exception(
"A replica set must contain at least 3 members.");
}
// Create 3 mongod processes
for (var mongoConfig : mongoConfigList) {
if (!mongoConfig.replication().getReplSetName().equals(replicaName)) {
throw new Exception(
"Replica set name must match in mongo configuration");
}
RuntimeConfig runtimeConfig = RuntimeConfig.builder()
// .defaultsWithLogger(Command.MongoD,logger)
.processOutput(outputFunction.apply(Command.MongoD))
.build();
var starter = MongodStarter.getInstance(runtimeConfig);
var mongodExe = starter.prepare(mongoConfig);
var process = mongodExe.start();
mongodProcessList.add(process);
}
Thread.sleep(1000);
var mo = MongoClientSettings.builder()
.applyToSocketSettings(builder -> builder.connectTimeout(10, TimeUnit.SECONDS)).applyToClusterSettings(
builder -> builder.hosts(Collections.singletonList(toAddress(mongoConfigList.get(0).net()))))
.build();
var mongo = MongoClients.create(mo);
var mongoAdminDB = mongo.getDatabase(ADMIN_DATABASE_NAME);
var cr = mongoAdminDB.runCommand(new Document("isMaster", 1));
logger.info("isMaster: {}", cr);
// Build BSON object replica set settings
DBObject replicaSetSetting = new BasicDBObject();
replicaSetSetting.put("_id", replicaName);
var members = new BasicDBList();
var i = 0;
for (var mongoConfig : mongoConfigList) {
DBObject host = new BasicDBObject();
host.put("_id", i++);
host.put("host", mongoConfig.net().getServerAddress().getHostName()
+ ":" + mongoConfig.net().getPort());
members.add(host);
}
replicaSetSetting.put("members", members);
logger.info(replicaSetSetting.toString());
// Initialize replica set
cr = mongoAdminDB.runCommand(new Document("replSetInitiate",
replicaSetSetting));
logger.info("replSetInitiate: {}", cr);
Thread.sleep(5000);
cr = mongoAdminDB.runCommand(new Document("replSetGetStatus", 1));
logger.info("replSetGetStatus: {}", cr);
// Check replica set status before to proceed
while (!isReplicaSetStarted(cr)) {
logger.info("Waiting for 3 seconds...");
Thread.sleep(1000);
cr = mongoAdminDB.runCommand(new Document("replSetGetStatus", 1));
logger.info("replSetGetStatus: {}", cr);
}
mongo.close();
mongo = null;
}
private boolean isReplicaSetStarted(Document setting) {
if (setting.get("members") == null) {
return false;
}
var members = (List) setting.get("members");
for (var m : members) {
var member = (Document) m;
logger.info(member.toString());
var state = member.getInteger("state", 0);
logger.info("state: {}", state);
// 1 - PRIMARY, 2 - SECONDARY, 7 - ARBITER
if (state != 1 && state != 2 && state != 7) {
return false;
}
}
return true;
}
private void initializeConfigServer(MongodConfig config) throws Exception {
if (!config.isConfigServer()) {
throw new Exception(
"Mongo configuration is not a defined for a config server.");
}
var starter = MongodStarter.getDefaultInstance();
var mongodExe = starter.prepare(config);
var process = mongodExe.start();
mongodProcessList.add(process);
}
private void initializeMongos() throws Exception {
var runtime = MongosStarter.getInstance(RuntimeConfig.builder()
// .defaultsWithLogger(Command.MongoS,logger)
.processOutput(outputFunction.apply(Command.MongoS))
.build());
mongosExecutable = runtime.prepare(config);
mongosProcess = mongosExecutable.start();
}
private void configureMongos() throws Exception {
Document cr;
var options = MongoClientSettings.builder()
.applyToSocketSettings(builder -> builder.connectTimeout(10, TimeUnit.SECONDS))
.applyToClusterSettings(builder -> builder.hosts(Collections.singletonList(toAddress(this.config.net()))))
.build();
try (var mongo = MongoClients.create(options)) {
var mongoAdminDB = mongo.getDatabase(ADMIN_DATABASE_NAME);
// Add shard from the replica set list
for (var entry : this.replicaSets
.entrySet()) {
var replicaName = entry.getKey();
var command = "";
for (var mongodConfig : entry.getValue()) {
if (command.isEmpty()) {
command = replicaName + "/";
} else {
command += ",";
}
command += mongodConfig.net().getServerAddress().getHostName()
+ ":" + mongodConfig.net().getPort();
}
logger.info("Execute add shard command: {}", command);
cr = mongoAdminDB.runCommand(new Document("addShard", command));
logger.info(cr.toString());
}
logger.info("Execute list shards.");
cr = mongoAdminDB.runCommand(new Document("listShards", 1));
logger.info(cr.toString());
// Enabled sharding at database level
logger.info("Enabled sharding at database level");
cr = mongoAdminDB.runCommand(new Document("enableSharding",
this.shardDatabase));
logger.info(cr.toString());
// Create index in sharded collection
logger.info("Create index in sharded collection");
var db = mongo.getDatabase(this.shardDatabase);
db.getCollection(this.shardCollection).createIndex(new Document(this.shardKey, 1));
// Shard the collection
logger.info("Shard the collection: {}.{}", this.shardDatabase, this.shardCollection);
var cmd = new Document();
cmd.put("shardCollection", this.shardDatabase + "." + this.shardCollection);
cmd.put("key", new BasicDBObject(this.shardKey, 1));
cr = mongoAdminDB.runCommand(cmd);
logger.info(cr.toString());
logger.info("Get info from config/shards");
var cursor = mongo.getDatabase("config").getCollection("shards").find();
for (Document item : cursor) {
logger.info(item.toString());
}
}
}
@SneakyThrows
private static ServerAddress toAddress(Net net) {
return new ServerAddress(net.getServerAddress(), net.getPort());
}
public void stop() {
for (var process : this.mongodProcessList) {
process.stop();
}
for (var process : this.mongodConfigProcessList) {
process.stop();
}
this.mongosProcess.stop();
}
}