mongodb.watch.scala

val client = MongoClient.apply("mongodb.uri")
val database = client.getDatabase("mongodb.database")
val collection = database.getCollection("mongodb.collection")

collection.watch()
  .fullDocument(FullDocument.UPDATE_LOOKUP)
  .subscribe(new Observer[ChangeStreamDocument[Document]] {
    override def onNext(changeDocument: ChangeStreamDocument[Document]): Unit = process(changeDocument)
    override def onError(e: Throwable): Unit = println(s"Error: $e")
    override def onComplete(): Unit = println("Completed")
  }
)