#337 - Add sample for MongoDB 3.6 Change Streams.

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<S, T> implements MessageListener<S, T> {
+
+	private final BlockingDeque<Message<S, T>> 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<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()));
+		messages.add(message);
+	}
+
+	int messageCount() {
+		return messages.size();
+	}
+}

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.
+ * <p />
+ * This time <a href="http://www.cbs.com/shows/star-trek-discovery/">Star Trek: Discovery</a> <br />
+ * <i>Just a trekkie - no relation to CBS in any way other than watching.</i>
+ *
+ * @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;
+	}
+}

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
+ * <a href="https://docs.mongodb.com/manual/changeStreams/">Change Streams</a> using the sync and reactivestreams 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
+ */
+@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 <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
+		 * {@link MessageListenerContainer}.
+		 * <p />
+		 * 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<ChangeStreamDocument<Document>, Person> messageListener = new CollectingMessageListener();
+
+		ChangeStreamRequest<Person> 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<ChangeStreamEvent<Person>> 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();
+	}
+}

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);
+	}
+
+}

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