* minor GridFSStreamObserver update

Author: sfxcode

CHANGES.md

@@ -2,6 +2,10 @@
 
 ## Versions
 
+### 1.8.1
+
+* GridFSStreamObserver suports now completed and resultLength
+
 ### 1.8.0
 
 * mongo-scala-driver 4.0.2

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

@@ -34,15 +34,14 @@ abstract class Base extends LazyLogging {
 
   def readConcern: ReadConcern = gridfsBucket.readConcern
 
-  def upload(
-    fileName: String,
-    source: Observable[ByteBuffer],
-    metadata: AnyRef = Document(),
-    chunkSizeBytes: Int = 1024 * 256): Observable[ObjectId] = {
+  def upload(fileName: String,
+             source: Observable[ByteBuffer],
+             metadata: AnyRef = Document(),
+             chunkSizeBytes: Int = 1024 * 256): Observable[ObjectId] = {
     val metadataDocument = {
       metadata match {
         case document: Document => document
-        case _ => Converter.toDocument(metadata)
+        case _                  => Converter.toDocument(metadata)
       }
     }
     val options: GridFSUploadOptions = new GridFSUploadOptions()
@@ -54,9 +53,11 @@ abstract class Base extends LazyLogging {
   def download(oid: ObjectId): GridFSDownloadObservable =
     gridfsBucket.downloadToObservable(oid)
 
-  def downloadToStream(oid: ObjectId, outputStream: OutputStream): Unit = {
+  def downloadToStream(oid: ObjectId, outputStream: OutputStream): GridFSStreamObserver = {
     val observable: GridFSDownloadObservable = gridfsBucket.downloadToObservable(oid)
-    observable.subscribe(GridFSStreamObserver(outputStream))
+    val observer                             = GridFSStreamObserver(outputStream)
+    observable.subscribe(observer)
+    observer
   }
 
 }

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

@@ -2,14 +2,18 @@ package com.sfxcode.nosql.mongo.gridfs
 
 import java.io.OutputStream
 import java.nio.ByteBuffer
+import java.util.concurrent.atomic.{ AtomicBoolean, AtomicInteger, AtomicLong }
 
 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 resultLength = new AtomicInteger(0)
 
   override def onNext(buffer: ByteBuffer): Unit = {
     val bytes = new Array[Byte](buffer.remaining())
+    resultLength.set(resultLength.get() + bytes.length)
     buffer.get(bytes, 0, bytes.length)
     buffer.clear()
     outputStream.write(bytes)
@@ -18,8 +22,12 @@ case class GridFSStreamObserver(outputStream: OutputStream) extends Observer[Byt
   override def onError(e: Throwable): Unit = {
     logger.error(e.getMessage, e)
     outputStream.close()
+    resultLength.set(-1)
+    completed.set(true)
   }
 
-  override def onComplete(): Unit =
+  override def onComplete(): Unit = {
     outputStream.close()
+    completed.set(true)
+  }
 }

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

@@ -2,6 +2,7 @@ package com.sfxcode.nosql
 
 import com.sfxcode.nosql.mongo.bson.BsonConverter
 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.types.ObjectId
@@ -73,4 +74,9 @@ package object mongo extends ObservableIncludes with ObservableImplicits {
 
   implicit def gridFSFileToBSonIdValue(file: GridFSFile): BsonValue = file.getId
 
+  implicit def observerToResult(observer: GridFSStreamObserver): Int = {
+    while (!observer.completed.get) {}
+    observer.resultLength.get()
+  }
+
 }

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

@@ -1,15 +1,18 @@
 package com.sfxcode.nosql.mongo.gridfs
 
+import better.files.File
 import com.sfxcode.nosql.mongo.GridFSDAO
 import com.sfxcode.nosql.mongo.database.DatabaseProvider
 
 /**
- * GridFS Database Sample
- */
+  * GridFS Database Sample
+  */
 object GridfsDatabase {
   val SourcePath = "src/test/resources/images/"
   val TargetPath = "/tmp/_files/"
 
+  File("/tmp/_files").createIfNotExists()
+
   case class ImageMetadata(name: String, group: String = "logos", version: Int = 1, indexSet: Set[Int] = Set(1, 2, 3))
 
   val provider = DatabaseProvider("test")

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

@@ -29,8 +29,14 @@ trait GridfsDatabaseFunctions {
   }
 
   def downloadImage(id: ObjectId, path: String): Unit = {
-    val file = File(path).touch()
-    ImageFilesDAO.downloadToStream(id, file.newOutputStream)
+    val file      = File(path)
+    val start     = System.currentTimeMillis()
+    val size: Int = ImageFilesDAO.downloadToStream(id, file.newOutputStream)
+
+    println(
+      "file: %s with size %s Bytes written in %s ms "
+        .format(file.pathAsString, size, System.currentTimeMillis() - start)
+    )
   }
 
   def findImage(id: ObjectId): GridFSFile = ImageFilesDAO.findById(id)