Массовая вставка/Вставка многих с Play Framework, ReactiveMongo

Question

Я создаю приложение с помощью Play Framework 2.5.0 и ReactiveMongo, и я трачу много времени, застрял на том, что было бы очень легко сделать на большинстве веб-языков ,

Эта вещь вставляет много документов одновременно. Для этого я должен использовать функцию ReactiveMongo bulkInsert.

я нашел this google group, что было очень простой пример, однако именно с 2013 года, и теперь подпись изменилась

из

def bulkInsert[T](enumerator: Enumerator[T]) 

в

def bulkInsert(documents: Stream[P.Document], ordered: Boolean, writeConcern: WriteConcern)(implicit ec: ExecutionContext): Future[MultiBulkWriteResult] 

Так вот я попытался взять этот пример и найти способ конвертировать Enumerator в Stream (не нашел никакого способа сделать это):

val schemasDocs: Seq[JsObject] = { 
    jsonSchemas.fields.map { 
    case (field, value) => Json.obj(field -> value) 
    } 
} 
val enumerator = Enumerator.enumerate(schemasDocs) 
val schemasStream = Source.fromPublisher(Streams.enumeratorToPublisher(enumerator)) // my attempt to turn enumerator into a Stream 
val schemasInsert = { 
    getCollection("schemas").flatMap(
    _.bulkInsert(schemasStream, true) 
) 
} 

Теперь я ловлю себя дайвинг в Акко, ReactiveMongo и Play API, чтобы попытаться создать поток JsObjects из послед из JsObjects ..

Тогда я попробовал другой подход: example с сайта ReactiveMongo в

val bulkDocs = schemasDocs.map(implicitly[collection.ImplicitlyDocumentProducer](_)) 
collection.bulkInsert(ordered=true)(bulkDocs: _*) 

дает мне ошибку, которая, как трудно отлаживать:

type mismatch; found : Seq[reactivemongo.play.json.collection.JSONCollection#ImplicitlyDocumentProducer] required: Seq[x$48.ImplicitlyDocumentProducer] 

Я предпочел бы не использовать потоковые nd использовать второе решение, так как мне не нравятся вещи, которые я не понимаю в своем коде.

Answer 1

Я только что нашел, как обращаться с bulkInsert. Существует пример

build.sbt

... 
libraryDependencies ++= Seq(
    "org.reactivemongo" %% "play2-reactivemongo" % "0.11.14" 
) 
... 

plugins.sbt

addSbtPlugin("com.typesafe.play" % "sbt-plugin" % "2.5.12") 

CxTransactionsRepository.scala

package cx.repository 

import cx.model.CxTransactionEntity 
import play.modules.reactivemongo.ReactiveMongoApi 
import reactivemongo.play.json.collection.JSONCollection 

import scala.concurrent.{ExecutionContext, Future} 

class CxTransactionsRepository @Inject()(val reactiveMongoApi: ReactiveMongoApi)(implicit ec: ExecutionContext){ 

    private val cxTransactionsCollectionFuture: Future[JSONCollection] = reactiveMongoApi.database.map(_.collection[JSONCollection]("cxTransactions")) 

    def bulkInsert(seq: Seq[CxTransactionEntity]): Future[Int] = { 
    for { 
     transactions <- cxTransactionsCollectionFuture 
     writeResult <- transactions.bulkInsert(ordered = false)(seq.map(implicitly[transactions.ImplicitlyDocumentProducer](_)): _*) 
    } yield { 
     writeResult.n 
    } 
    } 

}