add ChangeObserver

Author: sfxcode

src/main/scala/com/sfxcode/nosql/mongo/database/ChangeObserver.scala

@@ -0,0 +1,21 @@
+package com.sfxcode.nosql.mongo.database
+
+import com.typesafe.scalalogging.LazyLogging
+import org.mongodb.scala.model.changestream.ChangeStreamDocument
+import org.mongodb.scala.{ Observer, Subscription }
+
+case class ChangeObserver[A](onChangeCallback: ChangeStreamDocument[A] => Unit)
+    extends Observer[ChangeStreamDocument[A]]
+    with LazyLogging {
+
+  override def onSubscribe(subscription: Subscription): Unit = subscription.request(Long.MaxValue) // Request data
+
+  override def onNext(changeDocument: ChangeStreamDocument[A]): Unit =
+    onChangeCallback(changeDocument)
+
+  override def onError(throwable: Throwable): Unit =
+    logger.error(throwable.getMessage, throwable)
+
+  override def onComplete(): Unit = {}
+
+}

src/main/scala/com/sfxcode/nosql/mongo/database/DatabaseProvider.scala

@@ -40,6 +40,11 @@ class DatabaseProvider(config: MongoConfig, registry: CodecRegistry) extends Ser
     cachedDatabaseMap(databaseName)
   }
 
+  def addChangeObserver(observer: ChangeObserver[Document]): ChangeObserver[Document] = {
+    database().watch().subscribe(observer)
+    observer
+  }
+
   def collection(collectionName: String): MongoCollection[Document] =
     dao(collectionName).collection
 
@@ -60,7 +65,7 @@ class DatabaseProvider(config: MongoConfig, registry: CodecRegistry) extends Ser
     database(name).listCollectionNames()
 
   case class DocumentDao(provider: DatabaseProvider, collectionName: String)
-    extends MongoDAO[Document](this, collectionName)
+      extends MongoDAO[Document](this, collectionName)
 
 }
 
@@ -75,9 +80,8 @@ object DatabaseProvider {
   def apply(config: MongoConfig, registry: CodecRegistry = codecRegistry): DatabaseProvider =
     new DatabaseProvider(config, fromRegistries(registry, CustomRegistry, DEFAULT_CODEC_REGISTRY))
 
-  def fromPath(
-    configPath: String = MongoConfig.DefaultConfigPathPrefix,
-    registry: CodecRegistry = codecRegistry): DatabaseProvider =
+  def fromPath(configPath: String = MongoConfig.DefaultConfigPathPrefix,
+               registry: CodecRegistry = codecRegistry): DatabaseProvider =
     apply(MongoConfig.fromPath(configPath), fromRegistries(registry, CustomRegistry, DEFAULT_CODEC_REGISTRY))
 
 }

src/main/scala/com/sfxcode/nosql/mongo/operation/Base.scala

@@ -1,6 +1,6 @@
 package com.sfxcode.nosql.mongo.operation
 
-import com.sfxcode.nosql.mongo.database.MongoIndex
+import com.sfxcode.nosql.mongo.database.{ ChangeObserver, MongoIndex }
 import com.typesafe.scalalogging.LazyLogging
 import org.mongodb.scala.bson.conversions.Bson
 import org.mongodb.scala.model.Sorts._
@@ -19,26 +19,23 @@ abstract class Base[A]()(implicit ct: ClassTag[A]) extends LazyLogging {
 
   def drop(): Observable[Void] = coll.drop()
 
-  def createIndexForField(
-    fieldName: String,
-    sortAscending: Boolean = true,
-    options: IndexOptions = IndexOptions()): SingleObservable[String] =
+  def createIndexForField(fieldName: String,
+                          sortAscending: Boolean = true,
+                          options: IndexOptions = IndexOptions()): SingleObservable[String] =
     if (sortAscending) {
       createIndex(ascending(fieldName), options)
     } else {
       createIndex(descending(fieldName), options)
     }
 
-  def createIndexForFieldWithName(
-    fieldName: String,
-    sortAscending: Boolean = true,
-    name: String): SingleObservable[String] =
+  def createIndexForFieldWithName(fieldName: String,
+                                  sortAscending: Boolean = true,
+                                  name: String): SingleObservable[String] =
     createIndexForField(fieldName, sortAscending, MongoIndex.indexOptionsWithName(Some(name)))
 
-  def createUniqueIndexForField(
-    fieldName: String,
-    sortAscending: Boolean = true,
-    name: Option[String] = None): SingleObservable[String] =
+  def createUniqueIndexForField(fieldName: String,
+                                sortAscending: Boolean = true,
+                                name: Option[String] = None): SingleObservable[String] =
     createIndexForField(fieldName, sortAscending, MongoIndex.indexOptionsWithName(name).unique(true))
 
   def createHashedIndexForField(fieldName: String, options: IndexOptions = IndexOptions()): SingleObservable[String] =
@@ -47,15 +44,13 @@ abstract class Base[A]()(implicit ct: ClassTag[A]) extends LazyLogging {
   def createTextIndexForField(fieldName: String, options: IndexOptions = IndexOptions()): SingleObservable[String] =
     createIndex(Indexes.text(fieldName), options)
 
-  def createExpiringIndexForField(
-    fieldName: String,
-    duration: Duration,
-    sortAscending: Boolean = true,
-    name: Option[String] = None): SingleObservable[String] =
-    createIndexForField(
-      fieldName,
-      sortAscending,
-      MongoIndex.indexOptionsWithName(name).expireAfter(duration._1, duration._2))
+  def createExpiringIndexForField(fieldName: String,
+                                  duration: Duration,
+                                  sortAscending: Boolean = true,
+                                  name: Option[String] = None): SingleObservable[String] =
+    createIndexForField(fieldName,
+                        sortAscending,
+                        MongoIndex.indexOptionsWithName(name).expireAfter(duration._1, duration._2))
 
   def createIndex(key: Bson, options: IndexOptions = IndexOptions()): SingleObservable[String] =
     coll.createIndex(key, options)
@@ -72,4 +67,9 @@ abstract class Base[A]()(implicit ct: ClassTag[A]) extends LazyLogging {
 
   def hasIndexForField(fieldName: String): Boolean = MongoIndex.hasIndexForFieldWithName(listIndexes, fieldName)
 
+  def addChangeObserver(observer: ChangeObserver[A]): ChangeObserver[A] = {
+    coll.watch[A]().subscribe(observer)
+    observer
+  }
+
 }

src/test/scala/com/sfxcode/nosql/mongo/TestDatabase.scala

@@ -1,18 +1,33 @@
 package com.sfxcode.nosql.mongo
 
-import com.sfxcode.nosql.mongo.database.DatabaseProvider
+import com.mongodb.client.model.changestream.OperationType
+import com.sfxcode.nosql.mongo.database.{ ChangeObserver, DatabaseProvider }
 import com.sfxcode.nosql.mongo.model._
+import com.typesafe.scalalogging.LazyLogging
 import org.bson.codecs.configuration.CodecRegistries._
+import org.mongodb.scala.Document
 import org.mongodb.scala.bson.codecs.Macros._
+import org.mongodb.scala.model.changestream.ChangeStreamDocument
 
-object TestDatabase {
+object TestDatabase extends LazyLogging {
 
   private val registry           = fromProviders(classOf[Person], classOf[Friend], classOf[CodecTest])
   private val universityRegistry = fromProviders(classOf[Student], classOf[Score], classOf[Grade])
 
   val provider =
     DatabaseProvider.fromPath(configPath = "unit.test.mongo", registry = fromRegistries(registry, universityRegistry))
 
+  provider.addChangeObserver(ChangeObserver(consumeDatabaseChanges))
+
+  def consumeDatabaseChanges(changeStreamDocument: ChangeStreamDocument[Document]): Unit =
+    if (changeStreamDocument.getOperationType != OperationType.INSERT) {
+      logger.info(
+        "changed %s:%s with ID: %s".format(changeStreamDocument.getNamespace,
+                                           changeStreamDocument.getOperationType,
+                                           changeStreamDocument.getDocumentKey)
+      )
+    }
+
   object PersonDAO extends MongoDAO[Person](provider, "people")
 
   object BookDAO extends MongoDAO[Book](provider, "books")

src/test/scala/com/sfxcode/nosql/mongo/operation/CrudSpec.scala

@@ -3,18 +3,28 @@ package com.sfxcode.nosql.mongo.operation
 import com.sfxcode.nosql.mongo.TestDatabase._
 import com.sfxcode.nosql.mongo._
 import com.sfxcode.nosql.mongo.dao.PersonSpecification
+import com.sfxcode.nosql.mongo.database.ChangeObserver
 import com.sfxcode.nosql.mongo.model.CodecTest
-import org.specs2.mutable.Specification
-import org.specs2.specification.BeforeAll
+import com.typesafe.scalalogging.LazyLogging
+import org.mongodb.scala.model.changestream.ChangeStreamDocument
 
-class CrudSpec extends PersonSpecification {
+class CrudSpec extends PersonSpecification with LazyLogging {
 
   sequential
 
   override def beforeAll(): Unit = {
     super.beforeAll()
     CodecDao.drop().result()
+    CodecDao.addChangeObserver(ChangeObserver(consumeCodecChanges))
     CodecDao.insertOne(CodecTest()).result()
+
+    def consumeCodecChanges(changeStreamDocument: ChangeStreamDocument[CodecTest]): Unit =
+      logger.info(
+        "codec changed %s:%s with ID: %s".format(changeStreamDocument.getNamespace,
+                                                 changeStreamDocument.getOperationType,
+                                                 changeStreamDocument.getDocumentKey)
+      )
+
   }
 
   "Crud Operations" should {