Example for change streams in MongoDB 3.6

Author: ttrelle

mongodriver/.gitignore

@@ -0,0 +1,2 @@
+/.settings/
+/target/

mongodriver/pom.xml

@@ -0,0 +1,27 @@
+<?xml version="1.0"?>
+<project
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
+	xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>de.codecentric</groupId>
+		<artifactId>springdata-examples</artifactId>
+		<version>1.0.0</version>
+	</parent>
+	<groupId>de.codecentric</groupId>
+	<artifactId>mongodriver</artifactId>
+	<version>1.0.0</version>
+	<name>mongodb36</name>
+	<url>http://maven.apache.org</url>
+	<properties>
+		<mongo.driver.version>3.6.0</mongo.driver.version>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.mongodb</groupId>
+			<artifactId>mongo-java-driver</artifactId>
+			<version>${mongo.driver.version}</version>
+		</dependency>
+	</dependencies>
+</project>

mongodriver/src/main/java/de/codecentric/notifications/DocumentProducer.java

@@ -0,0 +1,28 @@
+package de.codecentric.notifications;
+
+import org.bson.Document;
+
+import com.mongodb.MongoClient;
+import com.mongodb.MongoClientURI;
+import com.mongodb.client.MongoCollection;
+
+public class DocumentProducer  {
+    
+	public static void main(String[] args) throws Exception {
+
+		MongoCollection<Document> eventCollection = 
+				new MongoClient(
+						new MongoClientURI("mongodb://localhost:27001,localhost:27002,localhost:27003/test?replicatSet=demo-dev")
+				).getDatabase("test").getCollection("events");    		
+    		
+    		long i = 0;
+    		while (true) {
+    			Document doc = new Document();
+    			doc.put("i", i++);
+    			doc.put("even", i % 2);
+    			eventCollection.insertOne(doc);
+    			//System.out.println("inserted: " + doc);
+    			Thread.sleep(2000L + (long)(1000*Math.random()));
+    		}
+    }
+}

mongodriver/src/main/java/de/codecentric/notifications/EventListener.java

@@ -0,0 +1,37 @@
+package de.codecentric.notifications;
+
+import static java.util.Arrays.asList;
+
+import org.bson.Document;
+
+import com.mongodb.Block;
+import com.mongodb.MongoClient;
+import com.mongodb.MongoClientURI;
+import com.mongodb.client.ChangeStreamIterable;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.model.Aggregates;
+import static com.mongodb.client.model.Filters.*;
+import com.mongodb.client.model.changestream.ChangeStreamDocument;
+
+public class EventListener  {
+    public static void main(String[] args) throws Exception {
+    	
+    		MongoCollection<Document> eventCollection = 
+    				new MongoClient(
+    						new MongoClientURI("mongodb://localhost:27001,localhost:27002,localhost:27003/test?replicatSet=demo-dev")
+    				).getDatabase("test").getCollection("events");
+	
+    		ChangeStreamIterable<Document> changes = eventCollection.watch(asList( 
+    				Aggregates.match( and( asList( 
+    					in("operationType", asList("insert")),
+    					eq("fullDocument.even", 1L)))
+    	    )));
+    		
+    		changes.forEach(new Block<ChangeStreamDocument<Document>>() {
+				@Override
+				public void apply(ChangeStreamDocument<Document> t) {
+					System.out.println("received: " + t.getFullDocument());
+				}
+    		});
+    }
+}

pom.xml

@@ -15,6 +15,7 @@
 		<module>springdata-neo4j</module>
 		<module>springdata-jpa</module>
 		<module>spring-boot-mongo</module>
+		<module>mongodb36</module>
 	</modules>
 
 	<properties>