Document Extensions added

Author: sfxcode

CHANGES.md

@@ -2,6 +2,14 @@
 
 ## Versions
 
+### 1.9.0
+
+* ObservableIncludes changed method names for better understanding =>
+  - result (A), results (Seq[A]), resultOption (Option[A]), resultList (List[A])
+  - **Breaking Change** : dropped => headResult (use now result)
+  - **Breaking Change** :  changed => result (old Option[A], new A) => use resultOption instead
+* Add Bulkwrite to Crud Operations
+
 ### 1.8.2
 
 * GridFSStreamObserver suports now completed and resultLength
@@ -83,7 +91,7 @@
 
 
 ### 1.4.0
-*  dropped operations with pattern <name>Result (use headResult, list result or implicit conversion instead)
+*  dropped operations with pattern <name>Result (use result, list result or implicit conversion instead)
 *  introduce Database Functions trait
 *  update docs
 

docs/src/main/paradox/dao/base.md

@@ -27,7 +27,7 @@ import com.sfxcode.nosql.mongo._
 
 and append
 
-* .headResult() for Single Result object
+* .result() for Single Result object
 * .resultList() for List result object
 
 or use implicit conversion:

docs/src/main/paradox/features/observable.md

@@ -25,12 +25,12 @@ See [MongoDB Scala Driver Documentation](http://mongodb.github.io/mongo-scala-dr
 
 Convert Observable to result object. Import mongo package is needed.
 
-Functions headResult anf resultList have an optional maxWait in seconds parameter (Default maxWait = 10 seconds).
+Functions result anf resultList have an optional maxWait in seconds parameter (Default maxWait = 10 seconds).
 
 ```scala
   import com.sfxcode.nosql.mongo._
 
-  def restaurantsSize = RestaurantDAO.count().headResult()
+  def restaurantsSize = RestaurantDAO.count().result()
 
   def findAllRestaurants = RestaurantDAO.find().resultList(maxWait = 20)
 

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

@@ -0,0 +1,32 @@
+package com.sfxcode.nosql.mongo.bson.convert
+
+import java.text.{ DateFormat, SimpleDateFormat }
+import java.util.{ Date, TimeZone }
+
+import com.typesafe.scalalogging.LazyLogging
+import org.bson.json.{ Converter, StrictJsonWriter }
+
+object JsonDateTimeConverter {
+  val Converter = new JsonDateTimeConverter
+
+  val UTC: TimeZone = TimeZone.getTimeZone("UTC")
+
+  val DateFormat: SimpleDateFormat = {
+    val f = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'")
+    f.setTimeZone(UTC)
+    f
+  }
+}
+
+class JsonDateTimeConverter(dateFormat: DateFormat = JsonDateTimeConverter.DateFormat)
+    extends Converter[java.lang.Long]
+    with LazyLogging {
+  override def convert(value: java.lang.Long, writer: StrictJsonWriter): Unit =
+    try {
+      val s = dateFormat.format(value)
+      writer.writeString(s)
+    } catch {
+      case e: Exception =>
+        logger.error(String.format("Failed to convert value %d to JSON date", value), e)
+    }
+}

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

@@ -8,7 +8,7 @@ import com.typesafe.scalalogging.LazyLogging
 import org.mongodb.scala.Observer
 
 case class GridFSStreamObserver(outputStream: OutputStream) extends Observer[ByteBuffer] with LazyLogging {
-  val completed    = new AtomicBoolean(false)
+  val completed = new AtomicBoolean(false)
   val resultLength = new AtomicLong(0)
 
   override def onNext(buffer: ByteBuffer): Unit = {

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

@@ -8,8 +8,12 @@ import org.mongodb.scala.model.Filters.equal
 
 abstract class Search extends Base {
 
-  def find(filter: Bson = Document(), sort: Bson = Document()): GridFSFindObservable =
-    gridfsBucket.find(filter).sort(sort)
+  def find(filter: Bson = Document(), sort: Bson = Document(), limit: Int = 0): GridFSFindObservable =
+    if (limit > 0) {
+      gridfsBucket.find(filter).sort(sort).limit(0)
+    } else {
+      gridfsBucket.find(filter).sort(sort)
+    }
 
   def findById(oid: ObjectId): GridFSFindObservable = find(equal("_id", oid))
 

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

@@ -1,12 +1,13 @@
 package com.sfxcode.nosql.mongo.operation
 
 import com.sfxcode.nosql.mongo.{ Converter, _ }
-import org.mongodb.scala.Observable
 import org.mongodb.scala.bson.conversions.Bson
 import org.mongodb.scala.model.Filters._
 import org.mongodb.scala.model._
 import org.mongodb.scala.result.{ DeleteResult, InsertManyResult, InsertOneResult, UpdateResult }
+import org.mongodb.scala.{ BulkWriteResult, Document, Observable, SingleObservable }
 
+import scala.collection.mutable.{ ArrayBuffer, ListBuffer }
 import scala.reflect.ClassTag
 
 abstract class Crud[A]()(implicit ct: ClassTag[A]) extends Search[A] {
@@ -23,6 +24,19 @@ abstract class Crud[A]()(implicit ct: ClassTag[A]) extends Search[A] {
   def insertMany(values: Seq[A], options: InsertManyOptions): Observable[InsertManyResult] =
     coll.insertMany(values, options)
 
+  // bulk write
+
+  def bulkWrite(requests: List[WriteModel[_ <: A]], ordered: Boolean = true): SingleObservable[BulkWriteResult] =
+    coll.bulkWrite(requests, BulkWriteOptions().ordered(ordered))
+
+  def bulkWriteMany(values: Seq[A], ordered: Boolean = true): SingleObservable[BulkWriteResult] = {
+    val requests: ArrayBuffer[WriteModel[_ <: A]] = ArrayBuffer()
+    values.foreach(value => {
+      requests.append(InsertOneModel(value))
+    })
+    bulkWrite(requests.toList, ordered)
+  }
+
   // update
 
   def replaceOne(value: A): Observable[UpdateResult] = {

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

@@ -17,15 +17,6 @@ trait ObservableIncludes {
   implicit class GenericObservable[C](val observable: Observable[C]) extends ImplicitObservable[C] {
     override val debugString: C => String = doc => doc.toString
 
-    def resultList(maxWait: Int = DefaultMaxWait): List[C] =
-      Await.result(asFuture(), Duration(maxWait, TimeUnit.SECONDS)).toList
-
-    def result(maxWait: Int = DefaultMaxWait): Option[C] = {
-      val list =
-        Await.result(asFuture(), Duration(maxWait, TimeUnit.SECONDS)).toList
-      list.headOption
-    }
-
   }
 
   trait ImplicitObservable[C] extends LazyLogging {
@@ -34,11 +25,17 @@ trait ObservableIncludes {
 
     def asFuture(): Future[Seq[C]] = observable.toFuture()
 
+    def result(maxWait: Int = DefaultMaxWait): C =
+      Await.result(observable.head(), Duration(maxWait, TimeUnit.SECONDS))
+
     def results(maxWait: Int = DefaultMaxWait): Seq[C] =
       Await.result(asFuture(), Duration(maxWait, TimeUnit.SECONDS))
 
-    def headResult(maxWait: Int = DefaultMaxWait): C =
-      Await.result(observable.head(), Duration(maxWait, TimeUnit.SECONDS))
+    def resultList(maxWait: Int = DefaultMaxWait): List[C] =
+      Await.result(asFuture(), Duration(maxWait, TimeUnit.SECONDS)).toList
+
+    def resultOption(maxWait: Int = DefaultMaxWait): Option[C] =
+      Await.result(observable.headOption(), Duration(maxWait, TimeUnit.SECONDS))
 
   }
 

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

@@ -14,8 +14,16 @@ abstract class Search[A]()(implicit ct: ClassTag[A]) extends Base[A] {
 
   protected def coll: MongoCollection[A]
 
-  def find(filter: Bson = Document(), sort: Bson = Document(), projection: Bson = Document()): FindObservable[A] =
-    coll.find(filter).sort(sort).projection(projection)
+  def find(
+    filter: Bson = Document(),
+    sort: Bson = Document(),
+    projection: Bson = Document(),
+    limit: Int = 0): FindObservable[A] =
+    if (limit > 0) {
+      coll.find(filter).sort(sort).projection(projection).limit(limit)
+    } else {
+      coll.find(filter).sort(sort).projection(projection)
+    }
 
   def findById(oid: ObjectId): FindObservable[A] = find(equal("_id", oid))
 
@@ -26,9 +34,9 @@ abstract class Search[A]()(implicit ct: ClassTag[A]) extends Base[A] {
     coll.distinct[BsonValue](fieldName, filter)
 
   def distinctResult[S <: Any](fieldName: String, filter: Bson = Document()): Seq[S] =
-    distinct(fieldName, filter).results().map(v => fromBson(v).asInstanceOf[S])
+    distinct(fieldName, filter).resultList().map(v => fromBson(v).asInstanceOf[S])
 
-  def findAggregated(aggregator: Seq[Bson]): AggregateObservable[A] =
-    coll.aggregate(aggregator)
+  def findAggregated(aggregator: Seq[Bson], allowDiskUse: Boolean = false): AggregateObservable[A] =
+    coll.aggregate(aggregator).allowDiskUse(allowDiskUse)
 
 }

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

@@ -1,10 +1,13 @@
 package com.sfxcode.nosql
 
+import com.sfxcode.nosql.mongo.Converter
 import com.sfxcode.nosql.mongo.bson.BsonConverter
+import com.sfxcode.nosql.mongo.bson.convert.JsonDateTimeConverter
 import com.sfxcode.nosql.mongo.database.MongoConfig
 import com.sfxcode.nosql.mongo.gridfs.GridFSStreamObserver
 import com.sfxcode.nosql.mongo.operation.ObservableIncludes
 import org.bson.BsonValue
+import org.bson.json.{ JsonMode, JsonWriterSettings }
 import org.bson.types.ObjectId
 import org.mongodb.scala.bson.conversions.Bson
 import org.mongodb.scala.gridfs.{ GridFSFile, GridFSFindObservable }
@@ -13,16 +16,50 @@ import org.mongodb.scala.{ Document, FindObservable, Observable, ObservableImpli
 import scala.collection.JavaConverters._
 import scala.language.implicitConversions
 
-package object mongo extends ObservableIncludes with ObservableImplicits {
+package object mongo extends MongoIncludes with DocumentIncludes {
+
+  implicit class DocumentExtensions[A <: Document](val document: A) extends AnyVal {
+
+    def asPlainMap: Map[String, Any] =
+      BsonConverter.asMap(document)
+
+    def asPlainJson: String = {
+      val builder = JsonWriterSettings.builder
+        .dateTimeConverter(new JsonDateTimeConverter())
+        .outputMode(JsonMode.RELAXED)
+      document.toJson(builder.build())
+    }
+  }
+}
+
+trait MongoIncludes extends ObservableIncludes with ObservableImplicits {
   implicit def stringToConfig(database: String): MongoConfig = MongoConfig(database)
 
-  implicit def observableToResult[T](obs: Observable[T]): T = obs.headResult()
+  implicit def observableToResult[T](obs: Observable[T]): T = obs.result()
 
   implicit def findObservableToResultList[T](obs: FindObservable[T]): List[T] =
     obs.resultList()
 
-  implicit def findObservableToResultOption[T](obs: FindObservable[T]): Option[T] = obs.result()
+  implicit def findObservableToResultOption[T](obs: FindObservable[T]): Option[T] = obs.resultOption()
+
+  // gridfs
+
+  implicit def gridFSFindObservableToFiles(observable: GridFSFindObservable): List[GridFSFile] =
+    observable.resultList()
+
+  implicit def gridFSFileToObjectId(file: GridFSFile): ObjectId =
+    file.getObjectId
+
+  implicit def gridFSFileToBSonIdValue(file: GridFSFile): BsonValue = file.getId
 
+  implicit def observerToResultLength(observer: GridFSStreamObserver): Long = {
+    while (!observer.completed.get) {}
+    observer.resultLength.get()
+  }
+
+}
+
+trait DocumentIncludes {
   implicit def mapToBson(value: Map[_, _]): Bson = Converter.toDocument(value)
 
   implicit def documentFromJavaMap(map: java.util.Map[String, Any]): Document =
@@ -63,20 +100,4 @@ package object mongo extends ObservableIncludes with ObservableImplicits {
 
   implicit def documentToObjectId(doc: Document): ObjectId =
     doc.getObjectId("_id")
-
-  // gridfs
-
-  implicit def gridFSFindObservableToFiles(observable: GridFSFindObservable): List[GridFSFile] =
-    observable.resultList()
-
-  implicit def gridFSFileToObjectId(file: GridFSFile): ObjectId =
-    file.getObjectId
-
-  implicit def gridFSFileToBSonIdValue(file: GridFSFile): BsonValue = file.getId
-
-  implicit def observerToResultLength(observer: GridFSStreamObserver): Long = {
-    while (!observer.completed.get) {}
-    observer.resultLength.get()
-  }
-
 }