add ChangeObserver

Author: sfxcode

CHANGES.md

@@ -2,6 +2,12 @@
 
 ## Versions
 
+### 1.9.1
+* database provider: runCommand added
+* database provider: addChangeObserver added
+* DAO changeObserver added
+* DAO stats (GridFSDAO fileStats, chunkStats) as case class added
+
 ### 1.9.0
 
 * ObservableIncludes changed method names for better understanding =>

src/main/scala/com/sfxcode/nosql/mongo/GridFSDAO.scala

@@ -1,6 +1,6 @@
 package com.sfxcode.nosql.mongo
 
-import com.sfxcode.nosql.mongo.database.DatabaseProvider
+import com.sfxcode.nosql.mongo.database.{ ChangeObserver, CollectionStats, DatabaseProvider }
 import com.sfxcode.nosql.mongo.gridfs.Metadata
 import org.bson.types.ObjectId
 import org.mongodb.scala.bson.conversions.Bson
@@ -13,12 +13,18 @@ abstract class GridFSDAO(provider: DatabaseProvider, bucketName: String) extends
   var bucket: GridFSBucket = {
     if (bucketName.contains(DatabaseProvider.CollectionSeparator)) {
       val newDatabaseName = bucketName.substring(0, bucketName.indexOf(DatabaseProvider.CollectionSeparator))
-      val newBucketName = bucketName.substring(bucketName.indexOf(DatabaseProvider.CollectionSeparator) + 1)
+      val newBucketName   = bucketName.substring(bucketName.indexOf(DatabaseProvider.CollectionSeparator) + 1)
       GridFSBucket(provider.database(newDatabaseName), newBucketName)
     } else {
       GridFSBucket(provider.database(), bucketName)
     }
   }
+  def addChangeObserver(observer: ChangeObserver[Document]): ChangeObserver[Document] =
+    Files.addChangeObserver(observer: ChangeObserver[Document])
+
+  def fileStats: Observable[CollectionStats] = Files.stats
+
+  def chunkStats: Observable[CollectionStats] = Chunks.stats
 
   protected def gridfsBucket: GridFSBucket = bucket
 

src/main/scala/com/sfxcode/nosql/mongo/MongoDAO.scala

@@ -2,32 +2,40 @@ package com.sfxcode.nosql.mongo
 
 import java.nio.charset.Charset
 
-import better.files.{ File, Scanner }
+import better.files.File
 import com.sfxcode.nosql.mongo.bson.DocumentHelper
-import com.sfxcode.nosql.mongo.database.DatabaseProvider
+import com.sfxcode.nosql.mongo.database.{ ChangeObserver, CollectionStats, DatabaseProvider }
 import com.sfxcode.nosql.mongo.operation.Crud
 import org.bson.json.JsonParseException
-import org.mongodb.scala.{ BulkWriteResult, Document, MongoCollection, SingleObservable }
+import org.mongodb.scala.{ BulkWriteResult, Document, MongoCollection, Observable, SingleObservable }
 
 import scala.collection.mutable.ArrayBuffer
 import scala.reflect.ClassTag
 
 /**
- * Created by tom on 20.01.17.
- */
+  * Created by tom on 20.01.17.
+  */
 abstract class MongoDAO[A](provider: DatabaseProvider, collectionName: String)(implicit ct: ClassTag[A])
-  extends Crud[A] {
+    extends Crud[A] {
 
   val collection: MongoCollection[A] = {
     if (collectionName.contains(DatabaseProvider.CollectionSeparator)) {
-      val newDatabaseName = collectionName.substring(0, collectionName.indexOf(DatabaseProvider.CollectionSeparator))
+      val newDatabaseName   = collectionName.substring(0, collectionName.indexOf(DatabaseProvider.CollectionSeparator))
       val newCollectionName = collectionName.substring(collectionName.indexOf(DatabaseProvider.CollectionSeparator) + 1)
       provider.database(newDatabaseName).getCollection[A](newCollectionName)
     } else {
       provider.database().getCollection[A](collectionName)
     }
   }
 
+  def addChangeObserver(observer: ChangeObserver[A]): ChangeObserver[A] = {
+    coll.watch[A]().subscribe(observer)
+    observer
+  }
+
+  def stats: Observable[CollectionStats] =
+    provider.runCommand(Map("collStats" -> collectionName)).map(document => CollectionStats(document))
+
   protected def coll: MongoCollection[A] = collection
 
   // internal object for raw document access

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

@@ -0,0 +1,42 @@
+package com.sfxcode.nosql.mongo.database
+
+import java.util.Date
+
+import com.sfxcode.nosql.mongo._
+import org.mongodb.scala.bson.Document
+
+case class CollectionStats(
+    ns: String,
+    collectionType: String,
+    size: Double,
+    count: Long,
+    storageSize: Double,
+    avgObjSize: Double,
+    nindexes: Double,
+    indexSizes: Map[String, Int],
+    totalIndexSize: Double,
+    indexDetails: Map[String, Map[String, Any]],
+    scaleFactor: Double,
+    ok: Long,
+    fetched: Date = new Date()
+)
+
+object CollectionStats {
+  def apply(document: Document): CollectionStats = {
+    val map = document.asPlainMap
+    CollectionStats(
+      map("ns").toString,
+      map.getOrElse("type", "Standard").toString,
+      map("size").asInstanceOf[Int],
+      map("count").asInstanceOf[Int],
+      map("storageSize").asInstanceOf[Int],
+      map("avgObjSize").asInstanceOf[Int],
+      map("nindexes").asInstanceOf[Int],
+      map("indexSizes").asInstanceOf[Map[String, Int]],
+      map("totalIndexSize").asInstanceOf[Int],
+      map("indexDetails").asInstanceOf[Map[String, Map[String, Any]]],
+      map("scaleFactor").asInstanceOf[Int],
+      map("ok").asInstanceOf[Double].toInt
+    )
+  }
+}

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

@@ -45,6 +45,9 @@ class DatabaseProvider(config: MongoConfig, registry: CodecRegistry) extends Ser
     observer
   }
 
+  def runCommand(document: Document, databaseName: String = config.database): SingleObservable[Document] =
+    database(databaseName).runCommand(document)
+
   def collection(collectionName: String): MongoCollection[Document] =
     dao(collectionName).collection
 

src/main/scala/com/sfxcode/nosql/mongo/gridfs/Metadata.scala

@@ -15,9 +15,11 @@ abstract class Metadata(provider: DatabaseProvider, bucketName: String) extends
 
   def filesCollectionName: String = "%s.%s".format(bucketName, "files")
 
+  def chunksCollectionName: String = "%s.%s".format(bucketName, "chunks")
+
   def updateMetadata(oid: ObjectId, value: Any): Observable[UpdateResult] = {
     val doc: BsonValue = BsonConverter.toBson(value)
-    val result = Files.updateOne(equal("_id", oid), set("metadata", doc))
+    val result         = Files.updateOne(equal("_id", oid), set("metadata", doc))
     result
   }
 
@@ -42,6 +44,7 @@ abstract class Metadata(provider: DatabaseProvider, bucketName: String) extends
   def updateMetadataElement(filter: Bson, key: String, value: Any): Observable[UpdateResult] =
     updateMetadataElements(filter, Map(key -> value))
 
-  object Files extends MongoDAO[Document](provider, filesCollectionName)
+  object Files  extends MongoDAO[Document](provider, filesCollectionName)
+  object Chunks extends MongoDAO[Document](provider, chunksCollectionName)
 
 }

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

@@ -67,9 +67,4 @@ 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/dao/BookDAOSpec.scala

@@ -1,7 +1,7 @@
 package com.sfxcode.nosql.mongo.dao
 
 import better.files.{ File, Resource }
-import com.sfxcode.nosql.mongo.TestDatabase.{ BookDAO, PersonDAO }
+import com.sfxcode.nosql.mongo.TestDatabase.BookDAO
 import com.sfxcode.nosql.mongo._
 import org.specs2.mutable.Specification
 import org.specs2.specification.BeforeAll
@@ -12,6 +12,10 @@ class BookDAOSpec extends Specification with BeforeAll {
   override def beforeAll(): Unit = {
     BookDAO.drop().result()
     BookDAO.importJsonFile(File(Resource.getUrl("json/books.json"))).result()
+
+    val stats = BookDAO.stats.result()
+    stats.count mustEqual 431
+
   }
 
   "BookDAO" should {

src/test/scala/com/sfxcode/nosql/mongo/database/DatabaseProviderSpec.scala

@@ -0,0 +1,27 @@
+package com.sfxcode.nosql.mongo.database
+
+import com.sfxcode.nosql.mongo.TestDatabase._
+import com.sfxcode.nosql.mongo._
+import org.mongodb.scala.Document
+import org.specs2.mutable.Specification
+
+class DatabaseProviderSpec extends Specification {
+
+  "Base Operations" should {
+
+    "must evaluate buildInfo" in {
+
+      val result: Document = provider.runCommand(Map("buildInfo" -> 1)).result()
+
+      result.getDouble("ok") mustEqual 1.0
+    }
+
+    "must evaluate collectionInfo" in {
+
+      val result: Document = provider.runCommand(Map("collStats" -> "books")).result()
+
+      val s = result.asPlainJson
+      result must not beEmpty
+    }
+  }
+}

src/test/scala/com/sfxcode/nosql/mongo/gridfs/GridFSDatabaseSpec.scala

@@ -66,6 +66,15 @@ class GridFSDatabaseSpec extends Specification with GridfsDatabaseFunctions with
 
     }
 
+    "find stats in file in" in {
+      val fileStats  = ImageFilesDAO.fileStats.result()
+      val chunkStats = ImageFilesDAO.chunkStats.result()
+
+      fileStats.count must be greaterThan 0
+      chunkStats.storageSize must be greaterThan 0
+
+    }
+
   }
 
   override def beforeAll(): Unit = {