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

#337 - Polishing.

Browse Source 
Use blocking queue for imperative synchronization to not depend on the actual machine timing. Refactor reactive API example to use StepVerifier.

Reuse embedded MongoDB utility.
This commit is contained in:
Mark Paluch
2018-02-09 10:54:39 +01:00
parent 2474846281
commit 5e108d35da
7 changed files with 147 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
mongodb/change-streams/pom.xml
View File

@@ -1,65 +1,55 @@
<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>
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>
<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-change-streams</artifactId>
<name>Spring Data MongoDB - Change Streams</name>
<artifactId>spring-data-mongodb-change-streams</artifactId>
<name>Spring Data MongoDB - Change Streams</name>
<dependencies>
<properties>
<mongodb.version>3.8.0-beta2</mongodb.version>
<mongo-driver-reactivestreams.version>1.9.0-beta1</mongo-driver-reactivestreams.version>
<spring-data-releasetrain.version>Lovelace-M3</spring-data-releasetrain.version>
</properties>
<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>
<profiles>
<dependency>
<groupId>org.mongodb</groupId>
<artifactId>mongo-java-driver</artifactId>
<version>3.6.2</version>
</dependency>
<!-- Override property as the module always needs Lovelace -->
<dependency>
<groupId>org.mongodb</groupId>
<artifactId>mongodb-driver-async</artifactId>
<version>3.6.2</version>
</dependency>
<profile>
<id>spring-data-next</id>
<properties>
<spring-data-releasetrain.version>Lovelace-M3</spring-data-releasetrain.version>
</properties>
</profile>
<dependency>
<groupId>org.mongodb</groupId>
<artifactId>mongodb-driver-reactivestreams</artifactId>
<version>1.7.0</version>
</dependency>
</profiles>
<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-mongodb</artifactId>
<version>2.1.0.M1</version>
</dependency>
<dependencies>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
</dependency>
<dependency>
<groupId>de.flapdoodle.embed</groupId>
<artifactId>de.flapdoodle.embed.mongo</artifactId>
<version>2.0.1</version>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>spring-data-mongodb-example-utils</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>

15
mongodb/change-streams/src/main/java/example/springdata/mongodb/CollectingMessageListener.java
View File

@@ -17,6 +17,7 @@ package example.springdata.mongodb;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.data.mongodb.core.messaging.Message;
import org.springframework.data.mongodb.core.messaging.MessageListener;
@@ -26,10 +27,12 @@ import org.springframework.data.mongodb.core.messaging.MessageListener;
* and printing the message itself to the console.
*
* @author Christoph Strobl
* @author Mark Paluch
*/
class CollectingMessageListener<S, T> implements MessageListener<S, T> {
private final BlockingDeque<Message<S, T>> messages = new LinkedBlockingDeque<>();
private final AtomicInteger count = new AtomicInteger();
/*
* (non-Javadoc)
@@ -38,12 +41,22 @@ class CollectingMessageListener<S, T> implements MessageListener<S, T> {
@Override
public void onMessage(Message<S, T> message) {
System.out.println(String.format("Received Message in collection %s.\n\trawsource: %s\n\tconverted: %s",
message.getProperties().getCollectionName(), message.getRaw(), message.getBody()));
count.incrementAndGet();
messages.add(message);
}
int messageCount() {
return messages.size();
return count.get();
}
void awaitNextMessages(int count) throws InterruptedException {
for (int i = 0; i < count; i++) {
messages.take();
}
}
}

113
mongodb/change-streams/src/test/java/example/springdata/mongodb/ChangeStreamsTests.java
View File

@@ -21,18 +21,19 @@ import static org.springframework.data.mongodb.core.query.Criteria.*;
import static org.springframework.data.mongodb.core.query.Query.*;
import static org.springframework.data.mongodb.core.query.Update.*;
import reactor.core.Disposable;
import example.springdata.mongodb.util.EmbeddedMongo;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import java.time.Duration;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
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.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.mongo.embedded.EmbeddedMongoAutoConfiguration;
import org.springframework.boot.test.autoconfigure.data.mongo.DataMongoTest;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -40,6 +41,8 @@ import org.springframework.data.mongodb.core.ChangeStreamEvent;
import org.springframework.data.mongodb.core.ChangeStreamOptions;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.ReactiveMongoOperations;
import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
import org.springframework.data.mongodb.core.SimpleReactiveMongoDatabaseFactory;
import org.springframework.data.mongodb.core.messaging.ChangeStreamRequest;
import org.springframework.data.mongodb.core.messaging.DefaultMessageListenerContainer;
import org.springframework.data.mongodb.core.messaging.Message;
@@ -48,23 +51,26 @@ import org.springframework.data.mongodb.core.messaging.Subscription;
import org.springframework.test.context.junit4.SpringRunner;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
import com.mongodb.reactivestreams.client.MongoClients;
/**
* A simple Test demonstrating required {@link Configuration} for consumption of MongoDB
* <a href="https://docs.mongodb.com/manual/changeStreams/">Change Streams</a> using the sync and reactivestreams java
* <a href="https://docs.mongodb.com/manual/changeStreams/">Change Streams</a> using the sync and Reactive Streams Java
* driver.
* <p />
* We currently need to have both tests in one class due to some shutdown error in flapdoodle which lets us start the
* process only once.
*
* @author Christoph Strobl
* @author Mark Paluch
*/
@RunWith(SpringRunner.class)
@DataMongoTest
public class ChangeStreamsTests extends ReplicaSetInitiatingTest {
public class ChangeStreamsTests {
public static @ClassRule EmbeddedMongo replSet = EmbeddedMongo.replSet().configure();
@Autowired MessageListenerContainer container; // for imperative style
@Autowired MongoTemplate template;
@Autowired ReactiveMongoOperations reactiveTemplate; // for reactive style
Person gabriel = new Person("Gabriel", "Lorca", 30);
@@ -74,10 +80,30 @@ public class ChangeStreamsTests extends ReplicaSetInitiatingTest {
/**
* Configuration? Yes we need a bit of it - Do not worry, it won't be much!
*/
@Configuration
@EnableAutoConfiguration
@SpringBootApplication(exclude = EmbeddedMongoAutoConfiguration.class)
static class Config {
/**
* Configure {@link SimpleMongoDbFactory} pointing to the embedded MongoDB connection.
*
* @return a new {@link SimpleReactiveMongoDatabaseFactory}.
*/
@Bean
SimpleMongoDbFactory mongoDbFactory() {
return new SimpleMongoDbFactory(replSet.getMongoClient(), "changestreams");
}
/**
* Configure {@link SimpleReactiveMongoDatabaseFactory} pointing to the embedded MongoDB connection.
*
* @return a new {@link SimpleReactiveMongoDatabaseFactory}.
*/
@Bean
SimpleReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory() {
return new SimpleReactiveMongoDatabaseFactory(MongoClients.create(replSet.getConnectionString()),
"changestreams");
}
/**
* Since listening to a <a href="https://docs.mongodb.com/manual/changeStreams/">Change Stream</a> using the sync
* MongoDB Java Driver is a blocking class, we need to move load to another {@link Thread} by simply using a
@@ -105,19 +131,18 @@ public class ChangeStreamsTests extends ReplicaSetInitiatingTest {
/**
* Use the {@link MessageListenerContainer} registered within the
* {@link org.springframework.context.ApplicationContext} to subscribe to a MongoDB Change Stream. Events published
* via {@link com.mongodb.client.ChangeStreamIterable} are passed to the
* {@link org.springframework.context.ApplicationContext} to subscribe to MongoDB Change Streams. Events published via
* {@link com.mongodb.client.ChangeStreamIterable} are passed to the
* {@link org.springframework.data.mongodb.core.messaging.MessageListener#onMessage(Message) MessageListener}.
*/
@Test
public void imperativeChangeEvents() throws InterruptedException {
CollectingMessageListener<ChangeStreamDocument<Document>, Person> messageListener = new CollectingMessageListener();
CollectingMessageListener<ChangeStreamDocument<Document>, Person> messageListener = new CollectingMessageListener<>();
ChangeStreamRequest<Person> request = ChangeStreamRequest.builder() //
ChangeStreamRequest<Person> request = ChangeStreamRequest.builder(messageListener) //
.collection("person") //
.filter(newAggregation(match(where("operationType").is("insert")))) // we are only interested in inserts
.publishTo(messageListener) //
.build();
Subscription subscription = container.register(request, Person.class);
@@ -126,7 +151,7 @@ public class ChangeStreamsTests extends ReplicaSetInitiatingTest {
template.save(gabriel);
template.save(ash);
Thread.sleep(200);
messageListener.awaitNextMessages(2);
assertThat(messageListener.messageCount()).isEqualTo(2); // first two insert events, so far so good
@@ -141,46 +166,46 @@ public class ChangeStreamsTests extends ReplicaSetInitiatingTest {
template.save(michael);
Thread.sleep(200);
messageListener.awaitNextMessages(1);
assertThat(messageListener.messageCount()).isEqualTo(3); // there we go, all events received.
}
/**
* Use the {@link reactor.core.publisher.Flux} to subscribe to a MongoDB Change Stream.
* Use a {@link reactor.core.publisher.Flux} to subscribe to MongoDB Change Streams.
*/
@Test
public void reactiveChangeEvents() throws InterruptedException {
public void reactiveChangeEvents() {
BlockingQueue<ChangeStreamEvent<Person>> documents = new LinkedBlockingQueue<>(100);
Flux<ChangeStreamEvent<Person>> changeStream = reactiveTemplate.changeStream(
newAggregation(match(where("operationType").is("insert"))), Person.class, ChangeStreamOptions.empty(),
"person");
Disposable disposable = reactiveTemplate.changeStream(newAggregation(match(where("operationType").is("insert"))),
Person.class, ChangeStreamOptions.empty(), "person").doOnNext(documents::add).subscribe();
StepVerifier.create(changeStream) //
.expectSubscription() //
.expectNoEvent(Duration.ofMillis(200)) // wait till change streams becomes active
Thread.sleep(200); // wait till the subscription becomes active
// Save documents and await their change events
.then(() -> {
StepVerifier.create(reactiveTemplate.save(gabriel)).expectNextCount(1).verifyComplete();
StepVerifier.create(reactiveTemplate.save(ash)).expectNextCount(1).verifyComplete();
}).expectNextCount(2) //
StepVerifier.create(reactiveTemplate.save(gabriel)).expectNextCount(1).verifyComplete();
StepVerifier.create(reactiveTemplate.save(ash)).expectNextCount(1).verifyComplete();
// Update a document
.then(() -> {
Thread.sleep(200);
StepVerifier.create(reactiveTemplate.update(Person.class) //
.matching(query(where("id").is(ash.getId()))) //
.apply(update("age", 40)) //
.first()).expectNextCount(1).verifyComplete();
}).expectNoEvent(Duration.ofMillis(200)) // updates are skipped
assertThat(documents.size()).isEqualTo(2); // first two insert events, so far so good
// Save another document and await its change event
.then(() -> {
StepVerifier.create(reactiveTemplate.save(michael)).expectNextCount(1).verifyComplete();
}).expectNextCount(1) // there we go, all events received.
StepVerifier.create(reactiveTemplate.update(Person.class) //
.matching(query(where("id").is(ash.getId()))) //
.apply(update("age", 40)) //
.first()).expectNextCount(1).verifyComplete();
Thread.sleep(200);
assertThat(documents.size()).isEqualTo(2); // updates are skipped
StepVerifier.create(reactiveTemplate.save(michael)).expectNextCount(1).verifyComplete();
Thread.sleep(200); // just give it some time to link receive all events
assertThat(documents.size()).isEqualTo(3); // there we go, all events received.
disposable.dispose();
.thenCancel() // change streams are infinite streams, at some point we need to unsubscribe
.verify();
}
}

70
mongodb/change-streams/src/test/java/example/springdata/mongodb/ReplicaSetInitiatingTest.java
View File

@@ -1,70 +0,0 @@
/*
* 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 java.util.Arrays;
import org.bson.Document;
import org.hamcrest.number.IsCloseTo;
import org.junit.Assume;
import org.junit.Before;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.mongodb.core.MongoTemplate;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.ServerAddress;
import com.mongodb.client.MongoDatabase;
/**
* Just a simple base Class initiating the required replica set.
*
* @author Christoph Strobl
*/
abstract class ReplicaSetInitiatingTest {
@Autowired MongoTemplate template;
@Value("${spring.data.mongodb.port}") Integer port;
/**
* Before we can actually continue we need to make sure that the Server works as a single node replica set which is
* required for Change Streams.
*/
@Before
public void setUp() {
MongoClient mongo = new MongoClient(new ServerAddress("localhost:" + port),
MongoClientOptions.builder().connectTimeout(10).build());
MongoDatabase admin = mongo.getDatabase("admin");
Document status = admin.runCommand(new Document("serverStatus", "1"));
if (!status.get("repl", Document.class).get("ismaster", Boolean.class)) {
Document result = admin.runCommand(new Document("replSetInitiate", new Document("_id", "spring-data-examples")
.append("members", Arrays.asList(new Document("_id", 0).append("host", "localhost:" + port)))));
Assume.assumeThat(result.getDouble("ok"), IsCloseTo.closeTo(1.0D, 0D));
}
mongo.close();
template.dropCollection(Person.class);
template.createCollection(Person.class);
}
}

4
mongodb/change-streams/src/test/resources/application.properties
View File

@@ -1,4 +0,0 @@
spring.mongodb.embedded.version=3.6.0
spring.mongodb.embedded.features=ONLY_WITH_SSL,ONLY_64BIT,NO_HTTP_INTERFACE_ARG
spring.mongodb.embedded.storage.repl-set-name=spring-data-examples
spring.data.mongodb.port=57143

17
mongodb/change-streams/src/test/resources/logback.xml Normal file
View File

@@ -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="example.springdata.mongodb.util.EmbeddedMongo" level="info"/>
<logger name="org.mongodb.driver.protocol" level="debug"/>
<root level="error">
<appender-ref ref="console"/>
</root>
</configuration>

13
mongodb/transactions/pom.xml
View File

@@ -37,6 +37,12 @@
<artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>spring-data-mongodb-example-utils</artifactId>
@@ -44,13 +50,6 @@
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<version>3.1.7.RELEASE</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>