gridfs handling

Author: sfxcode

CHANGES.md

@@ -2,6 +2,8 @@
 
 ## Versions
 * mongo-scala-driver 4.0.2
+* maxWaitQueueSize is removed since 4.0.0
+* 
 
 ### 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/database/MongoConfig.scala

@@ -33,7 +33,6 @@ case class MongoConfig(
         .maxConnectionIdleTime(poolOptions.maxConnectionIdleTime, TimeUnit.SECONDS)
         .maxSize(poolOptions.maxSize)
         .minSize(poolOptions.minSize)
-        .maxWaitQueueSize(poolOptions.maxWaitQueueSize)
         .maintenanceInitialDelay(poolOptions.maintenanceInitialDelay, TimeUnit.SECONDS)
         .build()
 
@@ -107,7 +106,6 @@ object MongoConfig {
       poolOptionsConfig("maxConnectionIdleTime", DefaultPoolMaxConnectionIdleTime),
       poolOptionsConfig("maxSize", DefaultPoolMaxSize),
       poolOptionsConfig("minSize", DefaultPoolMinSize),
-      poolOptionsConfig("maxWaitQueueSize", DefaultPoolMaxWaitQueueSize),
       poolOptionsConfig("maintenanceInitialDelay", DefaultPoolMaintenanceInitialDelay))
 
     MongoConfig(database, host, port, applicationName, userName, password, authDatabase, poolOptions)

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

@@ -6,13 +6,11 @@ case class MongoPoolOptions(
   maxConnectionIdleTime: Int = DefaultMaxConnectionIdleTime,
   maxSize: Int = DefaultMaxSize,
   minSize: Int = DefaultMinSize,
-  maxWaitQueueSize: Int = DefaultMaxWaitQueueSize,
   maintenanceInitialDelay: Int = DefaultMaintenanceInitialDelay) {}
 
 object MongoPoolOptions {
   val DefaultMaxConnectionIdleTime = 60
   val DefaultMaxSize = 50
   val DefaultMinSize = 0
-  val DefaultMaxWaitQueueSize = 500
   val DefaultMaintenanceInitialDelay = 0
 }

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

@@ -1,13 +1,12 @@
 package com.sfxcode.nosql.mongo.gridfs
 
 import java.nio.channels.AsynchronousFileChannel
-import java.nio.file.{ OpenOption, Path, StandardOpenOption }
+import java.nio.file.{OpenOption, Path, StandardOpenOption}
 
 import com.typesafe.scalalogging.LazyLogging
 import org.mongodb.scala.bson.ObjectId
-import org.mongodb.scala.gridfs.GridFSBucket
-import org.mongodb.scala.gridfs.helpers.AsynchronousChannelHelper.channelToOutputStream
-import org.mongodb.scala.{ Observable, ReadConcern, ReadPreference, WriteConcern }
+import org.mongodb.scala.gridfs.{GridFSBucket, GridFSDownloadObservable}
+import org.mongodb.scala.{Observable, ReadConcern, ReadPreference, WriteConcern}
 
 abstract class Base extends LazyLogging {
 
@@ -21,13 +20,8 @@ abstract class Base extends LazyLogging {
     metadataKey
   }
 
-  def downloadToStream(
-    oid: ObjectId,
-    outputPath: Path,
-    openOptions: Seq[OpenOption] = List(StandardOpenOption.CREATE, StandardOpenOption.WRITE)): Observable[Long] = {
-    val streamToDownloadTo: AsynchronousFileChannel =
-      AsynchronousFileChannel.open(outputPath, openOptions: _*)
-    gridfsBucket.downloadToStream(oid, channelToOutputStream(streamToDownloadTo))
+  def download(oid: ObjectId):GridFSDownloadObservable = {
+    gridfsBucket.downloadToObservable(oid)
   }
 
   def drop(): Observable[Void] = gridfsBucket.drop()

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

@@ -1,35 +1,38 @@
 package com.sfxcode.nosql.mongo.gridfs
 
-import java.io.InputStream
+import java.nio.ByteBuffer
 
 import com.mongodb.client.gridfs.model.GridFSUploadOptions
-import com.mongodb.internal.async.client.gridfs.AsyncInputStream
 import com.sfxcode.nosql.mongo.Converter
 import org.bson.types.ObjectId
-import org.mongodb.scala.gridfs.AsyncInputStream
-import org.mongodb.scala.gridfs.AsyncStreamHelper.toAsyncInputStream
-import org.mongodb.scala.{Document, Observable}
+import org.mongodb.scala.{ Document, Observable }
 
 abstract class Crud extends Search {
 
   def deleteOne(id: ObjectId): Observable[Void] = gridfsBucket.delete(id)
 
-  def insertOne(
-    fileName: String,
-    stream: InputStream,
-    metadata: AnyRef = Document(),
-    chunkSizeBytes: Int = chunkSizeBytes): Observable[ObjectId] = {
-    val streamToUploadFrom: AsyncInputStream = toAsyncInputStream(stream)
+  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)
+        case _                  => Converter.toDocument(metadata)
       }
     }
     val options: GridFSUploadOptions = new GridFSUploadOptions()
-      .chunkSizeBytes(1024 * 1204)
+      .chunkSizeBytes(chunkSizeBytes)
       .metadata(metadataDocument)
-    gridfsBucket.uploadFromStream(fileName, streamToUploadFrom, options)
+    gridfsBucket.uploadFromObservable(fileName, source, options)
   }
 
+  def upload(fileName: String,
+             source: Observable[ByteBuffer],
+             metadata: AnyRef = Document(),
+             chunkSizeBytes: Int = 1024 * 1204): Observable[ObjectId] =
+    insertOne(fileName, source, metadata, chunkSizeBytes)
+
+
+
 }

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

@@ -4,6 +4,7 @@ import com.sfxcode.nosql.mongo.database.DatabaseProvider
 import com.sfxcode.nosql.mongo.model._
 import com.sfxcode.nosql.mongo.operation.CrudObserver
 import org.mongodb.scala._
+import org.mongodb.scala.result.InsertManyResult
 
 object TestDatabase extends ObservableImplicits {
 
@@ -30,7 +31,7 @@ object TestDatabase extends ObservableImplicits {
 
   val persons: List[Person] = Person.personList
 
-  val insertResult: Void = PersonDAO.insertMany(persons)
+  val insertResult: InsertManyResult = PersonDAO.insertMany(persons)
 
   def printDatabaseStatus(): Unit = {
     val count: Long = PersonDAO.count()

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

@@ -3,7 +3,6 @@ package com.sfxcode.nosql.mongo.demo.tour
 import com.sfxcode.nosql.mongo._
 import com.sfxcode.nosql.mongo.model._
 import TestDatabase._
-import org.mongodb.scala.Void
 import org.mongodb.scala.result.{InsertManyResult, InsertOneResult}
 
 object QuickTour extends App {

src/test/scala/com/sfxcode/nosql/mongo/demo/tour/QuickTour.scala

@@ -3,7 +3,7 @@ package com.sfxcode.nosql.mongo.demo.tour
 import com.sfxcode.nosql.mongo._
 import com.sfxcode.nosql.mongo.model._
 import TestDatabase._
-import org.mongodb.scala.Void
+import org.mongodb.scala.result.InsertOneResult
 
 object TestTour extends App {
 
@@ -14,7 +14,7 @@ object TestTour extends App {
 
   val scalaBook = Book(Some(1), "Programming In Scala", 852, Author("Martin Odersky"), List(2, 4, 10))
 
-  val completed: Void = BookDAO.insertOne(scalaBook)
+  val completed: InsertOneResult = BookDAO.insertOne(scalaBook)
 
   assert(BookDAO.count().headResult() == 1)
 

src/test/scala/com/sfxcode/nosql/mongo/demo/tour/TestTour.scala

@@ -1,5 +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._