refactor GridFS to reactive streams

Author: sfxcode

CHANGES.md

@@ -1,9 +1,13 @@
 # Changes #
 
 ## Versions
+
+### 1.8.0
+
 * mongo-scala-driver 4.0.2
-* maxWaitQueueSize is removed since 4.0.0
-* 
+* maxWaitQueueSize is removed (since mongo-scala-driver 4.0.0)
+* GridFS Refactoring (Download / Upload use now reactive streams)
+* More Index Functions added
 
 ### 1.7.1
 * [mongo-scala-driver 2.9.0](https://mongodb.github.io/mongo-scala-driver/2.9/changelog/)

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

@@ -1,12 +1,14 @@
 package com.sfxcode.nosql.mongo.gridfs
 
-import java.nio.channels.AsynchronousFileChannel
-import java.nio.file.{OpenOption, Path, StandardOpenOption}
+import java.io.OutputStream
+import java.nio.ByteBuffer
 
+import com.mongodb.client.gridfs.model.GridFSUploadOptions
+import com.sfxcode.nosql.mongo.Converter
 import com.typesafe.scalalogging.LazyLogging
 import org.mongodb.scala.bson.ObjectId
 import org.mongodb.scala.gridfs.{GridFSBucket, GridFSDownloadObservable}
-import org.mongodb.scala.{Observable, ReadConcern, ReadPreference, WriteConcern}
+import org.mongodb.scala.{Document, Observable, ReadConcern, ReadPreference, WriteConcern}
 
 abstract class Base extends LazyLogging {
 
@@ -20,10 +22,6 @@ abstract class Base extends LazyLogging {
     metadataKey
   }
 
-  def download(oid: ObjectId):GridFSDownloadObservable = {
-    gridfsBucket.downloadToObservable(oid)
-  }
-
   def drop(): Observable[Void] = gridfsBucket.drop()
 
   def bucketName: String = gridfsBucket.bucketName
@@ -36,4 +34,32 @@ 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] = {
+    val metadataDocument = {
+      metadata match {
+        case document: Document => document
+        case _                  => Converter.toDocument(metadata)
+      }
+    }
+    val options: GridFSUploadOptions = new GridFSUploadOptions()
+      .chunkSizeBytes(chunkSizeBytes)
+      .metadata(metadataDocument)
+    gridfsBucket.uploadFromObservable(fileName, source, options)
+  }
+
+
+  def download(oid: ObjectId):GridFSDownloadObservable = {
+    gridfsBucket.downloadToObservable(oid)
+  }
+
+  def downloadToStream(oid: ObjectId, outputStream: OutputStream):Unit = {
+    val observable: GridFSDownloadObservable = gridfsBucket.downloadToObservable(oid)
+    observable.subscribe(GridFSStreamObserver(outputStream))
+  }
+
+
 }

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

@@ -1,38 +1,19 @@
 package com.sfxcode.nosql.mongo.gridfs
 
-import java.nio.ByteBuffer
+import java.io.InputStream
 
-import com.mongodb.client.gridfs.model.GridFSUploadOptions
-import com.sfxcode.nosql.mongo.Converter
-import org.bson.types.ObjectId
-import org.mongodb.scala.{ Document, Observable }
+import org.mongodb.scala.bson.ObjectId
+import org.mongodb.scala.{Document, Observable}
 
 abstract class Crud extends Search {
 
   def deleteOne(id: ObjectId): Observable[Void] = gridfsBucket.delete(id)
 
-  def insertOne(fileName: String,
-                source: Observable[ByteBuffer],
-                metadata: AnyRef = Document(),
-                chunkSizeBytes: Int = 1024 * 1204): Observable[ObjectId] = {
-    val metadataDocument = {
-      metadata match {
-        case document: Document => document
-        case _                  => Converter.toDocument(metadata)
-      }
-    }
-    val options: GridFSUploadOptions = new GridFSUploadOptions()
-      .chunkSizeBytes(chunkSizeBytes)
-      .metadata(metadataDocument)
-    gridfsBucket.uploadFromObservable(fileName, source, options)
+  def insertOne( fileName: String,
+                 stream: InputStream,
+                 metadata: AnyRef = Document(),
+                 chunkSizeBytes: Int = 1204 * 256, bufferSize:Int = 1024 * 64): Observable[ObjectId] = {
+    upload(fileName, GridFSStreamObservable(stream, bufferSize), metadata, chunkSizeBytes)
   }
 
-  def upload(fileName: String,
-             source: Observable[ByteBuffer],
-             metadata: AnyRef = Document(),
-             chunkSizeBytes: Int = 1024 * 1204): Observable[ObjectId] =
-    insertOne(fileName, source, metadata, chunkSizeBytes)
-
-
-
 }

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

@@ -0,0 +1,53 @@
+package com.sfxcode.nosql.mongo.gridfs
+
+
+import java.io.InputStream
+import java.util.concurrent.atomic.AtomicBoolean
+import java.nio.ByteBuffer
+
+import com.typesafe.scalalogging.LazyLogging
+import org.mongodb.scala.{Observable, Observer, Subscription}
+
+case class GridFSStreamObservable(inputStream: InputStream, bufferSize:Int = 1024 * 64) extends Observable[ByteBuffer] with LazyLogging{
+  val isPublishing = new AtomicBoolean(false)
+  val buffer = new Array[Byte](bufferSize)
+
+  override def subscribe(subscriber: Observer[_ >: ByteBuffer]): Unit = {
+    isPublishing.set(true)
+    subscriber.onSubscribe(GridFSSubscription(subscriber))
+  }
+
+  case class GridFSSubscription(subscriber: Observer[_ >: ByteBuffer]) extends Subscription  {
+
+    override def request(n: Long): Unit = {
+        try {
+          val len = inputStream.read(buffer)
+          if (len >= 0 && isPublishing.get()) {
+            val byteBuffer = ByteBuffer.wrap(buffer, 0, len)
+            subscriber.onNext(byteBuffer)
+          }
+          else {
+            subscriber.onComplete()
+            inputStream.close()
+          }
+        } catch {
+          case e:InterruptedException =>
+            Thread.currentThread.interrupt()
+            subscriber.onError(e)
+            inputStream.close()
+            logger.error(e.getMessage, e)
+          case t:Throwable =>
+            subscriber.onError(t)
+            inputStream.close()
+            logger.error(t.getMessage, t)
+        }
+    }
+
+    override def unsubscribe(): Unit = {
+      isPublishing.set(false)
+    }
+
+    override def isUnsubscribed: Boolean = ! isPublishing.get()
+  }
+}
+

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

@@ -0,0 +1,26 @@
+package com.sfxcode.nosql.mongo.gridfs
+
+import java.io.OutputStream
+import java.nio.ByteBuffer
+
+import com.typesafe.scalalogging.LazyLogging
+import org.mongodb.scala.Observer
+
+case class GridFSStreamObserver(outputStream: OutputStream) extends Observer[ByteBuffer] with LazyLogging {
+
+  override def onNext(buffer: ByteBuffer): Unit = {
+    val bytes = new Array[Byte](buffer.remaining())
+    buffer.get(bytes, 0, bytes.length)
+    buffer.clear()
+    outputStream.write(bytes)
+  }
+
+  override def onError(e: Throwable): Unit = {
+    logger.error(e.getMessage, e)
+    outputStream.close()
+  }
+
+  override def onComplete(): Unit = {
+    outputStream.close()
+  }
+}

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

@@ -19,4 +19,5 @@ abstract class Search extends Base {
   def findByMetadataValue(key: String, value: Any): GridFSFindObservable =
     find(createMetadataKey(key), value)
 
+
 }

src/test/resources/images/scala-logo.png

@@ -1,11 +1,8 @@
 package com.sfxcode.nosql.mongo.gridfs
 
-import java.io.FileOutputStream
-import java.nio.ByteBuffer
-
 import com.sfxcode.nosql.mongo._
-import org.bson.types.ObjectId
 import com.sfxcode.nosql.mongo.gridfs.GridfsDatabase._
+import org.bson.types.ObjectId
 
 object GridfsDatabaseApp extends App with GridfsDatabaseFunctions {
 

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

@@ -1,13 +1,9 @@
 package com.sfxcode.nosql.mongo.gridfs
 
-import java.io.FileOutputStream
-import java.nio.ByteBuffer
-
 import better.files.File
 import com.sfxcode.nosql.mongo._
 import com.sfxcode.nosql.mongo.gridfs.GridfsDatabase._
 import org.bson.types.ObjectId
-import org.mongodb.scala.Observable
 import org.mongodb.scala.bson.conversions.Bson
 import org.mongodb.scala.gridfs.GridFSFile
 import org.mongodb.scala.result.UpdateResult
@@ -29,18 +25,12 @@ trait GridfsDatabaseFunctions {
 
   def insertImage(path: String, metadata: AnyRef): ObjectId = {
     val file = File(path)
-    val buffer = ByteBuffer.wrap(file.loadBytes)
-    ImageFilesDAO.upload(file.name, Observable(Seq(buffer)), metadata)
+    ImageFilesDAO.insertOne(file.name, file.newInputStream, metadata, bufferSize = 1024 * 4)
   }
 
-  def downloadImage(id: ObjectId, path: String):Unit = {
+  def downloadImage(id: ObjectId, path: String): Unit = {
     val file = File(path).touch()
-
-    val result = ImageFilesDAO.download(id)
-
-    val fc = file.fileChannel.get
-    fc.write(result.asReadOnlyBuffer())
-    fc.close()
+    ImageFilesDAO.downloadToStream(id, file.newOutputStream)
   }
 
   def findImage(id: ObjectId): GridFSFile = ImageFilesDAO.findById(id)
@@ -49,11 +39,9 @@ trait GridfsDatabaseFunctions {
 
   def findImages(key: String, value: String): List[GridFSFile] = ImageFilesDAO.findByMetadataValue(key, value)
 
-  def updateMetadata(oid: ObjectId, value: Any): UpdateResult = {
+  def updateMetadata(oid: ObjectId, value: Any): UpdateResult =
     ImageFilesDAO.updateMetadata(oid, value).headResult()
-  }
 
-  def updateMetadataElements(filter: Bson, elements: Map[String, Any]): UpdateResult = {
+  def updateMetadataElements(filter: Bson, elements: Map[String, Any]): UpdateResult =
     ImageFilesDAO.updateMetadataElements(filter, elements).headResult()
-  }
 }