diff --git a/mongodb/change-streams/README.md b/mongodb/change-streams/README.md new file mode 100644 index 00000000..09d0704e --- /dev/null +++ b/mongodb/change-streams/README.md @@ -0,0 +1,49 @@ +# Spring Data MongoDB - Change Streams Example + +This project contains usage samples for consuming MongoDB 3.6 [Change Streams](https://docs.mongodb.com/manual/changeStreams/) using the imperative as well as the reactive MongoDB Java drivers. + +### Imperative Style + +Change stream events can be consumed using a `MessageListener` registered within a `MessageListenerContainer`. The container takes care of running the task in a separate `Thread` pushing events to the `MessageListener`. + +```java +@Configuration +class Config { + + @Bean + MessageListenerContainer messageListenerContainer(MongoTemplate template) { + return new DefaultMessageListenerContainer(template); + } +} +``` + +Once the `MessageListenerContainer` is in place `MessageListeners` can be registered. + +```java +MessageListener, Person> messageListener = (message) -> { + System.out.println("Hello " + message.getBody().getFirstname()); +}; + +ChangeStreamRequest request = ChangeStreamRequest.builder() + .collection("person") + .filter(newAggregation(match(where("operationType").is("insert")))) + .publishTo(messageListener) + .build(); + +Subscription subscription = messageListenerContainer.register(request, Person.class); + +// ... +``` + +### Reactive Style + +Change stream events be directly consumed via a `Flux` connected to the change stream. + +```java +Flux changeStream = reactiveTemplate + .changeStream(newAggregation(match(where("operationType").is("insert"))), + Person.class, ChangeStreamOptions.empty(), "person"); + +changeStream.doOnNext(event -> System.out.println("Hello " + event.getBody().getFirstname())) + .subscribe(); +``` \ No newline at end of file diff --git a/mongodb/change-streams/pom.xml b/mongodb/change-streams/pom.xml new file mode 100644 index 00000000..1b409e86 --- /dev/null +++ b/mongodb/change-streams/pom.xml @@ -0,0 +1,65 @@ + + 4.0.0 + + + org.springframework.data.examples + spring-data-mongodb-examples + 2.0.0.BUILD-SNAPSHOT + + + spring-data-mongodb-change-streams + Spring Data MongoDB - Change Streams + + + + + org.springframework.boot + spring-boot-starter-data-mongodb-reactive + + + org.mongodb + mongodb-driver + + + + + + org.mongodb + mongo-java-driver + 3.6.2 + + + + org.mongodb + mongodb-driver-async + 3.6.2 + + + + org.mongodb + mongodb-driver-reactivestreams + 1.7.0 + + + + org.springframework.data + spring-data-mongodb + 2.1.0.M1 + + + + io.projectreactor + reactor-test + test + + + + de.flapdoodle.embed + de.flapdoodle.embed.mongo + 2.0.1 + + + + + diff --git a/mongodb/change-streams/src/main/java/example/springdata/mongodb/CollectingMessageListener.java b/mongodb/change-streams/src/main/java/example/springdata/mongodb/CollectingMessageListener.java new file mode 100644 index 00000000..49cdd6fd --- /dev/null +++ b/mongodb/change-streams/src/main/java/example/springdata/mongodb/CollectingMessageListener.java @@ -0,0 +1,49 @@ +/* + * 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.concurrent.BlockingDeque; +import java.util.concurrent.LinkedBlockingDeque; + +import org.springframework.data.mongodb.core.messaging.Message; +import org.springframework.data.mongodb.core.messaging.MessageListener; + +/** + * Just a simple {@link MessageListener} implementation collecting {@link Message messages} in a {@link BlockingDeque} + * and printing the message itself to the console. + * + * @author Christoph Strobl + */ +class CollectingMessageListener implements MessageListener { + + private final BlockingDeque> messages = new LinkedBlockingDeque<>(); + + /* + * (non-Javadoc) + * @see org.springframework.data.mongodb.core.messaging.MessageListener#onMessage(org.springframework.data.mongodb.core.messaging.Message) + */ + @Override + public void onMessage(Message message) { + + System.out.println(String.format("Received Message in collection %s.\n\trawsource: %s\n\tconverted: %s", + message.getProperties().getCollectionName(), message.getRaw(), message.getBody())); + messages.add(message); + } + + int messageCount() { + return messages.size(); + } +} diff --git a/mongodb/change-streams/src/main/java/example/springdata/mongodb/Person.java b/mongodb/change-streams/src/main/java/example/springdata/mongodb/Person.java new file mode 100644 index 00000000..79d86b87 --- /dev/null +++ b/mongodb/change-streams/src/main/java/example/springdata/mongodb/Person.java @@ -0,0 +1,49 @@ +/* + * 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.Data; + +import org.bson.types.ObjectId; +import org.springframework.data.annotation.Id; +import org.springframework.data.mongodb.core.mapping.Field; + +/** + * The domain type of choice. + *

+ * This time Star Trek: Discovery
+ * Just a trekkie - no relation to CBS in any way other than watching. + * + * @author Christoph Strobl + */ +@Data +class Person { + + @Id ObjectId id; + + @Field("first_name") private String firstname; + + @Field("last_name") private String lastname; + + private int age; + + Person(String firstname, String lastname, int age) { + + this.firstname = firstname; + this.lastname = lastname; + this.age = age; + } +} diff --git a/mongodb/change-streams/src/test/java/example/springdata/mongodb/ChangeStreamsTests.java b/mongodb/change-streams/src/test/java/example/springdata/mongodb/ChangeStreamsTests.java new file mode 100644 index 00000000..7f3ca223 --- /dev/null +++ b/mongodb/change-streams/src/test/java/example/springdata/mongodb/ChangeStreamsTests.java @@ -0,0 +1,186 @@ +/* + * 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 static org.assertj.core.api.Assertions.*; +import static org.springframework.data.mongodb.core.aggregation.Aggregation.*; +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 reactor.test.StepVerifier; + +import java.time.Duration; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; + +import org.bson.Document; +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.test.autoconfigure.data.mongo.DataMongoTest; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +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.messaging.ChangeStreamRequest; +import org.springframework.data.mongodb.core.messaging.DefaultMessageListenerContainer; +import org.springframework.data.mongodb.core.messaging.Message; +import org.springframework.data.mongodb.core.messaging.MessageListenerContainer; +import org.springframework.data.mongodb.core.messaging.Subscription; +import org.springframework.test.context.junit4.SpringRunner; + +import com.mongodb.client.model.changestream.ChangeStreamDocument; + +/** + * A simple Test demonstrating required {@link Configuration} for consumption of MongoDB + * Change Streams using the sync and reactivestreams java + * driver. + *

+ * 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 + */ +@RunWith(SpringRunner.class) +@DataMongoTest +public class ChangeStreamsTests extends ReplicaSetInitiatingTest { + + @Autowired MessageListenerContainer container; // for imperative style + + @Autowired ReactiveMongoOperations reactiveTemplate; // for reactive style + + Person gabriel = new Person("Gabriel", "Lorca", 30); + Person michael = new Person("Michael", "Burnham", 30); + Person ash = new Person("Ash", "Tyler", 35); + + /** + * Configuration? Yes we need a bit of it - Do not worry, it won't be much! + */ + @Configuration + @EnableAutoConfiguration + static class Config { + + /** + * Since listening to a Change Stream using the sync + * MongoDB Java Driver is a blocking class, we need to move load to another {@link Thread} by simply using a + * {@link MessageListenerContainer}. + *

+ * As this is a {@link org.springframework.context.SmartLifecycle smart lifecycle component} we do actually not need + * to worry about its lifecycle, the resource allocation and freeing. + * + * @param template must not be {@literal null}. + * @return a new {@link MessageListenerContainer} + */ + @Bean + MessageListenerContainer messageListenerContainer(MongoTemplate template) { + + return new DefaultMessageListenerContainer(template) { + + /* auto startup will be changed for M2, so this should no longer be required. */ + @Override + public boolean isAutoStartup() { + return true; + } + }; + } + } + + /** + * 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.data.mongodb.core.messaging.MessageListener#onMessage(Message) MessageListener}. + */ + @Test + public void imperativeChangeEvents() throws InterruptedException { + + CollectingMessageListener, Person> messageListener = new CollectingMessageListener(); + + ChangeStreamRequest request = ChangeStreamRequest.builder() // + .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); + subscription.await(Duration.ofMillis(200)); // wait till the subscription becomes active + + template.save(gabriel); + template.save(ash); + + Thread.sleep(200); + + assertThat(messageListener.messageCount()).isEqualTo(2); // first two insert events, so far so good + + template.update(Person.class) // + .matching(query(where("id").is(ash.getId()))) // + .apply(update("age", 40)) // + .first(); + + Thread.sleep(200); + + assertThat(messageListener.messageCount()).isEqualTo(2); // updates are skipped + + template.save(michael); + + Thread.sleep(200); + + 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. + */ + @Test + public void reactiveChangeEvents() throws InterruptedException { + + BlockingQueue> documents = new LinkedBlockingQueue<>(100); + + Disposable disposable = reactiveTemplate.changeStream(newAggregation(match(where("operationType").is("insert"))), + Person.class, ChangeStreamOptions.empty(), "person").doOnNext(documents::add).subscribe(); + + Thread.sleep(200); // wait till the subscription becomes active + + StepVerifier.create(reactiveTemplate.save(gabriel)).expectNextCount(1).verifyComplete(); + StepVerifier.create(reactiveTemplate.save(ash)).expectNextCount(1).verifyComplete(); + + Thread.sleep(200); + + assertThat(documents.size()).isEqualTo(2); // first two insert events, so far so good + + 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(); + } +} diff --git a/mongodb/change-streams/src/test/java/example/springdata/mongodb/ReplicaSetInitiatingTest.java b/mongodb/change-streams/src/test/java/example/springdata/mongodb/ReplicaSetInitiatingTest.java new file mode 100644 index 00000000..a58edc38 --- /dev/null +++ b/mongodb/change-streams/src/test/java/example/springdata/mongodb/ReplicaSetInitiatingTest.java @@ -0,0 +1,70 @@ +/* + * 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); + } + +} diff --git a/mongodb/change-streams/src/test/resources/application.properties b/mongodb/change-streams/src/test/resources/application.properties new file mode 100644 index 00000000..bbf03ded --- /dev/null +++ b/mongodb/change-streams/src/test/resources/application.properties @@ -0,0 +1,4 @@ +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 \ No newline at end of file diff --git a/mongodb/pom.xml b/mongodb/pom.xml index f7788b91..907a68f2 100644 --- a/mongodb/pom.xml +++ b/mongodb/pom.xml @@ -18,6 +18,7 @@ aggregation + change-streams example fluent-api geo-json