经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Scala » 查看文章
PICE（4）：MongoDBStreaming - gRPC Protobuf conversion
来源：cnblogs　　作者：雪川大虫　　时间：2018/9/25 20:27:58　　对本文有异议

   前两篇我们介绍了JDBC和Cassandra的gRPC streaming实现。相对MongoDB来说，JDBC和Cassandra支持字符类型的query语句SQL,CQL，所以把query指令转换成protobuf structures是简单直接的。而MongoDB没有提供字符类的query，所以我们必须进行MongoDB query涉及的所有类型与protobuf类型的相互转换，实现gRPC功能会复杂的多。我们在这篇讨论里先介绍MongoDB query的protobuf转换。

在前面的MongoDB-Engine讨论里我们设计了个MGOContext作为JVM内部传递MongoDB query的数据结构：

  1.  case class MGOContext(
  2.                          dbName: String,
  3.                          collName: String,
  4.                          actionType: MGO_ACTION_TYPE = MGO_QUERY,
  5.                          action: Option[MGOCommands] = None
  6.                        ) {
  7.     ctx =>
  8.     def setDbName(name: String): MGOContext = ctx.copy(dbName = name)
  10.     def setCollName(name: String): MGOContext = ctx.copy(collName = name)
  12.     def setActionType(at: MGO_ACTION_TYPE): MGOContext = ctx.copy(actionType = at)
  14.     def setCommand(cmd: MGOCommands): MGOContext = ctx.copy(action = Some(cmd))
  15.   }

 

下面是这个结构支持的action清单：

  1.  object MGOCommands {
  3.     case class Count(filter: Option[Bson], options: Option[Any]) extends MGOCommands
  5.     case class Distict(fieldName: String, filter: Option[Bson]) extends MGOCommands
  7.     /*  org.mongodb.scala.FindObservable
  8.     import com.mongodb.async.client.FindIterable
  9.     val resultDocType = FindIterable[Document]
  10.     val resultOption = FindObservable(resultDocType)
  11.       .maxScan(...)
  12.     .limit(...)
  13.     .sort(...)
  14.     .project(...) */
  15.     case class Find(filter: Option[Bson] = None,
  16.                        andThen: Option[FindObservable[Document] => FindObservable[Document]]= None,
  17.                        firstOnly: Boolean = false) extends MGOCommands
  19.     case class DocumentStream(filter: Option[Bson] = None,
  20.                               andThen: Option[FindObservable[Document] => FindObservable[Document]] = None,
  21.                              ) extends MGOCommands
  23.     case class Aggregate(pipeLine: Seq[Bson]) extends MGOCommands
  25.     case class MapReduce(mapFunction: String, reduceFunction: String) extends MGOCommands
  27.     case class Insert(newdocs: Seq[Document], options: Option[Any] = None) extends MGOCommands
  29.     case class Delete(filter: Bson, options: Option[Any] = None, onlyOne: Boolean = false) extends MGOCommands
  31.     case class Replace(filter: Bson, replacement: Document, options: Option[Any] = None) extends MGOCommands
  33.     case class Update(filter: Bson, update: Bson, options: Option[Any] = None, onlyOne: Boolean = false) extends MGOCommands
  36.     case class BulkWrite(commands: List[WriteModel[Document]], options: Option[Any] = None) extends MGOCommands
  38.   }
  40.   object MGOAdmins {
  42.     case class DropCollection(collName: String) extends MGOCommands
  44.     case class CreateCollection(collName: String, options: Option[Any] = None) extends MGOCommands
  46.     case class ListCollection(dbName: String) extends MGOCommands
  48.     case class CreateView(viewName: String, viewOn: String, pipeline: Seq[Bson], options: Option[Any] = None) extends MGOCommands
  50.     case class CreateIndex(key: Bson, options: Option[Any] = None) extends MGOCommands
  52.     case class DropIndexByName(indexName: String, options: Option[Any] = None) extends MGOCommands
  54.     case class DropIndexByKey(key: Bson, options: Option[Any] = None) extends MGOCommands
  56.     case class DropAllIndexes(options: Option[Any] = None) extends MGOCommands
  58.   }

可以看到，我们必须把Bson、Document、FindObservable这几个类型对应到protobuf格式。下面是.proto文件里的部分内容：

  1. message MGODocument {
  2.   bytes document = 1;
  3. }
  5.  message MGOBson {
  6.   bytes bson = 1;
  7. }
  9. message ResultTransformer { //FindObservable
  10.    int32 optType = 1;
  11.    MGOBson bsonParam = 2;
  12.    int32 valueParam = 3;
  13. }
  15. message MGOAdminOptons {
  16.   string tarName = 1;
  17.   repeated MGOBson bsonParam  = 2;
  18.   OptionAny options = 3;
  19.   string objName = 4;
  20. }
  22. message MGOOperations {  //MGOContext
  23.   string dbName = 1;
  24.   string collName = 2;
  25.   int32 commandType = 3;
  26.   repeated MGOBson bsonParam = 4;
  27.   repeated ResultTransformer resultOptions = 5;
  28.   OptionAny options = 6;
  29.   repeated MGODocument documents = 7;
  30.   google.protobuf.BoolValue only = 8;
  31.   MGOAdminOptons adminOptions = 9;
  32. }

首先，Document是个serializable类，可以直接进行序列/反序列化：

  1.   val po = Document (
  2.     "ponum" -> "po18012301",
  3.     "vendor" -> "The smartphone compay",
  4.     "remarks" -> "urgent, rush order",
  5.     "podtl" -> Seq(
  6.       Document("item" -> "sony smartphone", "price" -> 2389.00, "qty" -> 1239, "packing" -> "standard"),
  7.       Document("item" -> "ericson smartphone", "price" -> 897.00, "qty" -> 1000, "payterm" -> "30 days")
  8.     )
  9.   )
  10.   println(po)
  11.   val pobytes = marshal(po)
  12.   println(s"po bytes: ${pobytes}")
  13.   val po1 = unmarshal[Document](pobytes)
  14.   println(s"back to po document: $po1")

下一个是Bson，它是个java interface：

  1. /**
  2.  * An interface for types that are able to render themselves into a {@code BsonDocument}.
  3.  *
  4.  * @since 3.0
  5.  */
  6. public interface Bson {
  7.     /**
  8.      * Render the filter into a BsonDocument.
  9.      *
  10.      * @param documentClass the document class in scope for the collection.  This parameter may be ignored, but it may be used to alter
  11.      *                      the structure of the returned {@code BsonDocument} based on some knowledge of the document class.
  12.      * @param codecRegistry the codec registry.  This parameter may be ignored, but it may be used to look up {@code Codec} instances for
  13.      *                      the document class or any other related class.
  14.      * @param <TDocument> the type of the document class
  15.      * @return the BsonDocument
  16.      */
  17.     <TDocument> BsonDocument toBsonDocument(Class<TDocument> documentClass, CodecRegistry codecRegistry);
  18. }

Bson只是一个interface，不是serilizable，不过BsonDocument可以：

  1. /**
  2.  * A type-safe container for a BSON document.  This class should NOT be sub-classed by third parties.
  3.  *
  4.  * @since 3.0
  5.  */
  6. public class BsonDocument extends BsonValue implements Map<String, BsonValue>, Cloneable, Bson, Serializable {...}

所以我们可以用BsonDocument来进行序列/反序列后在再用它来构建一个新的Bson对象：

  1.   def bsonToProto(bson: Bson) =
  2.     MGOBson(marshal(bson.toBsonDocument(
  3.       classOf[org.mongodb.scala.bson.collection.immutable.Document],DEFAULT_CODEC_REGISTRY)))
  5.   def protoToBson(proto: MGOBson): Bson = new Bson {
  6.     val bsdoc = unmarshal[BsonDocument](proto.bson)
  7.     override def toBsonDocument[TDocument](documentClass: Class[TDocument], codecRegistry: CodecRegistry): BsonDocument = bsdoc
  8.   }

最后是这个FindObservable：这个类型的应用场景是这样的：

  1.     /*  org.mongodb.scala.FindObservable
  2.     import com.mongodb.async.client.FindIterable
  3.     val resultDocType = FindIterable[Document]
  4.     val resultOption = FindObservable(resultDocType)
  5.       .maxScan(...)
  6.     .limit(...)
  7.     .sort(...)
  8.     .project(...) */
  9.     case class Find(filter: Option[Bson] = None,
  10.                        andThen: Option[FindObservable[Document] => FindObservable[Document]]= None,
  11.                        firstOnly: Boolean = false) extends MGOCommands

FindObservable类型的效果可以是一连串施用的结果，因为是FindObservable[A] => FindObservable[A]这样的款式，所以我们可以用一串FindObservable[Document]来进行序列/反序列化处理，然后再重新串连施用来获得最终的FindObservable。FindObservable对应的protobuf结构如下：

  1. message ResultTransformer { //FindObservable
  2.    int32 optType = 1;
  3.    MGOBson bsonParam = 2;
  4.    int32 valueParam = 3;
  5. }
  7.   type FOD_TYPE       = Int
  8.   val FOD_FIRST       = 0  //def first(): SingleObservable[TResult], return the first item
  9.   val FOD_FILTER      = 1  //def filter(filter: Bson): FindObservable[TResult]
  10.   val FOD_LIMIT       = 2  //def limit(limit: Int): FindObservable[TResult]
  11.   val FOD_SKIP        = 3  //def skip(skip: Int): FindObservable[TResult]
  12.   val FOD_PROJECTION  = 4  //def projection(projection: Bson): FindObservable[TResult]
  13.                            //Sets a document describing the fields to return for all matching documents
  14.   val FOD_SORT        = 5  //def sort(sort: Bson): FindObservable[TResult]
  15.   val FOD_PARTIAL     = 6  //def partial(partial: Boolean): FindObservable[TResult]
  16.                            //Get partial results from a sharded cluster if one or more shards are unreachable (instead of throwing an error)
  17.   val FOD_CURSORTYPE  = 7  //def cursorType(cursorType: CursorType): FindObservable[TResult]
  18.                            //Sets the cursor type
  19.   val FOD_HINT        = 8  //def hint(hint: Bson): FindObservable[TResult]
  20.                            //Sets the hint for which index to use. A null value means no hint is set
  21.   val FOD_MAX         = 9  //def max(max: Bson): FindObservable[TResult]
  22.                            //Sets the exclusive upper bound for a specific index. A null value means no max is set
  23.   val FOD_MIN         = 10 //def min(min: Bson): FindObservable[TResult]
  24.                            //Sets the minimum inclusive lower bound for a specific index. A null value means no max is set
  25.   val FOD_RETURNKEY   = 11 //def returnKey(returnKey: Boolean): FindObservable[TResult]
  26.                            //Sets the returnKey. If true the find operation will return only the index keys in the resulting documents
  27.   val FOD_SHOWRECORDID=12  //def showRecordId(showRecordId: Boolean): FindObservable[TResult]
  28.                            //Sets the showRecordId. Set to true to add a field `\$recordId` to the returned documents
  30.   case class ResultOptions(
  31.                           optType: FOD_TYPE,
  32.                           bsonParam: Option[Bson] = None,
  33.                           valueParam: Int = 0 ){
  34.     def toProto = new sdp.grpc.services.ResultTransformer(
  35.       optType = this.optType,
  36.       bsonParam = this.bsonParam.map {b => sdp.grpc.services.MGOBson(marshal(b))},
  37.       valueParam = this.valueParam
  38.     )
  39.     def toFindObservable: FindObservable[Document] => FindObservable[Document] = find => {
  40.       optType match {
  41.         case  FOD_FIRST        => find
  42.         case  FOD_FILTER       => find.filter(bsonParam.get)
  43.         case  FOD_LIMIT        => find.limit(valueParam)
  44.         case  FOD_SKIP         => find.skip(valueParam)
  45.         case  FOD_PROJECTION   => find.projection(bsonParam.get)
  46.         case  FOD_SORT         => find.sort(bsonParam.get)
  47.         case  FOD_PARTIAL      => find.partial(valueParam != 0)
  48.         case  FOD_CURSORTYPE   => find
  49.         case  FOD_HINT         => find.hint(bsonParam.get)
  50.         case  FOD_MAX          => find.max(bsonParam.get)
  51.         case  FOD_MIN          => find.min(bsonParam.get)
  52.         case  FOD_RETURNKEY    => find.returnKey(valueParam != 0)
  53.         case  FOD_SHOWRECORDID => find.showRecordId(valueParam != 0)
  55.       }
  56.     }
  57.   }
  58.   object ResultOptions {
  59.     def fromProto(msg: sdp.grpc.services.ResultTransformer) = new ResultOptions(
  60.       optType = msg.optType,
  61.       bsonParam = msg.bsonParam.map(b => unmarshal[Bson](b.bson)),
  62.       valueParam = msg.valueParam
  63.     )
  65.   }

我们可以用这个ResultOptions类型的toProto,fromProto来进行protobuf的转换处理。然后用aggregation实现连串施用：

  1.      def toResultTransformer(rts: Seq[ResultTransformer]): FindObservable[Document] => FindObservable[Document] = findObj =>
  2.         rts.foldRight(findObj)((a,b) => ResultOptions.fromProto(a).toFindObservable(b))

下面这个函数示范了Find Context的反序列：

  1.   def CtxFromProto(proto: MGOOperations): MGOContext = proto.commandType match {
  2.     case MGO_COMMAND_FIND => {
  3.       var ctx = new MGOContext(
  4.         dbName = proto.dbName,
  5.         collName = proto.collName,
  6.         actionType = MGO_QUERY,
  7.         action = Some(Find())
  8.       )
  9.       def toResultTransformer(rts: Seq[ResultTransformer]): FindObservable[Document] => FindObservable[Document] = findObj =>
  10.         rts.foldRight(findObj)((a,b) => ResultOptions.fromProto(a).toFindObservable(b))
  12.       (proto.bsonParam, proto.resultOptions, proto.only) match {
  13.         case (Nil, Nil, None) => ctx
  14.         case (Nil, Nil, Some(b)) => ctx.setCommand(Find(None, None, b))
  15.         case (bp,Nil,None) => ctx.setCommand(
  16.           Find(Some(protoToBson(bp.head)),None,false))
  17.         case (bp,Nil,Some(b)) => ctx.setCommand(
  18.           Find(Some(protoToBson(bp.head)),None,b))
  19.         case (bp,fo,None) => {
  20.              ctx.setCommand(
  21.             Find(Some(protoToBson(bp.head)),Some(toResultTransformer(fo)),false))
  22.         }
  23.         case (bp,fo,Some(b)) => {
  24.           ctx.setCommand(
  25.             Find(Some(protoToBson(bp.head)),Some(toResultTransformer(fo)),b))
  26.         }
  27.         case _ => ctx
  28.       }
  29.     }
  30.   }

具体的应用示范例如下：

  1.  val eqState = equal("state","California")
  2.   val proj = exclude("rowid","_id")
  3.   val rtxfmr = Seq(
  4.     ResultOptions(
  5.     optType = FOD_LIMIT,
  6.     valueParam = 3)
  7.     ,ResultOptions(
  8.     optType = FOD_PROJECTION,
  9.     bsonParam = Some(proj))
  10.   )
  12.   val protoCtx = MGOProtoMsg(
  13.     dbName = "testdb",
  14.     collName = "aqmrpt",
  15.     commandType = MGO_COMMAND_FIND,
  16.     bsonParam = Seq(eqState),
  17.     resultOptions = rtxfmr
  18.   ).toProto
  20.   val findCtx = CtxFromProto(protoCtx)
  21.   val futFind = mgoQuery[Seq[Document]](findCtx)
  22.   futFind.onComplete {
  23.     case Success(docs) => docs.asInstanceOf[Seq[Document]].foreach{doc => println(doc.toJson())}
  24.     case Failure(e) => println(e.getMessage)
  25.   } 

下面是本次讨论的部分源代码：

MongoDBEngine.scala

  1. package sdp.mongo.engine
  2. import java.text.SimpleDateFormat
  4. import akka.NotUsed
  5. import akka.stream.alpakka.mongodb.scaladsl._
  6. import akka.stream.scaladsl.{Flow, Sink, Source}
  7. import org.mongodb.scala.MongoClient
  8. import org.mongodb.scala.bson.collection.immutable.Document
  9. import org.bson.conversions.Bson
  10. import org.mongodb.scala._
  11. import org.mongodb.scala.model._
  12. import java.util.Calendar
  14. import scala.collection.JavaConverters._
  15. import sdp.file.Streaming._
  16. import akka.stream.Materializer
  17. import org.mongodb.scala.bson.{BsonArray, BsonBinary}
  19. import scala.concurrent._
  20. import scala.concurrent.duration._
  22. import sdp.logging.LogSupport
  24. object MGOContext {
  25.   type MGO_ACTION_TYPE = Int
  26.   val MGO_QUERY        = 0
  27.   val MGO_UPDATE       = 1
  28.   val MGO_ADMIN        = 2
  31.   trait MGOCommands
  33.   object MGOCommands {
  35.     case class Count(filter: Option[Bson], options: Option[Any]) extends MGOCommands
  37.     case class Distict(fieldName: String, filter: Option[Bson]) extends MGOCommands
  39.     /*  org.mongodb.scala.FindObservable
  40.     import com.mongodb.async.client.FindIterable
  41.     val resultDocType = FindIterable[Document]
  42.     val resultOption = FindObservable(resultDocType)
  43.       .maxScan(...)
  44.     .limit(...)
  45.     .sort(...)
  46.     .project(...) */
  47.     case class Find(filter: Option[Bson] = None,
  48.                        andThen: Option[FindObservable[Document] => FindObservable[Document]]= None,
  49.                        firstOnly: Boolean = false) extends MGOCommands
  51.     case class DocumentStream(filter: Option[Bson] = None,
  52.                               andThen: Option[FindObservable[Document] => FindObservable[Document]] = None,
  53.                              ) extends MGOCommands
  55.     case class Aggregate(pipeLine: Seq[Bson]) extends MGOCommands
  57.     case class MapReduce(mapFunction: String, reduceFunction: String) extends MGOCommands
  59.     case class Insert(newdocs: Seq[Document], options: Option[Any] = None) extends MGOCommands
  61.     case class Delete(filter: Bson, options: Option[Any] = None, onlyOne: Boolean = false) extends MGOCommands
  63.     case class Replace(filter: Bson, replacement: Document, options: Option[Any] = None) extends MGOCommands
  65.     case class Update(filter: Bson, update: Bson, options: Option[Any] = None, onlyOne: Boolean = false) extends MGOCommands
  68.     case class BulkWrite(commands: List[WriteModel[Document]], options: Option[Any] = None) extends MGOCommands
  70.   }
  72.   object MGOAdmins {
  74.     case class DropCollection(collName: String) extends MGOCommands
  76.     case class CreateCollection(collName: String, options: Option[Any] = None) extends MGOCommands
  78.     case class ListCollection(dbName: String) extends MGOCommands
  80.     case class CreateView(viewName: String, viewOn: String, pipeline: Seq[Bson], options: Option[Any] = None) extends MGOCommands
  82.     case class CreateIndex(key: Bson, options: Option[Any] = None) extends MGOCommands
  84.     case class DropIndexByName(indexName: String, options: Option[Any] = None) extends MGOCommands
  86.     case class DropIndexByKey(key: Bson, options: Option[Any] = None) extends MGOCommands
  88.     case class DropAllIndexes(options: Option[Any] = None) extends MGOCommands
  90.   }
  92.   case class MGOContext(
  93.                          dbName: String,
  94.                          collName: String,
  95.                          actionType: MGO_ACTION_TYPE = MGO_QUERY,
  96.                          action: Option[MGOCommands] = None
  97.                        ) {
  98.     ctx =>
  99.     def setDbName(name: String): MGOContext = ctx.copy(dbName = name)
  101.     def setCollName(name: String): MGOContext = ctx.copy(collName = name)
  103.     def setActionType(at: MGO_ACTION_TYPE): MGOContext = ctx.copy(actionType = at)
  105.     def setCommand(cmd: MGOCommands): MGOContext = ctx.copy(action = Some(cmd))
  106.   }
  108.   object MGOContext {
  109.     def apply(db: String, coll: String) = new MGOContext(db, coll)
  110.   }
  112.   case class MGOBatContext(contexts: Seq[MGOContext], tx: Boolean = false) {
  113.     ctxs =>
  114.     def setTx(txopt: Boolean): MGOBatContext = ctxs.copy(tx = txopt)
  115.     def appendContext(ctx: MGOContext): MGOBatContext =
  116.       ctxs.copy(contexts = contexts :+ ctx)
  117.   }
  119.   object MGOBatContext {
  120.     def apply(ctxs: Seq[MGOContext], tx: Boolean = false) = new MGOBatContext(ctxs,tx)
  121.   }
  123.   type MGODate = java.util.Date
  124.   def mgoDate(yyyy: Int, mm: Int, dd: Int): MGODate = {
  125.     val ca = Calendar.getInstance()
  126.     ca.set(yyyy,mm,dd)
  127.     ca.getTime()
  128.   }
  129.   def mgoDateTime(yyyy: Int, mm: Int, dd: Int, hr: Int, min: Int, sec: Int): MGODate = {
  130.     val ca = Calendar.getInstance()
  131.     ca.set(yyyy,mm,dd,hr,min,sec)
  132.     ca.getTime()
  133.   }
  134.   def mgoDateTimeNow: MGODate = {
  135.     val ca = Calendar.getInstance()
  136.     ca.getTime
  137.   }
  140.   def mgoDateToString(dt: MGODate, formatString: String): String = {
  141.     val fmt= new SimpleDateFormat(formatString)
  142.     fmt.format(dt)
  143.   }
  145.   type MGOBlob = BsonBinary
  146.   type MGOArray = BsonArray
  148.   def fileToMGOBlob(fileName: String, timeOut: FiniteDuration = 60 seconds)(
  149.     implicit mat: Materializer) = FileToByteArray(fileName,timeOut)
  151.   def mgoBlobToFile(blob: MGOBlob, fileName: String)(
  152.     implicit mat: Materializer) =  ByteArrayToFile(blob.getData,fileName)
  154.   def mgoGetStringOrNone(doc: Document, fieldName: String) = {
  155.     if (doc.keySet.contains(fieldName))
  156.       Some(doc.getString(fieldName))
  157.     else None
  158.   }
  159.   def mgoGetIntOrNone(doc: Document, fieldName: String) = {
  160.     if (doc.keySet.contains(fieldName))
  161.       Some(doc.getInteger(fieldName))
  162.     else None
  163.   }
  164.   def mgoGetLonggOrNone(doc: Document, fieldName: String) = {
  165.     if (doc.keySet.contains(fieldName))
  166.       Some(doc.getLong(fieldName))
  167.     else None
  168.   }
  169.   def mgoGetDoubleOrNone(doc: Document, fieldName: String) = {
  170.     if (doc.keySet.contains(fieldName))
  171.       Some(doc.getDouble(fieldName))
  172.     else None
  173.   }
  174.   def mgoGetBoolOrNone(doc: Document, fieldName: String) = {
  175.     if (doc.keySet.contains(fieldName))
  176.       Some(doc.getBoolean(fieldName))
  177.     else None
  178.   }
  179.   def mgoGetDateOrNone(doc: Document, fieldName: String) = {
  180.     if (doc.keySet.contains(fieldName))
  181.       Some(doc.getDate(fieldName))
  182.     else None
  183.   }
  184.   def mgoGetBlobOrNone(doc: Document, fieldName: String) = {
  185.     if (doc.keySet.contains(fieldName))
  186.       doc.get(fieldName).asInstanceOf[Option[MGOBlob]]
  187.     else None
  188.   }
  189.   def mgoGetArrayOrNone(doc: Document, fieldName: String) = {
  190.     if (doc.keySet.contains(fieldName))
  191.       doc.get(fieldName).asInstanceOf[Option[MGOArray]]
  192.     else None
  193.   }
  195.   def mgoArrayToDocumentList(arr: MGOArray): scala.collection.immutable.List[org.bson.BsonDocument] = {
  196.     (arr.getValues.asScala.toList)
  197.       .asInstanceOf[scala.collection.immutable.List[org.bson.BsonDocument]]
  198.   }
  200.   type MGOFilterResult = FindObservable[Document] => FindObservable[Document]
  201. }
  202. object MGOEngine extends LogSupport {
  204.   import MGOContext._
  205.   import MGOCommands._
  206.   import MGOAdmins._
  209.   object TxUpdateMode {
  210.     private def mgoTxUpdate(ctxs: MGOBatContext, observable: SingleObservable[ClientSession])(
  211.               implicit client: MongoClient, ec: ExecutionContext): SingleObservable[ClientSession] = {
  212.       log.info(s"mgoTxUpdate> calling ...")
  213.       observable.map(clientSession => {
  215.         val transactionOptions =
  216.           TransactionOptions.builder()
  217.             .readConcern(ReadConcern.SNAPSHOT)
  218.             .writeConcern(WriteConcern.MAJORITY).build()
  220.         clientSession.startTransaction(transactionOptions)
  222.         val fut = Future.traverse(ctxs.contexts) { ctx =>
  223.           mgoUpdateObservable[Completed](ctx).map(identity).toFuture()
  224.         }
  226.         Await.ready(fut, 3 seconds)
  228.         clientSession
  229.       })
  230.     }
  232.     private def commitAndRetry(observable: SingleObservable[Completed]): SingleObservable[Completed] = {
  233.       log.info(s"commitAndRetry> calling ...")
  234.       observable.recoverWith({
  235.         case e: MongoException if e.hasErrorLabel(MongoException.UNKNOWN_TRANSACTION_COMMIT_RESULT_LABEL) => {
  236.           log.warn("commitAndRetry> UnknownTransactionCommitResult, retrying commit operation ...")
  237.           commitAndRetry(observable)
  238.         }
  239.         case e: Exception => {
  240.           log.error(s"commitAndRetry> Exception during commit ...: $e")
  241.           throw e
  242.         }
  243.       })
  244.     }
  246.     private def runTransactionAndRetry(observable: SingleObservable[Completed]): SingleObservable[Completed] = {
  247.       log.info(s"runTransactionAndRetry> calling ...")
  248.       observable.recoverWith({
  249.         case e: MongoException if e.hasErrorLabel(MongoException.TRANSIENT_TRANSACTION_ERROR_LABEL) => {
  250.           log.warn("runTransactionAndRetry> TransientTransactionError, aborting transaction and retrying ...")
  251.           runTransactionAndRetry(observable)
  252.         }
  253.       })
  254.     }
  256.     def mgoTxBatch(ctxs: MGOBatContext)(
  257.             implicit client: MongoClient, ec: ExecutionContext): Future[Completed] = {
  259.       log.info(s"mgoTxBatch>  MGOBatContext: ${ctxs}")
  261.       val updateObservable: Observable[ClientSession] = mgoTxUpdate(ctxs, client.startSession())
  262.       val commitTransactionObservable: SingleObservable[Completed] =
  263.         updateObservable.flatMap(clientSession => clientSession.commitTransaction())
  264.       val commitAndRetryObservable: SingleObservable[Completed] = commitAndRetry(commitTransactionObservable)
  266.       runTransactionAndRetry(commitAndRetryObservable)
  267.     }.toFuture()
  268.   }
  271.   def mgoUpdateBatch(ctxs: MGOBatContext)(implicit client: MongoClient, ec: ExecutionContext): Future[Completed] = {
  272.     log.info(s"mgoUpdateBatch>  MGOBatContext: ${ctxs}")
  273.     if (ctxs.tx) {
  274.         TxUpdateMode.mgoTxBatch(ctxs)
  275.       } else {
  277.         val fut = Future.traverse(ctxs.contexts) { ctx =>
  278.           mgoUpdate[Completed](ctx).map(identity) }
  280.         Await.ready(fut, 3 seconds)
  281.         Future.successful(new Completed)
  283.       }
  285.   }
  287.   // T => FindIterable  e.g List[Document]
  288.   def mgoQuery[T](ctx: MGOContext, Converter: Option[Document => Any] = None)(implicit client: MongoClient): Future[T] = {
  289.     log.info(s"mgoQuery>  MGOContext: ${ctx}")
  291.     val db = client.getDatabase(ctx.dbName)
  292.     val coll = db.getCollection(ctx.collName)
  293.     if ( ctx.action == None) {
  294.       log.error(s"mgoQuery> uery action cannot be null!")
  295.       throw new IllegalArgumentException("query action cannot be null!")
  296.     }
  297.     ctx.action.get match {
  298.       /* count */
  299.       case Count(Some(filter), Some(opt)) => //SingleObservable
  300.         coll.countDocuments(filter, opt.asInstanceOf[CountOptions])
  301.           .toFuture().asInstanceOf[Future[T]]
  302.       case Count(Some(filter), None) => //SingleObservable
  303.         coll.countDocuments(filter).toFuture()
  304.           .asInstanceOf[Future[T]]
  305.       case Count(None, None) => //SingleObservable
  306.         coll.countDocuments().toFuture()
  307.           .asInstanceOf[Future[T]]
  308.       /* distinct */
  309.       case Distict(field, Some(filter)) => //DistinctObservable
  310.         coll.distinct(field, filter).toFuture()
  311.           .asInstanceOf[Future[T]]
  312.       case Distict(field, None) => //DistinctObservable
  313.         coll.distinct((field)).toFuture()
  314.           .asInstanceOf[Future[T]]
  315.       /* find */
  316.       case Find(None, None, false) => //FindObservable
  317.         if (Converter == None) coll.find().toFuture().asInstanceOf[Future[T]]
  318.         else coll.find().map(Converter.get).toFuture().asInstanceOf[Future[T]]
  319.       case Find(None, None, true) => //FindObservable
  320.         if (Converter == None) coll.find().first().head().asInstanceOf[Future[T]]
  321.         else coll.find().first().map(Converter.get).head().asInstanceOf[Future[T]]
  322.       case Find(Some(filter), None, false) => //FindObservable
  323.         if (Converter == None) coll.find(filter).toFuture().asInstanceOf[Future[T]]
  324.         else coll.find(filter).map(Converter.get).toFuture().asInstanceOf[Future[T]]
  325.       case Find(Some(filter), None, true) => //FindObservable
  326.         if (Converter == None) coll.find(filter).first().head().asInstanceOf[Future[T]]
  327.         else coll.find(filter).first().map(Converter.get).head().asInstanceOf[Future[T]]
  328.       case Find(None, Some(next), _) => //FindObservable
  329.         if (Converter == None) next(coll.find[Document]()).toFuture().asInstanceOf[Future[T]]
  330.         else next(coll.find[Document]()).map(Converter.get).toFuture().asInstanceOf[Future[T]]
  331.       case Find(Some(filter), Some(next), _) => //FindObservable
  332.         if (Converter == None) next(coll.find[Document](filter)).toFuture().asInstanceOf[Future[T]]
  333.         else next(coll.find[Document](filter)).map(Converter.get).toFuture().asInstanceOf[Future[T]]
  334.       /* aggregate AggregateObservable*/
  335.       case Aggregate(pline) => coll.aggregate(pline).toFuture().asInstanceOf[Future[T]]
  336.       /* mapReduce MapReduceObservable*/
  337.       case MapReduce(mf, rf) => coll.mapReduce(mf, rf).toFuture().asInstanceOf[Future[T]]
  338.       /* list collection */
  339.       case ListCollection(dbName) =>   //ListConllectionObservable
  340.         client.getDatabase(dbName).listCollections().toFuture().asInstanceOf[Future[T]]
  342.     }
  343.   }
  345.   //T => Completed, result.UpdateResult, result.DeleteResult
  346.   def mgoUpdate[T](ctx: MGOContext)(implicit client: MongoClient): Future[T] =
  347.     mgoUpdateObservable[T](ctx).toFuture()
  349.   def mgoUpdateObservable[T](ctx: MGOContext)(implicit client: MongoClient): SingleObservable[T] = {
  350.     log.info(s"mgoUpdateObservable>  MGOContext: ${ctx}")
  352.     val db = client.getDatabase(ctx.dbName)
  353.     val coll = db.getCollection(ctx.collName)
  354.     if ( ctx.action == None) {
  355.       log.error(s"mgoUpdateObservable> uery action cannot be null!")
  356.       throw new IllegalArgumentException("query action cannot be null!")
  357.     }
  358.     ctx.action.get match {
  359.       /* insert */
  360.       case Insert(docs, Some(opt)) =>                  //SingleObservable[Completed]
  361.         if (docs.size > 1)
  362.           coll.insertMany(docs, opt.asInstanceOf[InsertManyOptions]).asInstanceOf[SingleObservable[T]]
  363.         else coll.insertOne(docs.head, opt.asInstanceOf[InsertOneOptions]).asInstanceOf[SingleObservable[T]]
  364.       case Insert(docs, None) =>                       //SingleObservable
  365.         if (docs.size > 1) coll.insertMany(docs).asInstanceOf[SingleObservable[T]]
  366.         else coll.insertOne(docs.head).asInstanceOf[SingleObservable[T]]
  367.       /* delete */
  368.       case Delete(filter, None, onlyOne) =>            //SingleObservable
  369.         if (onlyOne) coll.deleteOne(filter).asInstanceOf[SingleObservable[T]]
  370.         else coll.deleteMany(filter).asInstanceOf[SingleObservable[T]]
  371.       case Delete(filter, Some(opt), onlyOne) =>       //SingleObservable
  372.         if (onlyOne) coll.deleteOne(filter, opt.asInstanceOf[DeleteOptions]).asInstanceOf[SingleObservable[T]]
  373.         else coll.deleteMany(filter, opt.asInstanceOf[DeleteOptions]).asInstanceOf[SingleObservable[T]]
  374.       /* replace */
  375.       case Replace(filter, replacement, None) =>        //SingleObservable
  376.         coll.replaceOne(filter, replacement).asInstanceOf[SingleObservable[T]]
  377.       case Replace(filter, replacement, Some(opt)) =>    //SingleObservable
  378.         coll.replaceOne(filter, replacement, opt.asInstanceOf[UpdateOptions]).asInstanceOf[SingleObservable[T]]
  379.       /* update */
  380.       case Update(filter, update, None, onlyOne) =>      //SingleObservable
  381.         if (onlyOne) coll.updateOne(filter, update).asInstanceOf[SingleObservable[T]]
  382.         else coll.updateMany(filter, update).asInstanceOf[SingleObservable[T]]
  383.       case Update(filter, update, Some(opt), onlyOne) => //SingleObservable
  384.         if (onlyOne) coll.updateOne(filter, update, opt.asInstanceOf[UpdateOptions]).asInstanceOf[SingleObservable[T]]
  385.         else coll.updateMany(filter, update, opt.asInstanceOf[UpdateOptions]).asInstanceOf[SingleObservable[T]]
  386.       /* bulkWrite */
  387.       case BulkWrite(commands, None) =>                  //SingleObservable
  388.         coll.bulkWrite(commands).asInstanceOf[SingleObservable[T]]
  389.       case BulkWrite(commands, Some(opt)) =>             //SingleObservable
  390.         coll.bulkWrite(commands, opt.asInstanceOf[BulkWriteOptions]).asInstanceOf[SingleObservable[T]]
  391.     }
  392.   }
  394.   def mgoAdmin(ctx: MGOContext)(implicit client: MongoClient): SingleObservable[Completed] = {
  395.     log.info(s"mgoAdmin>  MGOContext: ${ctx}")
  397.     val db = client.getDatabase(ctx.dbName)
  398.     val coll = db.getCollection(ctx.collName)
  399.     if ( ctx.action == None) {
  400.       log.error(s"mgoAdmin> uery action cannot be null!")
  401.       throw new IllegalArgumentException("query action cannot be null!")
  402.     }
  403.     ctx.action.get match {
  404.       /* drop collection */
  405.       case DropCollection(collName) =>                   //SingleObservable
  406.         val coll = db.getCollection(collName)
  407.         coll.drop()
  408.       /* create collection */
  409.       case CreateCollection(collName, None) =>           //SingleObservable
  410.         db.createCollection(collName)
  411.       case CreateCollection(collName, Some(opt)) =>      //SingleObservable
  412.         db.createCollection(collName, opt.asInstanceOf[CreateCollectionOptions])
  413.       /* list collection
  414.       case ListCollection(dbName) =>   //ListConllectionObservable
  415.         client.getDatabase(dbName).listCollections().toFuture().asInstanceOf[Future[T]]
  416.         */
  417.       /* create view */
  418.       case CreateView(viewName, viewOn, pline, None) =>       //SingleObservable
  419.         db.createView(viewName, viewOn, pline)
  420.       case CreateView(viewName, viewOn, pline, Some(opt)) =>  //SingleObservable
  421.         db.createView(viewName, viewOn, pline, opt.asInstanceOf[CreateViewOptions])
  422.       /* create index */
  423.       case CreateIndex(key, None) =>                     //SingleObservable
  424.         coll.createIndex(key).asInstanceOf[SingleObservable[Completed]]
  425.       case CreateIndex(key, Some(opt)) =>                //SingleObservable
  426.         coll.createIndex(key, opt.asInstanceOf[IndexOptions]).asInstanceOf[SingleObservable[Completed]]
  427.       /* drop index */
  428.       case DropIndexByName(indexName, None) =>           //SingleObservable
  429.         coll.dropIndex(indexName)
  430.       case DropIndexByName(indexName, Some(opt)) =>      //SingleObservable
  431.         coll.dropIndex(indexName, opt.asInstanceOf[DropIndexOptions])
  432.       case DropIndexByKey(key, None) =>                  //SingleObservable
  433.         coll.dropIndex(key)
  434.       case DropIndexByKey(key, Some(opt)) =>             //SingleObservable
  435.         coll.dropIndex(key, opt.asInstanceOf[DropIndexOptions])
  436.       case DropAllIndexes(None) =>                       //SingleObservable
  437.         coll.dropIndexes()
  438.       case DropAllIndexes(Some(opt)) =>                  //SingleObservable
  439.         coll.dropIndexes(opt.asInstanceOf[DropIndexOptions])
  440.     }
  441.   }
  443. /*
  444.     def mgoExecute[T](ctx: MGOContext)(implicit client: MongoClient): Future[T] = {
  445.     val db = client.getDatabase(ctx.dbName)
  446.     val coll = db.getCollection(ctx.collName)
  447.     ctx.action match {
  448.       /* count */
  449.       case Count(Some(filter), Some(opt)) =>   //SingleObservable
  450.         coll.countDocuments(filter, opt.asInstanceOf[CountOptions])
  451.           .toFuture().asInstanceOf[Future[T]]
  452.       case Count(Some(filter), None) =>        //SingleObservable
  453.         coll.countDocuments(filter).toFuture()
  454.           .asInstanceOf[Future[T]]
  455.       case Count(None, None) =>                //SingleObservable
  456.         coll.countDocuments().toFuture()
  457.           .asInstanceOf[Future[T]]
  458.       /* distinct */
  459.       case Distict(field, Some(filter)) =>     //DistinctObservable
  460.         coll.distinct(field, filter).toFuture()
  461.           .asInstanceOf[Future[T]]
  462.       case Distict(field, None) =>             //DistinctObservable
  463.         coll.distinct((field)).toFuture()
  464.           .asInstanceOf[Future[T]]
  465.       /* find */
  466.       case Find(None, None, optConv, false) =>  //FindObservable
  467.         if (optConv == None) coll.find().toFuture().asInstanceOf[Future[T]]
  468.         else coll.find().map(optConv.get).toFuture().asInstanceOf[Future[T]]
  469.       case Find(None, None, optConv, true) =>   //FindObservable
  470.         if (optConv == None) coll.find().first().head().asInstanceOf[Future[T]]
  471.         else coll.find().first().map(optConv.get).head().asInstanceOf[Future[T]]
  472.       case Find(Some(filter), None, optConv, false) =>   //FindObservable
  473.         if (optConv == None) coll.find(filter).toFuture().asInstanceOf[Future[T]]
  474.         else coll.find(filter).map(optConv.get).toFuture().asInstanceOf[Future[T]]
  475.       case Find(Some(filter), None, optConv, true) =>   //FindObservable
  476.         if (optConv == None) coll.find(filter).first().head().asInstanceOf[Future[T]]
  477.         else coll.find(filter).first().map(optConv.get).head().asInstanceOf[Future[T]]
  478.       case Find(None, Some(next), optConv, _) =>   //FindObservable
  479.         if (optConv == None) next(coll.find[Document]()).toFuture().asInstanceOf[Future[T]]
  480.         else next(coll.find[Document]()).map(optConv.get).toFuture().asInstanceOf[Future[T]]
  481.       case Find(Some(filter), Some(next), optConv, _) =>  //FindObservable
  482.         if (optConv == None) next(coll.find[Document](filter)).toFuture().asInstanceOf[Future[T]]
  483.         else next(coll.find[Document](filter)).map(optConv.get).toFuture().asInstanceOf[Future[T]]
  484.       /* aggregate AggregateObservable*/
  485.       case Aggregate(pline) => coll.aggregate(pline).toFuture().asInstanceOf[Future[T]]
  486.       /* mapReduce MapReduceObservable*/
  487.       case MapReduce(mf, rf) => coll.mapReduce(mf, rf).toFuture().asInstanceOf[Future[T]]
  488.       /* insert */
  489.       case Insert(docs, Some(opt)) =>                  //SingleObservable[Completed]
  490.         if (docs.size > 1) coll.insertMany(docs, opt.asInstanceOf[InsertManyOptions]).toFuture()
  491.           .asInstanceOf[Future[T]]
  492.         else coll.insertOne(docs.head, opt.asInstanceOf[InsertOneOptions]).toFuture()
  493.           .asInstanceOf[Future[T]]
  494.       case Insert(docs, None) =>                       //SingleObservable
  495.         if (docs.size > 1) coll.insertMany(docs).toFuture().asInstanceOf[Future[T]]
  496.         else coll.insertOne(docs.head).toFuture().asInstanceOf[Future[T]]
  497.       /* delete */
  498.       case Delete(filter, None, onlyOne) =>            //SingleObservable
  499.         if (onlyOne) coll.deleteOne(filter).toFuture().asInstanceOf[Future[T]]
  500.         else coll.deleteMany(filter).toFuture().asInstanceOf[Future[T]]
  501.       case Delete(filter, Some(opt), onlyOne) =>       //SingleObservable
  502.         if (onlyOne) coll.deleteOne(filter, opt.asInstanceOf[DeleteOptions]).toFuture().asInstanceOf[Future[T]]
  503.         else coll.deleteMany(filter, opt.asInstanceOf[DeleteOptions]).toFuture().asInstanceOf[Future[T]]
  504.       /* replace */
  505.       case Replace(filter, replacement, None) =>        //SingleObservable
  506.         coll.replaceOne(filter, replacement).toFuture().asInstanceOf[Future[T]]
  507.       case Replace(filter, replacement, Some(opt)) =>    //SingleObservable
  508.         coll.replaceOne(filter, replacement, opt.asInstanceOf[UpdateOptions]).toFuture().asInstanceOf[Future[T]]
  509.       /* update */
  510.       case Update(filter, update, None, onlyOne) =>      //SingleObservable
  511.         if (onlyOne) coll.updateOne(filter, update).toFuture().asInstanceOf[Future[T]]
  512.         else coll.updateMany(filter, update).toFuture().asInstanceOf[Future[T]]
  513.       case Update(filter, update, Some(opt), onlyOne) => //SingleObservable
  514.         if (onlyOne) coll.updateOne(filter, update, opt.asInstanceOf[UpdateOptions]).toFuture().asInstanceOf[Future[T]]
  515.         else coll.updateMany(filter, update, opt.asInstanceOf[UpdateOptions]).toFuture().asInstanceOf[Future[T]]
  516.       /* bulkWrite */
  517.       case BulkWrite(commands, None) =>                  //SingleObservable
  518.         coll.bulkWrite(commands).toFuture().asInstanceOf[Future[T]]
  519.       case BulkWrite(commands, Some(opt)) =>             //SingleObservable
  520.         coll.bulkWrite(commands, opt.asInstanceOf[BulkWriteOptions]).toFuture().asInstanceOf[Future[T]]
  522.       /* drop collection */
  523.       case DropCollection(collName) =>                   //SingleObservable
  524.         val coll = db.getCollection(collName)
  525.         coll.drop().toFuture().asInstanceOf[Future[T]]
  526.       /* create collection */
  527.       case CreateCollection(collName, None) =>           //SingleObservable
  528.         db.createCollection(collName).toFuture().asInstanceOf[Future[T]]
  529.       case CreateCollection(collName, Some(opt)) =>      //SingleObservable
  530.         db.createCollection(collName, opt.asInstanceOf[CreateCollectionOptions]).toFuture().asInstanceOf[Future[T]]
  531.       /* list collection */
  532.       case ListCollection(dbName) =>   //ListConllectionObservable
  533.         client.getDatabase(dbName).listCollections().toFuture().asInstanceOf[Future[T]]
  534.       /* create view */
  535.       case CreateView(viewName, viewOn, pline, None) =>       //SingleObservable
  536.         db.createView(viewName, viewOn, pline).toFuture().asInstanceOf[Future[T]]
  537.       case CreateView(viewName, viewOn, pline, Some(opt)) =>  //SingleObservable
  538.         db.createView(viewName, viewOn, pline, opt.asInstanceOf[CreateViewOptions]).toFuture().asInstanceOf[Future[T]]
  539.       /* create index */
  540.       case CreateIndex(key, None) =>                     //SingleObservable
  541.         coll.createIndex(key).toFuture().asInstanceOf[Future[T]]
  542.       case CreateIndex(key, Some(opt)) =>                //SingleObservable
  543.         coll.createIndex(key, opt.asInstanceOf[IndexOptions]).toFuture().asInstanceOf[Future[T]]
  544.       /* drop index */
  545.       case DropIndexByName(indexName, None) =>           //SingleObservable
  546.         coll.dropIndex(indexName).toFuture().asInstanceOf[Future[T]]
  547.       case DropIndexByName(indexName, Some(opt)) =>      //SingleObservable
  548.         coll.dropIndex(indexName, opt.asInstanceOf[DropIndexOptions]).toFuture().asInstanceOf[Future[T]]
  549.       case DropIndexByKey(key, None) =>                  //SingleObservable
  550.         coll.dropIndex(key).toFuture().asInstanceOf[Future[T]]
  551.       case DropIndexByKey(key, Some(opt)) =>             //SingleObservable
  552.         coll.dropIndex(key, opt.asInstanceOf[DropIndexOptions]).toFuture().asInstanceOf[Future[T]]
  553.       case DropAllIndexes(None) =>                       //SingleObservable
  554.         coll.dropIndexes().toFuture().asInstanceOf[Future[T]]
  555.       case DropAllIndexes(Some(opt)) =>                  //SingleObservable
  556.         coll.dropIndexes(opt.asInstanceOf[DropIndexOptions]).toFuture().asInstanceOf[Future[T]]
  557.     }
  558.   }
  559. */
  562.   def mongoStream(ctx: MGOContext)(
  563.     implicit client: MongoClient, ec: ExecutionContextExecutor): Source[Document, NotUsed] = {
  564.     log.info(s"mongoStream>  MGOContext: ${ctx}")
  566.     val db = client.getDatabase(ctx.dbName)
  567.     val coll = db.getCollection(ctx.collName)
  568.     if ( ctx.action == None) {
  569.       log.error(s"mongoStream> uery action cannot be null!")
  570.       throw new IllegalArgumentException("query action cannot be null!")
  571.     }
  572.     ctx.action.get match {
  573.       case DocumentStream(None, None) =>
  574.         MongoSource(coll.find())
  575.       case DocumentStream(Some(filter), None) =>
  576.         MongoSource(coll.find(filter))
  577.       case DocumentStream(None, Some(next)) =>
  578.         MongoSource(next(coll.find()))
  579.       case DocumentStream(Some(filter), Some(next)) =>
  580.         MongoSource(next(coll.find(filter)))
  581.     }
  582.   }
  584. }
  587. object MongoActionStream {
  589.   import MGOContext._
  591.   case class StreamingInsert[A](dbName: String,
  592.                                 collName: String,
  593.                                 converter: A => Document,
  594.                                 parallelism: Int = 1
  595.                                ) extends MGOCommands
  597.   case class StreamingDelete[A](dbName: String,
  598.                                 collName: String,
  599.                                 toFilter: A => Bson,
  600.                                 parallelism: Int = 1,
  601.                                 justOne: Boolean = false
  602.                                ) extends MGOCommands
  604.   case class StreamingUpdate[A](dbName: String,
  605.                                 collName: String,
  606.                                 toFilter: A => Bson,
  607.                                 toUpdate: A => Bson,
  608.                                 parallelism: Int = 1,
  609.                                 justOne: Boolean = false
  610.                                ) extends MGOCommands
  613.   case class InsertAction[A](ctx: StreamingInsert[A])(
  614.     implicit mongoClient: MongoClient) {
  616.     val database = mongoClient.getDatabase(ctx.dbName)
  617.     val collection = database.getCollection(ctx.collName)
  619.     def performOnRow(implicit ec: ExecutionContext): Flow[A, Document, NotUsed] =
  620.       Flow[A].map(ctx.converter)
  621.         .mapAsync(ctx.parallelism)(doc => collection.insertOne(doc).toFuture().map(_ => doc))
  622.   }
  624.   case class UpdateAction[A](ctx: StreamingUpdate[A])(
  625.     implicit mongoClient: MongoClient) {
  627.     val database = mongoClient.getDatabase(ctx.dbName)
  628.     val collection = database.getCollection(ctx.collName)
  630.     def performOnRow(implicit ec: ExecutionContext): Flow[A, A, NotUsed] =
  631.       if (ctx.justOne) {
  632.         Flow[A]
  633.           .mapAsync(ctx.parallelism)(a =>
  634.             collection.updateOne(ctx.toFilter(a), ctx.toUpdate(a)).toFuture().map(_ => a))
  635.       } else
  636.         Flow[A]
  637.           .mapAsync(ctx.parallelism)(a =>
  638.             collection.updateMany(ctx.toFilter(a), ctx.toUpdate(a)).toFuture().map(_ => a))
  639.   }
  642.   case class DeleteAction[A](ctx: StreamingDelete[A])(
  643.     implicit mongoClient: MongoClient) {
  645.     val database = mongoClient.getDatabase(ctx.dbName)
  646.     val collection = database.getCollection(ctx.collName)
  648.     def performOnRow(implicit ec: ExecutionContext): Flow[A, A, NotUsed] =
  649.       if (ctx.justOne) {
  650.         Flow[A]
  651.           .mapAsync(ctx.parallelism)(a =>
  652.             collection.deleteOne(ctx.toFilter(a)).toFuture().map(_ => a))
  653.       } else
  654.         Flow[A]
  655.           .mapAsync(ctx.parallelism)(a =>
  656.             collection.deleteMany(ctx.toFilter(a)).toFuture().map(_ => a))
  657.   }
  659. }
  661. object MGOHelpers {
  663.   implicit class DocumentObservable[C](val observable: Observable[Document]) extends ImplicitObservable[Document] {
  664.     override val converter: (Document) => String = (doc) => doc.toJson
  665.   }
  667.   implicit class GenericObservable[C](val observable: Observable[C]) extends ImplicitObservable[C] {
  668.     override val converter: (C) => String = (doc) => doc.toString
  669.   }
  671.   trait ImplicitObservable[C] {
  672.     val observable: Observable[C]
  673.     val converter: (C) => String
  675.     def results(): Seq[C] = Await.result(observable.toFuture(), 10 seconds)
  677.     def headResult() = Await.result(observable.head(), 10 seconds)
  679.     def printResults(initial: String = ""): Unit = {
  680.       if (initial.length > 0) print(initial)
  681.       results().foreach(res => println(converter(res)))
  682.     }
  684.     def printHeadResult(initial: String = ""): Unit = println(s"${initial}${converter(headResult())}")
  685.   }
  687.   def getResult[T](fut: Future[T], timeOut: Duration = 1 second): T = {
  688.     Await.result(fut, timeOut)
  689.   }
  691.   def getResults[T](fut: Future[Iterable[T]], timeOut: Duration = 1 second): Iterable[T] = {
  692.     Await.result(fut, timeOut)
  693.   }
  695.   import monix.eval.Task
  696.   import monix.execution.Scheduler.Implicits.global
  698.   final class FutureToTask[A](x: => Future[A]) {
  699.     def asTask: Task[A] = Task.deferFuture[A](x)
  700.   }
  702.   final class TaskToFuture[A](x: => Task[A]) {
  703.     def asFuture: Future[A] = x.runAsync
  704.   }
  706. }

MgoProtoConversion.scala

  1. package sdp.mongo.engine
  2. import org.mongodb.scala.bson.collection.immutable.Document
  3. import org.bson.conversions.Bson
  4. import sdp.grpc.services._
  5. import protobuf.bytes.Converter._
  6. import com.google.protobuf.ByteString
  7. import MGOContext._
  8. import MGOAdmins._
  9. import MGOCommands._
  10. import org.bson.BsonDocument
  11. import org.bson.codecs.configuration.CodecRegistry
  12. import org.mongodb.scala.bson.codecs.DEFAULT_CODEC_REGISTRY
  13. import org.mongodb.scala.FindObservable
  15. object MgoProtoConvertion {
  16.   /*  org.mongodb.scala.FindObservable
  17.   import com.mongodb.async.client.FindIterable
  18.   val resultDocType = FindIterable[Document]
  19.   val resultOption = FindObservable(resultDocType)
  20.     .maxScan(...)
  21.   .limit(...)
  22.   .sort(...)
  23.   .project(...) */
  25.   type FOD_TYPE       = Int
  26.   val FOD_FIRST       = 0  //def first(): SingleObservable[TResult], return the first item
  27.   val FOD_FILTER      = 1  //def filter(filter: Bson): FindObservable[TResult]
  28.   val FOD_LIMIT       = 2  //def limit(limit: Int): FindObservable[TResult]
  29.   val FOD_SKIP        = 3  //def skip(skip: Int): FindObservable[TResult]
  30.   val FOD_PROJECTION  = 4  //def projection(projection: Bson): FindObservable[TResult]
  31.                            //Sets a document describing the fields to return for all matching documents
  32.   val FOD_SORT        = 5  //def sort(sort: Bson): FindObservable[TResult]
  33.   val FOD_PARTIAL     = 6  //def partial(partial: Boolean): FindObservable[TResult]
  34.                            //Get partial results from a sharded cluster if one or more shards are unreachable (instead of throwing an error)
  35.   val FOD_CURSORTYPE  = 7  //def cursorType(cursorType: CursorType): FindObservable[TResult]
  36.                            //Sets the cursor type
  37.   val FOD_HINT        = 8  //def hint(hint: Bson): FindObservable[TResult]
  38.                            //Sets the hint for which index to use. A null value means no hint is set
  39.   val FOD_MAX         = 9  //def max(max: Bson): FindObservable[TResult]
  40.                            //Sets the exclusive upper bound for a specific index. A null value means no max is set
  41.   val FOD_MIN         = 10 //def min(min: Bson): FindObservable[TResult]
  42.                            //Sets the minimum inclusive lower bound for a specific index. A null value means no max is set
  43.   val FOD_RETURNKEY   = 11 //def returnKey(returnKey: Boolean): FindObservable[TResult]
  44.                            //Sets the returnKey. If true the find operation will return only the index keys in the resulting documents
  45.   val FOD_SHOWRECORDID=12  //def showRecordId(showRecordId: Boolean): FindObservable[TResult]
  46.                            //Sets the showRecordId. Set to true to add a field `\$recordId` to the returned documents
  48.   case class ResultOptions(
  49.                           optType: FOD_TYPE,
  50.                           bsonParam: Option[Bson] = None,
  51.                           valueParam: Int = 0 ){
  52.     def toProto = new sdp.grpc.services.ResultTransformer(
  53.       optType = this.optType,
  54.       bsonParam = this.bsonParam.map {b => sdp.grpc.services.MGOBson(marshal(b))},
  55.       valueParam = this.valueParam
  56.     )
  57.     def toFindObservable: FindObservable[Document] => FindObservable[Document] = find => {
  58.       optType match {
  59.         case  FOD_FIRST        => find
  60.         case  FOD_FILTER       => find.filter(bsonParam.get)
  61.         case  FOD_LIMIT        => find.limit(valueParam)
  62.         case  FOD_SKIP         => find.skip(valueParam)
  63.         case  FOD_PROJECTION   => find.projection(bsonParam.get)
  64.         case  FOD_SORT         => find.sort(bsonParam.get)
  65.         case  FOD_PARTIAL      => find.partial(valueParam != 0)
  66.         case  FOD_CURSORTYPE   => find
  67.         case  FOD_HINT         => find.hint(bsonParam.get)
  68.         case  FOD_MAX          => find.max(bsonParam.get)
  69.         case  FOD_MIN          => find.min(bsonParam.get)
  70.         case  FOD_RETURNKEY    => find.returnKey(valueParam != 0)
  71.         case  FOD_SHOWRECORDID => find.showRecordId(valueParam != 0)
  73.       }
  74.     }
  75.   }
  76.   object ResultOptions {
  77.     def fromProto(msg: sdp.grpc.services.ResultTransformer) = new ResultOptions(
  78.       optType = msg.optType,
  79.       bsonParam = msg.bsonParam.map(b => unmarshal[Bson](b.bson)),
  80.       valueParam = msg.valueParam
  81.     )
  83.   }
  86.   type MGO_COMMAND_TYPE = Int
  87.   val MGO_COMMAND_FIND            = 0
  88.   val MGO_COMMAND_COUNT           = 20
  89.   val MGO_COMMAND_DISTICT         = 21
  90.   val MGO_COMMAND_DOCUMENTSTREAM  = 1
  91.   val MGO_COMMAND_AGGREGATE       = 2
  92.   val MGO_COMMAND_INSERT          = 3
  93.   val MGO_COMMAND_DELETE          = 4
  94.   val MGO_COMMAND_REPLACE         = 5
  95.   val MGO_COMMAND_UPDATE          = 6
  98.   val MGO_ADMIN_DROPCOLLECTION    = 8
  99.   val MGO_ADMIN_CREATECOLLECTION  = 9
  100.   val MGO_ADMIN_LISTCOLLECTION    = 10
  101.   val MGO_ADMIN_CREATEVIEW        = 11
  102.   val MGO_ADMIN_CREATEINDEX       = 12
  103.   val MGO_ADMIN_DROPINDEXBYNAME   = 13
  104.   val MGO_ADMIN_DROPINDEXBYKEY    = 14
  105.   val MGO_ADMIN_DROPALLINDEXES    = 15
  108.   case class MGOAdminCtx(
  109.                            tarName: String = "",
  110.                            bsonParam: Seq[Bson] = Nil,
  111.                            options: Option[Any] = None,
  112.                            objName: String = ""
  113.      ){
  114.     def toProto = sdp.grpc.services.MGOAdminOptons(
  115.       tarName = this.tarName,
  116.       bsonParam = this.bsonParam.map {b => sdp.grpc.services.MGOBson(marshal(b))},
  117.       objName = this.objName,
  118.       options = this.options.map(b => OptionAny(marshal(b)))
  120.     )
  121.   }
  123.   object MGOAdminCtx {
  124.     def fromProto(msg: sdp.grpc.services.MGOAdminOptons) = new MGOAdminCtx(
  125.       tarName = msg.tarName,
  126.       bsonParam = msg.bsonParam.map(b => unmarshal[Bson](b.bson))
  127.     )
  128.   }
  130.   case class MGOProtoMsg(
  131.           dbName: String = "",
  132.           collName: String = "",
  133.           commandType: MGO_COMMAND_TYPE,
  134.           bsonParam: Seq[Bson] = Nil,
  135.           resultOptions: Seq[ResultOptions] = Nil,
  136.           options: ByteString = com.google.protobuf.ByteString.EMPTY,
  137.           documents: Seq[Document] = Nil,
  138.           only: Boolean = false,
  139.           adminOptions: Option[MGOAdminCtx] = None
  140.       ){
  142.     def toProto = new sdp.grpc.services.MGOOperations(
  143.       dbName = this.dbName,
  144.       collName = this.collName,
  145.       commandType = this.commandType,
  146.       bsonParam = this.bsonParam.map(bsonToProto),
  147.       resultOptions = this.resultOptions.map(_.toProto),
  148.       documents = this.documents.map(d => sdp.grpc.services.MGODocument(marshal(d))),
  149.       only = Some(this.only),
  150.       adminOptions = this.adminOptions.map(_.toProto)
  151.     )
  153.   }
  155.   object MGOProtoMsg {
  156.     def fromProto(msg: sdp.grpc.services.MGOOperations) = new MGOProtoMsg(
  157.       dbName = msg.dbName,
  158.       collName = msg.collName,
  159.       commandType = msg.commandType,
  160.       bsonParam = msg.bsonParam.map(protoToBson),
  161.       resultOptions = msg.resultOptions.map(r => ResultOptions.fromProto(r))
  162.     )
  163.   }
  165.   def bsonToProto(bson: Bson) =
  166.     MGOBson(marshal(bson.toBsonDocument(
  167.       classOf[org.mongodb.scala.bson.collection.immutable.Document],DEFAULT_CODEC_REGISTRY)))
  169.   def protoToBson(proto: MGOBson): Bson = new Bson {
  170.     val bsdoc = unmarshal[BsonDocument](proto.bson)
  171.     override def toBsonDocument[TDocument](documentClass: Class[TDocument], codecRegistry: CodecRegistry): BsonDocument = bsdoc
  172.   }
  174.   def CtxFromProto(proto: MGOOperations): MGOContext = proto.commandType match {
  175.     case MGO_COMMAND_FIND => {
  176.       var ctx = new MGOContext(
  177.         dbName = proto.dbName,
  178.         collName = proto.collName,
  179.         actionType = MGO_QUERY,
  180.         action = Some(Find())
  181.       )
  182.       def toResultTransformer(rts: Seq[ResultTransformer]): FindObservable[Document] => FindObservable[Document] = findObj =>
  183.         rts.foldRight(findObj)((a,b) => ResultOptions.fromProto(a).toFindObservable(b))
  185.       (proto.bsonParam, proto.resultOptions, proto.only) match {
  186.         case (Nil, Nil, None) => ctx
  187.         case (Nil, Nil, Some(b)) => ctx.setCommand(Find(None, None, b))
  188.         case (bp,Nil,None) => ctx.setCommand(
  189.           Find(Some(protoToBson(bp.head)),None,false))
  190.         case (bp,Nil,Some(b)) => ctx.setCommand(
  191.           Find(Some(protoToBson(bp.head)),None,b))
  192.         case (bp,fo,None) => {
  193.              ctx.setCommand(
  194.             Find(Some(protoToBson(bp.head)),Some(toResultTransformer(fo)),false))
  195.         }
  196.         case (bp,fo,Some(b)) => {
  197.           ctx.setCommand(
  198.             Find(Some(protoToBson(bp.head)),Some(toResultTransformer(fo)),b))
  199.         }
  200.         case _ => ctx
  201.       }
  202.     }
  203.   }
  205. }

BytesConverter.scala

  1. package protobuf.bytes
  2. import java.io.{ByteArrayInputStream,ByteArrayOutputStream,ObjectInputStream,ObjectOutputStream}
  3. import com.google.protobuf.ByteString
  4. object Converter {
  6.   def marshal(value: Any): ByteString = {
  7.     val stream: ByteArrayOutputStream = new ByteArrayOutputStream()
  8.     val oos = new ObjectOutputStream(stream)
  9.     oos.writeObject(value)
  10.     oos.close()
  11.     ByteString.copyFrom(stream.toByteArray())
  12.   }
  14.   def unmarshal[A](bytes: ByteString): A = {
  15.     val ois = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray))
  16.     val value = ois.readObject()
  17.     ois.close()
  18.     value.asInstanceOf[A]
  19.   }
  21. }

FindDemo.scala

  1. package demo.sdp.mgo.localapp
  2. import akka.actor.ActorSystem
  3. import akka.stream.ActorMaterializer
  4. import org.mongodb.scala._
  6. import scala.util._
  7. import scala.collection.JavaConverters._
  8. import sdp.mongo.engine._
  9. import org.mongodb.scala.model._
  10. import akka.stream.scaladsl.{Sink, Source}
  11. import org.bson.codecs.configuration.CodecRegistry
  12. import org.mongodb.scala.bson.{BsonDocument, BsonValue}
  13. import scalikejdbc._
  14. import sdp.jdbc.engine._
  15. import sdp.jdbc.config._
  16. object ProtoTests extends App {
  17.   import MGOContext._
  18.   import MGOEngine._
  19.   import MGOCommands._
  20.   import MongoActionStream._
  21.   import MgoProtoConvertion._
  22.   import org.mongodb.scala.model._
  23.   import Projections._
  24.   import Filters._
  26.   implicit val system = ActorSystem()
  27.   implicit val mat = ActorMaterializer()
  28.   implicit val ec = system.dispatcher
  30.   val clientSettings: MongoClientSettings = MongoClientSettings.builder()
  31.     .applyToClusterSettings {b =>
  32.       b.hosts(List(new ServerAddress("localhost")).asJava)
  33.     }.build()
  35.   implicit val client: MongoClient = MongoClient(clientSettings)
  38.   val eqState = equal("state","California")
  39.   val proj = exclude("rowid","_id")
  40.   val rtxfmr = Seq(
  41.     ResultOptions(
  42.     optType = FOD_LIMIT,
  43.     valueParam = 3)
  44.     ,ResultOptions(
  45.     optType = FOD_PROJECTION,
  46.     bsonParam = Some(proj))
  47.   )
  49.   val protoCtx = MGOProtoMsg(
  50.     dbName = "testdb",
  51.     collName = "aqmrpt",
  52.     commandType = MGO_COMMAND_FIND,
  53.     bsonParam = Seq(eqState),
  54.     resultOptions = rtxfmr
  55.   ).toProto
  57.   val findCtx = CtxFromProto(protoCtx)
  58.   val futFind = mgoQuery[Seq[Document]](findCtx)
  59.   futFind.onComplete {
  60.     case Success(docs) => docs.asInstanceOf[Seq[Document]].foreach{doc => println(doc.toJson())}
  61.     case Failure(e) => println(e.getMessage)
  62.   }
  64.   scala.io.StdIn.readLine()
  65.   system.terminate()
  67. }

 

 

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Scala热门文章

  • scala (3) Function 和 Method
  • flink 两个datastream实现left_join的两种方法
  • scala case class + BeanProperty+fastjson 快速实现类的序列化与反序列化
  • Scala解析MDF文件
  • flink 如何实现对watermark 的checkpoint,防止数据复写
  • Spark笔记(一):错误总结
  • 踩坑 scala 打包上传 找不到类
  • akka-typed(2) - typed-actor交流方式和交流协议
  • Scala(二)：元组、数组、映射
  • scala 命名规范

    • Scala推荐文章

  • Scala中Array和List的区别说明
  • alpakka-kafka(5)-kafka集群配置与分布式应用部署
  • alpakka-kafka(2)-consumer
  • alpakka-kafka(1)-producer
  • akka-typed(1) - actor生命周期管理
  • akka-typed(2) - typed-actor交流方式和交流协议
  • akka-typed(3) - PersistentActor has EventSourcedBehavior
  • akka-typed(4) - EventSourcedBehavior in action
  • kka-typed(5) - cluster：集群节点状态监视
  • akka-typed(6) - cluster：group router, cluster-load-balancing

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号