add compressor support

Author: sfxcode

build.sbt

@@ -51,6 +51,10 @@ libraryDependencies += "com.github.pathikrit" %% "better-files" % "3.8.0" % Test
 
 libraryDependencies += "org.mongodb.scala" %% "mongo-scala-driver" % "4.0.2"
 
+libraryDependencies += "org.xerial.snappy" % "snappy-java" % "1.1.7.3" % Provided
+
+libraryDependencies += "com.github.luben" % "zstd-jni" % "1.4.4-9" % Provided
+
 libraryDependencies += "com.typesafe" % "config" % "1.4.0"
 
 libraryDependencies += "com.typesafe.scala-logging" %% "scala-logging" % "3.9.2"

src/main/scala/com/sfxcode/nosql/mongo/bson/convert/JsonDateTimeConverter.scala

@@ -19,8 +19,8 @@ object JsonDateTimeConverter {
 }
 
 class JsonDateTimeConverter(dateFormat: DateFormat = JsonDateTimeConverter.DateFormat)
-    extends Converter[java.lang.Long]
-    with LazyLogging {
+  extends Converter[java.lang.Long]
+  with LazyLogging {
   override def convert(value: java.lang.Long, writer: StrictJsonWriter): Unit =
     try {
       val s = dateFormat.format(value)

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

@@ -2,24 +2,26 @@ package com.sfxcode.nosql.mongo.database
 
 import java.util.concurrent.TimeUnit
 
+import com.mongodb.MongoCompressor
 import com.mongodb.MongoCredential.createCredential
 import com.sfxcode.nosql.mongo.database.MongoConfig._
 import com.typesafe.config.{ Config, ConfigFactory }
 import org.mongodb.scala.connection._
 import org.mongodb.scala.{ MongoClientSettings, MongoCredential, ServerAddress }
 
 import scala.collection.JavaConverters._
-
-case class MongoConfig(
-  database: String,
-  host: String = DefaultHost,
-  port: Int = DefaultPort,
-  applicationName: String = DefaultApplicationName,
-  userName: Option[String] = None,
-  password: Option[String] = None,
-  authDatabase: String = DefaultAuthenticationDatabaseName,
-  poolOptions: MongoPoolOptions = MongoPoolOptions(),
-  customClientSettings: Option[MongoClientSettings] = None) {
+import scala.collection.mutable.ArrayBuffer
+
+case class MongoConfig(database: String,
+                       host: String = DefaultHost,
+                       port: Int = DefaultPort,
+                       applicationName: String = DefaultApplicationName,
+                       userName: Option[String] = None,
+                       password: Option[String] = None,
+                       authDatabase: String = DefaultAuthenticationDatabaseName,
+                       poolOptions: MongoPoolOptions = MongoPoolOptions(),
+                       compressors: List[String] = List(),
+                       customClientSettings: Option[MongoClientSettings] = None) {
 
   val clientSettings: MongoClientSettings = {
     if (customClientSettings.isDefined) {
@@ -36,12 +38,25 @@ case class MongoConfig(
         .maintenanceInitialDelay(poolOptions.maintenanceInitialDelay, TimeUnit.SECONDS)
         .build()
 
+      val compressorList = new ArrayBuffer[MongoCompressor]()
+      compressors.foreach(compression => {
+        if (ComressionSnappy.equalsIgnoreCase(compression)) {
+          compressorList.+=(MongoCompressor.createSnappyCompressor())
+        } else if (ComressionZlib.equalsIgnoreCase(compression)) {
+          compressorList.+=(MongoCompressor.createZlibCompressor())
+        } else if (ComressionZstd.equalsIgnoreCase(compression)) {
+          compressorList.+=(MongoCompressor.createZstdCompressor())
+        }
+      })
+
       val builder = MongoClientSettings
         .builder()
         .applicationName(applicationName)
         .applyToConnectionPoolSettings(
-          (b: com.mongodb.connection.ConnectionPoolSettings.Builder) => b.applySettings(connectionPoolSettings))
+          (b: com.mongodb.connection.ConnectionPoolSettings.Builder) => b.applySettings(connectionPoolSettings)
+        )
         .applyToClusterSettings((b: com.mongodb.connection.ClusterSettings.Builder) => b.applySettings(clusterSettings))
+        .compressorList(compressorList.asJava)
 
       if (userName.isDefined && password.isDefined) {
         val credential: MongoCredential = createCredential(userName.get, authDatabase, password.get.toCharArray)
@@ -55,17 +70,21 @@ case class MongoConfig(
 }
 
 object MongoConfig {
-  val DefaultHost = "127.0.0.1"
-  val DefaultPort = 27017
+  val DefaultHost                       = "127.0.0.1"
+  val DefaultPort                       = 27017
   val DefaultAuthenticationDatabaseName = "admin"
-  val DefaultApplicationName = "simple-mongo-app"
+  val DefaultApplicationName            = "simple-mongo-app"
 
-  val DefaultPoolMaxConnectionIdleTime = 60
-  val DefaultPoolMaxSize = 50
-  val DefaultPoolMinSize = 0
-  val DefaultPoolMaxWaitQueueSize = 500
+  val DefaultPoolMaxConnectionIdleTime   = 60
+  val DefaultPoolMaxSize                 = 50
+  val DefaultPoolMinSize                 = 0
+  val DefaultPoolMaxWaitQueueSize        = 500
   val DefaultPoolMaintenanceInitialDelay = 0
 
+  val ComressionSnappy = "snappy"
+  val ComressionZlib   = "zlib"
+  val ComressionZstd   = "zstd"
+
   val DefaultConfigPathPrefix = "mongo"
 
   def fromPath(configPath: String = DefaultConfigPathPrefix): MongoConfig = {
@@ -89,28 +108,35 @@ object MongoConfig {
         default
       }
 
-    def portConfig: Int =
+    val port: Int =
       if (conf.hasPath("%s.port".format(configPath))) {
         conf.getInt("%s.port".format(configPath))
       } else {
         DefaultPort
       }
 
-    val host = stringConfig("host", DefaultHost).get
-    val port = portConfig
-    val database = stringConfig("database").get
-    val userName = stringConfig("userName")
-    val password = stringConfig("password")
-    val authDatabase = stringConfig("authDatabase", DefaultAuthenticationDatabaseName).get
+    val compressors: List[String] =
+      if (conf.hasPath("%s.compressors".format(configPath))) {
+        conf.getStringList("%s.compressors".format(configPath)).asScala.toList
+      } else {
+        List()
+      }
+
+    val host            = stringConfig("host", DefaultHost).get
+    val database        = stringConfig("database").get
+    val userName        = stringConfig("userName")
+    val password        = stringConfig("password")
+    val authDatabase    = stringConfig("authDatabase", DefaultAuthenticationDatabaseName).get
     val applicationName = stringConfig("applicationName", DefaultApplicationName).get
 
     val poolOptions = MongoPoolOptions(
       poolOptionsConfig("maxConnectionIdleTime", DefaultPoolMaxConnectionIdleTime),
       poolOptionsConfig("maxSize", DefaultPoolMaxSize),
       poolOptionsConfig("minSize", DefaultPoolMinSize),
-      poolOptionsConfig("maintenanceInitialDelay", DefaultPoolMaintenanceInitialDelay))
+      poolOptionsConfig("maintenanceInitialDelay", DefaultPoolMaintenanceInitialDelay)
+    )
 
-    MongoConfig(database, host, port, applicationName, userName, password, authDatabase, poolOptions)
+    MongoConfig(database, host, port, applicationName, userName, password, authDatabase, poolOptions, compressors)
   }
 
 }

src/test/resources/application.conf

@@ -1,5 +1,17 @@
 test.mongo.database = "test"
 
+unit.test.mongo {
+  database = "config_path_test"
+  host = "localhost"
+  port = 27017
+  applicationName = "simple-mongo-config-test"
+  pool {
+    minSize = 5
+    maxSize = 10
+  }
+  compressors = ["snappy"]
+}
+
 config.test.mongo {
   database = "config_path_test"
   host = "localhost"
@@ -9,6 +21,7 @@ config.test.mongo {
     minSize = 5
     maxSize = 100
   }
+  compressors = ["snappy", "unknown"]
 }
 
 config.test.auth.mongo {

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

@@ -17,11 +17,12 @@ object TestDatabase extends ObservableImplicits {
 
   private val personRegistry = fromProviders(classOf[Person], classOf[Friend])
 
-  private val lineRegistry = fromProviders(classOf[Line], classOf[Position])
+  private val lineRegistry      = fromProviders(classOf[Line], classOf[Position])
   private val codecTestRegistry = fromProviders(classOf[CodecTest])
 
   val provider =
-    DatabaseProvider("simple_mongo_test", fromRegistries(bookRegistry, personRegistry, lineRegistry, codecTestRegistry))
+    DatabaseProvider.fromPath("unit.test.mongo",
+                              fromRegistries(bookRegistry, personRegistry, lineRegistry, codecTestRegistry))
 
   object BookDAO extends MongoDAO[Book](provider, "books")
 

version.sbt

@@ -1 +1 @@
-version in ThisBuild := "1.8.7"
+version in ThisBuild := "1.8.8"