MongoSync

Author: sfxcode

build.sbt

@@ -59,7 +59,7 @@ libraryDependencies += "org.mongodb.scala" %% "mongo-scala-driver" % "4.0.3"
 
 libraryDependencies += "org.xerial.snappy" % "snappy-java" % "1.1.7.5" % Provided
 
-libraryDependencies += "com.github.luben" % "zstd-jni" % "1.4.4-11" % Provided
+libraryDependencies += "com.github.luben" % "zstd-jni" % "1.4.5-2" % Provided
 
 libraryDependencies += "de.bwaldvogel" % "mongo-java-server" % "1.29.1" % Provided
 

project/build.properties

@@ -1 +1 @@
-sbt.version=1.3.10
+sbt.version=1.3.11

src/main/reference.conf renamed to src/main/resources/reference.conf

@@ -3,4 +3,6 @@ com.sfxcode.nosql.mongo.sync {
   maxWait = 600
   syncColumnLastSync = "_lastSync"
   syncColumnLastUpdate = "_lastUpdate"
+  writeSyncLogOnMaster = false
+  syncLogTableName = "mongo-sync-log"
 }

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.{ConfigHelper, MongoIndex}
 import com.typesafe.scalalogging.LazyLogging
 import org.mongodb.scala.bson.conversions.Bson
 import org.mongodb.scala.model.Sorts._

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

@@ -1,5 +1,7 @@
 package com.sfxcode.nosql.mongo.operation
 
+import java.util.Date
+
 import com.sfxcode.nosql.mongo.database.DatabaseProvider
 import com.sfxcode.nosql.mongo.{Converter, _}
 import org.mongodb.scala.bson.conversions.Bson
@@ -10,6 +12,8 @@ import org.mongodb.scala.{BulkWriteResult, Observable, SingleObservable}
 
 import scala.collection.mutable.ArrayBuffer
 import scala.reflect.ClassTag
+import Updates._
+import com.sfxcode.nosql.mongo.sync.MongoSyncOperation
 
 abstract class Crud[A]()(implicit ct: ClassTag[A]) extends Search[A] {
 
@@ -77,6 +81,9 @@ abstract class Crud[A]()(implicit ct: ClassTag[A]) extends Search[A] {
   def updateMany(filter: Bson, update: Bson, options: UpdateOptions): Observable[UpdateResult] =
     coll.updateMany(filter, update, options)
 
+  def touchInternal(filter: Bson): Observable[UpdateResult] =
+    updateMany(filter, set(MongoSyncOperation.SyncColumnLastUpdate, new Date()))
+
   // delete
 
   def deleteOne(filter: Bson): Observable[DeleteResult] = coll.deleteOne(filter)

src/main/scala/com/sfxcode/nosql/mongo/sync/MongoSyncOperation.scala

@@ -71,20 +71,29 @@ case class MongoSyncOperation(
       countBefore: Int,
       diff: Seq[Document]
   ): MongoSyncResult = {
-    val syncInfo: MongoSyncInfo = MongoSyncInfo()
+    val start    = System.currentTimeMillis()
+    val syncDate = new Date()
     if (diff.nonEmpty) {
       val idSet: Set[ObjectId]           = diff.map(doc => doc.getObjectId(idColumnName)).toSet
       val documentsToSync: Seq[Document] = left.dao(collectionName).find(valueFilter(idColumnName, idSet)).results()
       right.dao(collectionName).bulkWriteMany(documentsToSync).result()
       val update = combine(
-        set(MongoSyncOperation.SyncColumnLastSync, syncInfo.syncDate),
-        set(MongoSyncOperation.SyncColumnLastUpdate, syncInfo.updateDate)
+        set(MongoSyncOperation.SyncColumnLastSync, syncDate),
+        set(MongoSyncOperation.SyncColumnLastUpdate, syncDate)
       )
       left.dao(collectionName).updateMany(Map(), update).result()
       right.dao(collectionName).updateMany(Map(), update).result()
     }
     val countAfter: Int = right.dao(collectionName).count().result().toInt
-    MongoSyncResult(collectionName, true, diff.size, countBefore, countAfter, syncInfo)
+    MongoSyncResult(
+      collectionName,
+      syncDate,
+      true,
+      diff.size,
+      countBefore,
+      countAfter,
+      (System.currentTimeMillis() - start)
+    )
   }
 }
 
@@ -97,16 +106,20 @@ object MongoSyncOperation extends ConfigHelper {
   val SyncColumnLastUpdate: String =
     stringConfig(configPath = "com.sfxcode.nosql.mongo.sync", key = "syncColumnLastUpdate", default = "_lastUpdate").get
 
+  val WriteSyncLogOnMaster = booleanConfig(configPath = "com.sfxcode.nosql.mongo.sync", key = "writeSyncLogOnMaster")
+  val SyncLogTableName: String =
+    stringConfig(configPath = "com.sfxcode.nosql.mongo.sync", key = "syncLogTableName", default = "mongo-sync-log").get
 }
 
-case class MongoSyncInfo(id: Any = new ObjectId(), syncDate: Date = new Date(), updateDate: Date = new Date())
+//case class MongoSyncInfo(id: Any = new ObjectId(), syncDate: Date = new Date(), updateDate: Date = new Date())
 
 case class MongoSyncResult(
     collectionName: String,
+    syncDate: Date = new Date(),
     acknowleged: Boolean = false,
     synced: Int = -1,
     countBefore: Int = -1,
     countAfter: Int = -1,
-    syncInfo: MongoSyncInfo = MongoSyncInfo(),
+    syncTime: Long = -1,
     exception: Option[Exception] = None
 )

src/main/scala/com/sfxcode/nosql/mongo/sync/MongoSyncer.scala

@@ -1,13 +1,20 @@
 package com.sfxcode.nosql.mongo.sync
 
+import com.sfxcode.nosql.mongo._
 import com.sfxcode.nosql.mongo.database.{DatabaseProvider, MongoConfig}
+import org.bson.codecs.configuration.CodecRegistries.fromProviders
+import org.mongodb.scala.bson.codecs.Macros._
 
 import scala.collection.mutable
 
 case class MongoSyncer(sourceConfig: MongoConfig, targetConfig: MongoConfig) {
-  val source: DatabaseProvider = DatabaseProvider(sourceConfig)
+  private val registry = fromProviders(classOf[MongoSyncResult])
+
+  val source: DatabaseProvider = DatabaseProvider(sourceConfig, registry)
   val target: DatabaseProvider = DatabaseProvider(targetConfig)
 
+  object MongoSyncResultDAO extends MongoDAO[MongoSyncResult](source, MongoSyncOperation.SyncLogTableName)
+
   var terminated = false
 
   private val operationMap = new mutable.HashMap[String, MongoSyncOperation]()
@@ -19,10 +26,16 @@ case class MongoSyncer(sourceConfig: MongoConfig, targetConfig: MongoConfig) {
     if (terminated) {
       throw MongoSyncException("MongoSyncer already terminated")
     }
-    operationMap
+
+    val result = operationMap
       .get(collectionName)
       .map(op => op.excecute(source, target))
       .getOrElse(List(MongoSyncResult(collectionName)))
+
+    if (MongoSyncOperation.WriteSyncLogOnMaster) {
+      MongoSyncResultDAO.insertMany(result).results()
+    }
+    result
   }
 
   def syncAll(): List[MongoSyncResult] = operationMap.keys.flatMap(key => sync(key)).toList

src/test/resources/application.conf

@@ -1,5 +1,8 @@
 test.mongo.database = "test"
 
+com.sfxcode.nosql.mongo.sync.writeSyncLogOnMaster = true
+
+
 unit.test.mongo {
   database = "simple-mongo-unit-test"
   host = "localhost"

src/test/scala/com/sfxcode/nosql/mongo/sync/SyncSpec.scala

@@ -1,6 +1,7 @@
 package com.sfxcode.nosql.mongo.sync
 
 import com.sfxcode.nosql.mongo.server.LocalServer
+import com.sfxcode.nosql.mongo.sync.TestSync._
 import com.sfxcode.nosql.mongo.test.UniversityDatabase
 import org.specs2.mutable.Specification
 import org.specs2.specification.{AfterAll, BeforeAll}
@@ -20,36 +21,69 @@ class SyncSpec extends Specification with BeforeAll with AfterAll {
   "Collection" should {
 
     "be synced from source to target" in {
-      var result: MongoSyncResult = TestSync.mongoSyncer.sync(TestSync.TestCollectionSourceTargetName).head
+      var result: MongoSyncResult = TestSync.mongoSyncer.sync(TestCollectionSourceTargetName).head
       result.acknowleged must beTrue
-      TestSync.insertIntoSource(500)
-      result = TestSync.mongoSyncer.sync(TestSync.TestCollectionSourceTargetName).head
+      TestSync.insertIntoSource(500, TestCollectionSourceTargetName)
+      result = TestSync.mongoSyncer.sync(TestCollectionSourceTargetName).head
       result.acknowleged must beTrue
       result.synced mustEqual 500
       result.countBefore mustEqual 0
       result.countAfter mustEqual 500
-      TestSync.targetCount mustEqual 500
-      result = TestSync.mongoSyncer.sync(TestSync.TestCollectionSourceTargetName).head
+      TestSync.targetCount(TestCollectionSourceTargetName) mustEqual 500
+      result = TestSync.mongoSyncer.sync(TestCollectionSourceTargetName).head
       result.acknowleged must beTrue
       result.countBefore mustEqual 500
       result.synced mustEqual 0
 
-      TestSync.insertIntoSource(5)
-      result = TestSync.mongoSyncer.sync(TestSync.TestCollectionSourceTargetName).head
+      TestSync.insertIntoSource(5, TestCollectionSourceTargetName)
+      result = TestSync.mongoSyncer.sync(TestCollectionSourceTargetName).head
       result.acknowleged must beTrue
       result.countBefore mustEqual 500
       result.countAfter mustEqual 505
       result.synced mustEqual 5
-      TestSync.targetCount mustEqual 505
+      TestSync.targetCount(TestCollectionSourceTargetName) mustEqual 505
 
-      TestSync.insertIntoTarget(5)
-      TestSync.targetCount mustEqual 510
-      val resultList = TestSync.mongoSyncer.sync(TestSync.TestCollectionSourceTargetName)
+      TestSync.insertIntoTarget(5, TestCollectionSourceTargetName)
+      TestSync.targetCount(TestCollectionSourceTargetName) mustEqual 510
+      val resultList = TestSync.mongoSyncer.sync(TestCollectionSourceTargetName)
       result = resultList.head
       result.acknowleged must beTrue
       result.countBefore mustEqual 510
       result.synced mustEqual 0
-      TestSync.sourceCount mustEqual 505
+      TestSync.sourceCount(TestCollectionSourceTargetName) mustEqual 505
+    }
+
+    "be synced two way" in {
+      var result: MongoSyncResult = TestSync.mongoSyncer.sync(TestCollectionTwoWayName).head
+      result.acknowleged must beTrue
+      TestSync.insertIntoSource(500, TestCollectionTwoWayName)
+      result = TestSync.mongoSyncer.sync(TestCollectionTwoWayName).head
+      result.acknowleged must beTrue
+      result.synced mustEqual 500
+      result.countBefore mustEqual 0
+      result.countAfter mustEqual 500
+      TestSync.targetCount(TestCollectionTwoWayName) mustEqual 500
+      result = TestSync.mongoSyncer.sync(TestCollectionTwoWayName).head
+      result.acknowleged must beTrue
+      result.countBefore mustEqual 500
+      result.synced mustEqual 0
+
+      TestSync.insertIntoSource(5, TestCollectionTwoWayName)
+      result = TestSync.mongoSyncer.sync(TestCollectionTwoWayName).head
+      result.acknowleged must beTrue
+      result.countBefore mustEqual 500
+      result.countAfter mustEqual 505
+      result.synced mustEqual 5
+      TestSync.targetCount(TestCollectionTwoWayName) mustEqual 505
+
+      TestSync.insertIntoTarget(5, TestCollectionTwoWayName)
+      TestSync.targetCount(TestCollectionTwoWayName) mustEqual 510
+      val resultList = TestSync.mongoSyncer.sync(TestCollectionTwoWayName)
+      result = resultList.head
+      result.acknowleged must beTrue
+      result.countBefore mustEqual 510
+      result.synced mustEqual 0
+      TestSync.sourceCount(TestCollectionTwoWayName) mustEqual 510
     }
 
   }

src/test/scala/com/sfxcode/nosql/mongo/sync/TestSync.scala

@@ -6,34 +6,41 @@ import org.mongodb.scala.bson.collection.immutable.Document
 object TestSync {
 
   val TestCollectionSourceTargetName = "sync-test-source-target"
+  val TestCollectionTwoWayName       = "sync-test-two-way"
 
   val mongoSyncer: MongoSyncer =
     MongoSyncer.fromPath(sourceConfigPath = "unit.test.mongo", targetConfigPath = "unit.test.mongo.local")
 
   mongoSyncer.addOperation(MongoSyncOperation(TestCollectionSourceTargetName, SyncDirection.SourceToTarget))
+  mongoSyncer.addOperation(MongoSyncOperation(TestCollectionTwoWayName, SyncDirection.TwoWay))
   reset()
 
-  def insertIntoSource(count: Int = 1): Unit =
+  def insertIntoSource(count: Int = 1, collectionName: String): Unit =
     (1 to count).foreach { _ =>
       mongoSyncer.source
-        .collection(TestCollectionSourceTargetName)
+        .collection(collectionName)
         .insertOne(Document("_id" -> new ObjectId(), "string" -> "Hallo", "long" -> 1))
         .result()
     }
 
-  def insertIntoTarget(count: Int = 1): Unit =
+  def insertIntoTarget(count: Int = 1, collectionName: String): Unit =
     (1 to count).foreach { _ =>
       mongoSyncer.target
-        .collection(TestCollectionSourceTargetName)
+        .collection(collectionName)
         .insertOne(Document("_id" -> new ObjectId(), "string" -> "Hallo", "long" -> 1))
         .result()
     }
 
-  def sourceCount: Long = mongoSyncer.source.dao(TestCollectionSourceTargetName).count().result()
-  def targetCount: Long = mongoSyncer.target.dao(TestCollectionSourceTargetName).count().result()
+  def sourceCount(collectionName: String): Long =
+    mongoSyncer.source.dao(collectionName).count().result()
+  def targetCount(collectionName: String): Long =
+    mongoSyncer.target.dao(collectionName).count().result()
 
   def reset(): Unit = {
     mongoSyncer.source.collection(TestCollectionSourceTargetName).drop().result()
     mongoSyncer.target.collection(TestCollectionSourceTargetName).drop().result()
+
+    mongoSyncer.source.collection(TestCollectionTwoWayName).drop().result()
+    mongoSyncer.target.collection(TestCollectionTwoWayName).drop().result()
   }
 }