经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置:技术经验 » Java相关 » Scala » 查看文章
Akka-Cluster(2)- distributed pub/sub mechanism 分布式发布/订阅机制
来源:cnblogs  作者:雪川大虫  时间:2018/11/9 11:16:51  对本文有异议

   上期我们介绍了cluster singleton,它的作用是保证在一个集群环境里永远会有唯一一个singleton实例存在。具体使用方式是在集群所有节点部署ClusterSingletonManager,由集群中的leader节点选定其中一个节点并指示上面的ClusterSingletonManager运行一个cluster singleton实例。与singleton实例交互则通过即时构建ClusterSingletonProxy实例当作沟通目标。从应用场景来说cluster singleton应该是某种pull模式的应用:我们把singleton当作中央操作协调,比如说管理一个任务清单,多个ClusterSingletonProxy从任务清单中获取(pull)自己应该执行的任务。如果需要实现push模式的任务派送:即由singleton主动通知集群里某种类型的actor执行任务,那么通过ClusterSingletonProxy沟通就不适用了,使用pub/sub方式是一个可行的解决方案。

distributed pub/sub含两种发布方式:Publish/Send,分别代表群发和点对点发布模式。在集群环境里每个节点上akka-cluster系统都提供一个DistributedPubSubMediator实例作为该节点向外发布及订阅消息的渠道。发布者publisher只对每个节点发布一次消息,再由每个节点上唯一的Mediator接收并转发给节点本地所有订阅该类消息的subscriber。Publish是个类型:

  1. final case class Publish(topic: String, msg: Any, sendOneMessageToEachGroup: Boolean) extends DistributedPubSubMessage {
  2. def this(topic: String, msg: Any) = this(topic, msg, sendOneMessageToEachGroup = false)
  3. }
  4. object Publish {
  5. def apply(topic: String, msg: Any) = new Publish(topic, msg)
  6. }
  7. ...
  8. }

发布操作就是把一个Publish消息发给本节点的Mediator:

  1. class Publisher extends Actor {
  2. import DistributedPubSubMediator.Publish
  3. // activate the extension
  4. val mediator = DistributedPubSub(context.system).mediator
  5. def receive = {
  6. case in: String ?
  7. val out = in.toUpperCase
  8. mediator ! Publish("content", outsendOneMessageToEachGroup = true)
  9. }
  10. }

sendOneMessageToEachGroup默认=false,代表发布消息不会送达用group ID订阅的subscriber,true则代表消息不会送达没用group ID订阅的subscriber。同样Subscribe也是个类型:

  1. final case class Subscribe(topic: String, group: Option[String], ref: ActorRef) {
  2. require(topic != null && topic != "", "topic must be defined")
  3. /**
  4. * Convenience constructor with `group` None
  5. */
  6. def this(topic: String, ref: ActorRef) = this(topic, None, ref)
  7. /**
  8. * Java API: constructor with group: String
  9. */
  10. def this(topic: String, group: String, ref: ActorRef) = this(topic, Some(group), ref)
  11. }
  12. object Subscribe {
  13. def apply(topic: String, ref: ActorRef) = new Subscribe(topic, ref)
  14. }

订阅操作即向本地Mediator发送Subscribe消息:

  1. val mediator = DistributedPubSub(context.system).mediator
  2. // subscribe to the topic named "content"
  3. mediator ! Subscribe("content", self)
  4. def receive = {
  5. case s: String ?
  6. log.info("Got {}", s)
  7. case SubscribeAck(Subscribe("content", None, `self`)) ?
  8. log.info("subscribing")
  9. }
  10. mediator ! UnSubscribe("content", self)
  11. def receive = {
  12. ...
  13. case UnSubscribeAck ?
  14. log.info("unsubscribing")
  15. }

取消订阅则发送UnSubscribe消息。

Send/Put是一种点对点模式,不需要topic作为订阅标的。同样:Send和Put都是消息类型,Put代表订阅:

 

  1. val mediator = DistributedPubSub(context.system).mediator
  2. // register to the path
  3. mediator ! Put(self)
  4. def receive = {
  5. case s: String ?
  6. log.info("Got {}", s)
  7. }
  8. }
  9. mediator ! DistributedPubSubMediator.Remove(path)

 

Put在Mediator上登记了self,包括path,所以取消订阅也就是从Mediator上取消特定path。由于是点对点模式,Send就是针对某个path发送消息:

  1. final case class Send(path: String, msg: Any, localAffinity: Boolean) extends DistributedPubSubMessage {
  2. /**
  3. * Convenience constructor with `localAffinity` false
  4. */
  5. def this(path: String, msg: Any) = this(path, msg, localAffinity = false)
  6. }
  7. final case class SendToAll(path: String, msg: Any, allButSelf: Boolean = false) extends DistributedPubSubMessage {
  8. def this(path: String, msg: Any) = this(path, msg, allButSelf = false)
  9. }

Send通过特定的路由策略从多个在不同节点上的匹配path选定一个节点发送消息。localAffinity=true代表消息发送节点本地登记的匹配path actor优先。SendToAll则代表对所有登记了匹配path的节点发送消息:

  1. class Sender extends Actor {
  2. import DistributedPubSubMediator.Send
  3. // activate the extension
  4. val mediator = DistributedPubSub(context.system).mediator
  5. def receive = {
  6. case in: String ?
  7. val out = in.toUpperCase
  8. mediator ! Send(path = "/user/destination", msg = out, localAffinity = true)
  9. }
  10. }
  11. class SenderToAll extends Actor {
  12. import DistributedPubSubMediator.Send
  13. // activate the extension
  14. val mediator = DistributedPubSub(context.system).mediator
  15. def receive = {
  16. case in: String ?
  17. val out = in.toUpperCase
  18. mediator ! SendToAll(path = "/user/destination", msg = out)
  19. }
  20. }

下面我们还是举个例子来示范distributed pub/sub,同时示范对利用protobuf格式作为消息类型来实现发布/订阅机制。忽然想起前面介绍过的MongoDBStreaming,里面跨集群节点的数据库操作指令都是protobuf格式进行序列化的。在这个例子里我们把publisher作为一个数据库指挥,把MongoDB操作指令发布出去,然后subscriber订阅数据库操作指令。收到消息后解包成MongoDB操作指令,然后对数据库操作。

我们首先看看在application.conf里是如何配置消息序列化格式的:

  1. actor {
  2. provider = "cluster"
  3. serializers {
  4. java = "akka.serialization.JavaSerializer"
  5. proto = "akka.remote.serialization.ProtobufSerializer"
  6. }
  7. serialization-bindings {
  8. "java.lang.String" = java
  9. "scalapb.GeneratedMessage" = proto
  10. }
  11. }

例子里的protobuf消息是由scalapb从.proto文件中自动产生的。下面我们先定义subscriber:

  1. trait PubMessage {}
  2. case class Gossip(msg: String) extends PubMessage
  3. case object StopTalk
  4. class Subscriber extends Actor with ActorLogging {
  5. import monix.execution.Scheduler.Implicits.global
  6. implicit val mgosys = context.system
  7. implicit val ec = mgosys.dispatcher
  8. val clientSettings: MongoClientSettings = MongoClientSettings.builder()
  9. .applyToClusterSettings {b =>
  10. b.hosts(List(new ServerAddress("localhost:27017")).asJava)
  11. }.build()
  12. implicit val client: MongoClient = MongoClient(clientSettings)
  13. val mediator = DistributedPubSub(context.system).mediator
  14. override def preStart() = {
  15. mediator ! Subscribe("talks", self)
  16. mediator ! Subscribe("mongodb", self)
  17. super.preStart()
  18. }
  19. override def receive: Receive = {
  20. case Gossip(msg) =>
  21. log.info(s"******* received message: $msg by ${self}")
  22. case SubscribeAck(sub) =>
  23. log.info(s"******* $self Subscribed to ${sub.topic} ...")
  24. case UnsubscribeAck(sub) =>
  25. log.info(s"******* $self Unsubscribed from ${sub.topic} ...")
  26. case StopTalk =>
  27. mediator ! Unsubscribe("talks", self)
  28. mediator ! Unsubscribe("mongodb", self)
  29. case someProto @ Some(proto:ProtoMGOContext) =>
  30. val ctx = MGOContext.fromProto(proto)
  31. log.info(s"****** received MGOContext: $someProto *********")
  32. val task = mgoUpdate[Completed](ctx).toTask
  33. task.runOnComplete {
  34. case Success(s) => println("operations completed successfully.")
  35. case Failure(exception) => println(s"error: ${exception.getMessage}")
  36. }
  37. case msg => log.info(s"**********received some messaged: $msg *********")
  38. }
  39. }
  40. object Subscriber {
  41. def props = Props(new Subscriber)
  42. def create(port: Int): ActorRef = {
  43. val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=$port")
  44. .withFallback(ConfigFactory.load())
  45. val system = ActorSystem("PubSubSystem", config)
  46. system.actorOf(props, s"subscriber$port")
  47. }
  48. }

因为subscriber需要执行MongoDB指令,所有必须定义一个客户端:

  1. val clientSettings: MongoClientSettings = MongoClientSettings.builder()
  2. .applyToClusterSettings {b =>
  3. b.hosts(List(new ServerAddress("localhost:27017")).asJava)
  4. }.build()
  5. implicit val client: MongoClient = MongoClient(clientSettings)
  6. ...
  7. case someProto @ Some(proto:ProtoMGOContext) =>
  8. val ctx = MGOContext.fromProto(proto)
  9. log.info(s"****** received MGOContext: $someProto *********")
  10. val task = mgoUpdate[Completed](ctx).toTask
  11. task.runOnComplete {
  12. case Success(s) => println("operations completed successfully.")
  13. case Failure(exception) => println(s"error: ${exception.getMessage}")
  14. }

分别订阅两种消息: 

 

  1. override def preStart() = {
  2. mediator ! Subscribe("talks", self)
  3. mediator ! Subscribe("mongodb", self)
  4. super.preStart()
  5. }
  6. ...
  7. case StopTalk =>
  8. mediator ! Unsubscribe("talks", self)
  9. mediator ! Unsubscribe("mongodb", self)

 

publisher是这样定义的:

  1. class Publisher extends Actor with ActorLogging {
  2. val mediator = DistributedPubSub(context.system).mediator
  3. val ctx = MGOContext("testdb","friends")
  4. override def receive: Receive = {
  5. case Gossip(msg) =>
  6. mediator ! Publish("talks", Gossip(msg))
  7. log.info(s"published message: $msg")
  8. case StopTalk =>
  9. mediator ! Publish("talks", StopTalk)
  10. log.info("everyone stop!")
  11. case doc @ Document(_) =>
  12. val c = ctx.setCommand(MGOCommands.Insert(Seq(doc)))
  13. log.info(s"*****publishing mongo command: ${c}")
  14. mediator ! Publish("mongodb",c.toSomeProto)
  15. }
  16. }
  17. object Publisher {
  18. def props = Props(new Publisher)
  19. def create(port: Int): ActorRef = {
  20. val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=${port}")
  21. .withFallback(ConfigFactory.load())
  22. val system = ActorSystem("PubSubSystem", config)
  23. system.actorOf(props, "publisher")
  24. }
  25. }

publisher构造指令:将一个Document当作一条记录插入到MongoDB friends表里。指令被转换成protobuf格式:

  1. val ctx = MGOContext("testdb","friends")
  2. override def receive: Receive = {
  3. ...
  4. case doc @ Document(_) =>
  5. val c = ctx.setCommand(MGOCommands.Insert(Seq(doc)))
  6. log.info(s"*****publishing mongo command: ${c}")
  7. mediator ! Publish("mongodb",c.toSomeProto)
  8. }

下面是publisher和subscriber应用示范:

  1. package pubsubdemo
  2. import org.mongodb.scala._
  3. object PubSubDemo extends App {
  4. val publisher = Publisher.create(2551) //seed node
  5. scala.io.StdIn.readLine()
  6. Subscriber.create(2552)
  7. scala.io.StdIn.readLine()
  8. Subscriber.create(2553)
  9. scala.io.StdIn.readLine()
  10. publisher ! Gossip("hello everyone!")
  11. scala.io.StdIn.readLine()
  12. publisher ! Gossip("do you hear me ?")
  13. scala.io.StdIn.readLine()
  14. //MongoDB 操作示范
  15. val peter = Document("name" -> "MAGRET KOON", "age" -> 28)
  16. publisher ! peter
  17. scala.io.StdIn.readLine()
  18. publisher ! StopTalk
  19. scala.io.StdIn.readLine()
  20. }

值得注意的是:系统构建了两个subscriber, 2552和2553,意味MongoDB操作指令会被重复执行两次。不过我们的示范不在意这些细节。

下面是这次讨论中的示范源代码:

project/scalapb.sbt

  1. addSbtPlugin("com.thesamet" % "sbt-protoc" % "0.99.18")
  2. libraryDependencies ++= Seq(
  3. "com.thesamet.scalapb" %% "compilerplugin" % "0.7.4"
  4. )

build.sbt

  1. import scalapb.compiler.Version.scalapbVersion
  2. import scalapb.compiler.Version.grpcJavaVersion
  3. name := "distributed-pub-sub"
  4. version := "0.1"
  5. scalaVersion := "2.12.7"
  6. scalacOptions += "-Ypartial-unification"
  7. libraryDependencies := Seq(
  8. "com.typesafe.akka" %% "akka-actor" % "2.5.17",
  9. "com.typesafe.akka" %% "akka-cluster-tools" % "2.5.17",
  10. "com.thesamet.scalapb" %% "scalapb-runtime" % scalapbVersion % "protobuf",
  11. // "io.grpc" % "grpc-netty" % grpcJavaVersion,
  12. "com.thesamet.scalapb" %% "scalapb-runtime-grpc" % scalapbVersion,
  13. "io.monix" %% "monix" % "2.3.0",
  14. //for mongodb 4.0
  15. "org.mongodb.scala" %% "mongo-scala-driver" % "2.4.0",
  16. "com.lightbend.akka" %% "akka-stream-alpakka-mongodb" % "0.20",
  17. //other dependencies
  18. "co.fs2" %% "fs2-core" % "0.9.7",
  19. "ch.qos.logback" % "logback-classic" % "1.2.3",
  20. "org.typelevel" %% "cats-core" % "0.9.0",
  21. "io.monix" %% "monix-execution" % "3.0.0-RC1",
  22. "io.monix" %% "monix-eval" % "3.0.0-RC1"
  23. )
  24. PB.targets in Compile := Seq(
  25. scalapb.gen() -> (sourceManaged in Compile).value
  26. )

protobuf/sdp.proto

  1. syntax = "proto3";
  2. import "google/protobuf/wrappers.proto";
  3. import "google/protobuf/any.proto";
  4. import "scalapb/scalapb.proto";
  5. option (scalapb.options) = {
  6. // use a custom Scala package name
  7. // package_name: "io.ontherocks.introgrpc.demo"
  8. // don't append file name to package
  9. flat_package: true
  10. // generate one Scala file for all messages (services still get their own file)
  11. single_file: true
  12. // add imports to generated file
  13. // useful when extending traits or using custom types
  14. // import: "io.ontherocks.hellogrpc.RockingMessage"
  15. // code to put at the top of generated file
  16. // works only with `single_file: true`
  17. //preamble: "sealed trait SomeSealedTrait"
  18. };
  19. package sdp.grpc.services;
  20. message ProtoDate {
  21. int32 yyyy = 1;
  22. int32 mm = 2;
  23. int32 dd = 3;
  24. }
  25. message ProtoTime {
  26. int32 hh = 1;
  27. int32 mm = 2;
  28. int32 ss = 3;
  29. int32 nnn = 4;
  30. }
  31. message ProtoDateTime {
  32. ProtoDate date = 1;
  33. ProtoTime time = 2;
  34. }
  35. message ProtoAny {
  36. bytes value = 1;
  37. }

protobuf/mgo.proto

  1. syntax = "proto3";
  2. import "google/protobuf/wrappers.proto";
  3. import "google/protobuf/any.proto";
  4. import "scalapb/scalapb.proto";
  5. option (scalapb.options) = {
  6. // use a custom Scala package name
  7. // package_name: "io.ontherocks.introgrpc.demo"
  8. // don't append file name to package
  9. flat_package: true
  10.  
  11. // generate one Scala file for all messages (services still get their own file)
  12. single_file: true
  13.  
  14. // add imports to generated file
  15. // useful when extending traits or using custom types
  16. // import: "io.ontherocks.hellogrpc.RockingMessage"
  17. // code to put at the top of generated file
  18. // works only with `single_file: true`
  19. //preamble: "sealed trait SomeSealedTrait"
  20. };
  21. /*
  22. * Demoes various customization options provided by ScalaPBs.
  23. */
  24. package sdp.grpc.services;
  25. import "sdp.proto";
  26. message ProtoMGOBson {
  27. bytes bson = 1;
  28. }
  29. message ProtoMGODocument {
  30. bytes document = 1;
  31. }
  32. message ProtoMGOResultOption { //FindObservable
  33. int32 optType = 1;
  34. ProtoMGOBson bsonParam = 2;
  35. int32 valueParam = 3;
  36. }
  37. message ProtoMGOAdmin{
  38. string tarName = 1;
  39. repeated ProtoMGOBson bsonParam = 2;
  40. ProtoAny options = 3;
  41. string objName = 4;
  42. }
  43. message ProtoMGOContext { //MGOContext
  44. string dbName = 1;
  45. string collName = 2;
  46. int32 commandType = 3;
  47. repeated ProtoMGOBson bsonParam = 4;
  48. repeated ProtoMGOResultOption resultOptions = 5;
  49. repeated string targets = 6;
  50. ProtoAny options = 7;
  51. repeated ProtoMGODocument documents = 8;
  52. google.protobuf.BoolValue only = 9;
  53. ProtoMGOAdmin adminOptions = 10;
  54. }

PubSubActor.scala

  1. package pubsubdemo
  2. import akka.actor._
  3. import akka.cluster.pubsub.DistributedPubSubMediator._
  4. import akka.cluster.pubsub._
  5. import com.typesafe.config._
  6. import akka.actor.ActorSystem
  7. import org.mongodb.scala._
  8. import sdp.grpc.services.ProtoMGOContext
  9. import sdp.mongo.engine.MGOClasses._
  10. import sdp.mongo.engine.MGOEngine._
  11. import sdp.result.DBOResult._
  12. import scala.collection.JavaConverters._
  13. import scala.util._
  14. trait PubMessage {}
  15. case class Gossip(msg: String) extends PubMessage
  16. case object StopTalk
  17. class Subscriber extends Actor with ActorLogging {
  18. import monix.execution.Scheduler.Implicits.global
  19. implicit val mgosys = context.system
  20. implicit val ec = mgosys.dispatcher
  21. val clientSettings: MongoClientSettings = MongoClientSettings.builder()
  22. .applyToClusterSettings {b =>
  23. b.hosts(List(new ServerAddress("localhost:27017")).asJava)
  24. }.build()
  25. implicit val client: MongoClient = MongoClient(clientSettings)
  26. val mediator = DistributedPubSub(context.system).mediator
  27. override def preStart() = {
  28. mediator ! Subscribe("talks", self)
  29. mediator ! Subscribe("mongodb", self)
  30. super.preStart()
  31. }
  32. override def receive: Receive = {
  33. case Gossip(msg) =>
  34. log.info(s"******* received message: $msg by ${self}")
  35. case SubscribeAck(sub) =>
  36. log.info(s"******* $self Subscribed to ${sub.topic} ...")
  37. case UnsubscribeAck(sub) =>
  38. log.info(s"******* $self Unsubscribed from ${sub.topic} ...")
  39. case StopTalk =>
  40. mediator ! Unsubscribe("talks", self)
  41. mediator ! Unsubscribe("mongodb", self)
  42. case someProto @ Some(proto:ProtoMGOContext) =>
  43. val ctx = MGOContext.fromProto(proto)
  44. log.info(s"****** received MGOContext: $someProto *********")
  45. val task = mgoUpdate[Completed](ctx).toTask
  46. task.runOnComplete {
  47. case Success(s) => println("operations completed successfully.")
  48. case Failure(exception) => println(s"error: ${exception.getMessage}")
  49. }
  50. case msg => log.info(s"**********received some messaged: $msg *********")
  51. }
  52. }
  53. object Subscriber {
  54. def props = Props(new Subscriber)
  55. def create(port: Int): ActorRef = {
  56. val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=$port")
  57. .withFallback(ConfigFactory.load())
  58. val system = ActorSystem("PubSubSystem", config)
  59. system.actorOf(props, s"subscriber$port")
  60. }
  61. }
  62. class Publisher extends Actor with ActorLogging {
  63. val mediator = DistributedPubSub(context.system).mediator
  64. val ctx = MGOContext("testdb","friends")
  65. override def receive: Receive = {
  66. case Gossip(msg) =>
  67. mediator ! Publish("talks", Gossip(msg))
  68. log.info(s"published message: $msg")
  69. case StopTalk =>
  70. mediator ! Publish("talks", StopTalk)
  71. log.info("everyone stop!")
  72. case doc @ Document(_) =>
  73. val c = ctx.setCommand(MGOCommands.Insert(Seq(doc)))
  74. log.info(s"*****publishing mongo command: ${c}")
  75. mediator ! Publish("mongodb",c.toSomeProto)
  76. }
  77. }
  78. object Publisher {
  79. def props = Props(new Publisher)
  80. def create(port: Int): ActorRef = {
  81. val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=${port}")
  82. .withFallback(ConfigFactory.load())
  83. val system = ActorSystem("PubSubSystem", config)
  84. system.actorOf(props, "publisher")
  85. }
  86. }

PubSubDemo.scala

  1. package pubsubdemo
  2. import org.mongodb.scala._
  3. object PubSubDemo extends App {
  4. val publisher = Publisher.create(2551) //seed node
  5. scala.io.StdIn.readLine()
  6. Subscriber.create(2552)
  7. scala.io.StdIn.readLine()
  8. Subscriber.create(2553)
  9. scala.io.StdIn.readLine()
  10. publisher ! Gossip("hello everyone!")
  11. scala.io.StdIn.readLine()
  12. publisher ! Gossip("do you hear me ?")
  13. scala.io.StdIn.readLine()
  14. //MongoDB 操作示范
  15. val peter = Document("name" -> "MAGRET KOON", "age" -> 28)
  16. publisher ! peter
  17. scala.io.StdIn.readLine()
  18. publisher ! StopTalk
  19. scala.io.StdIn.readLine()
  20. }

MongoDBEngine.scala

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

MGOProtoConversion.scala

package sdp.mongo.engine
import org.mongodb.scala.bson.collection.immutable.Document
import org.bson.conversions.Bson
import sdp.grpc.services._
import protobuf.bytes.Converter._
import MGOClasses._
import MGOAdmins._
import MGOCommands._
import org.bson.BsonDocument
import org.bson.codecs.configuration.CodecRegistry
import org.mongodb.scala.bson.codecs.DEFAULT_CODEC_REGISTRY
import org.mongodb.scala.FindObservable

object MGOProtoConversion {

  type MGO_COMMAND_TYPE = Int
  val MGO_COMMAND_FIND            = 0
  val MGO_COMMAND_COUNT           = 20
  val MGO_COMMAND_DISTICT         = 21
  val MGO_COMMAND_DOCUMENTSTREAM  = 1
  val MGO_COMMAND_AGGREGATE       = 2
  val MGO_COMMAND_INSERT          = 3
  val MGO_COMMAND_DELETE          = 4
  val MGO_COMMAND_REPLACE         = 5
  val MGO_COMMAND_UPDATE          = 6


  val MGO_ADMIN_DROPCOLLECTION    = 8
  val MGO_ADMIN_CREATECOLLECTION  = 9
  val MGO_ADMIN_LISTCOLLECTION    = 10
  val MGO_ADMIN_CREATEVIEW        = 11
  val MGO_ADMIN_CREATEINDEX       = 12
  val MGO_ADMIN_DROPINDEXBYNAME   = 13
  val MGO_ADMIN_DROPINDEXBYKEY    = 14
  val MGO_ADMIN_DROPALLINDEXES    = 15


  case class AdminContext(
                           tarName: String = "",
                           bsonParam: Seq[Bson] = Nil,
                           options: Option[Any] = None,
                           objName: String = ""
                         ){
    def toProto = sdp.grpc.services.ProtoMGOAdmin(
      tarName = this.tarName,
      bsonParam = this.bsonParam.map {b => sdp.grpc.services.ProtoMGOBson(marshal(b))},
      objName = this.objName,
      options = this.options.map(b => ProtoAny(marshal(b)))

    )
  }

  object AdminContext {
    def fromProto(msg: sdp.grpc.services.ProtoMGOAdmin) = new AdminContext(
      tarName = msg.tarName,
      bsonParam = msg.bsonParam.map(b => unmarshal[Bson](b.bson)),
      objName = msg.objName,
      options = msg.options.map(b => unmarshal[Any](b.value))
    )
  }

  case class Context(
                      dbName: String = "",
                      collName: String = "",
                      commandType: MGO_COMMAND_TYPE,
                      bsonParam: Seq[Bson] = Nil,
                      resultOptions: Seq[ResultOptions] = Nil,
                      options: Option[Any] = None,
                      documents: Seq[Document] = Nil,
                      targets: Seq[String] = Nil,
                      only: Boolean = false,
                      adminOptions: Option[AdminContext] = None
                    ){

    def toProto = new sdp.grpc.services.ProtoMGOContext(
      dbName = this.dbName,
      collName = this.collName,
      commandType = this.commandType,
      bsonParam = this.bsonParam.map(bsonToProto),
      resultOptions = this.resultOptions.map(_.toProto),
      options = { if(this.options == None)
        None //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
      else
        Some(ProtoAny(marshal(this.options.get))) },
      documents = this.documents.map(d => sdp.grpc.services.ProtoMGODocument(marshal(d))),
      targets = this.targets,
      only = Some(this.only),
      adminOptions = this.adminOptions.map(_.toProto)
    )

  }

  object MGODocument {
    def fromProto(msg: sdp.grpc.services.ProtoMGODocument): Document =
      unmarshal[Document](msg.document)
    def toProto(doc: Document): sdp.grpc.services.ProtoMGODocument =
      new ProtoMGODocument(marshal(doc))
  }

  object MGOProtoMsg {
    def fromProto(msg: sdp.grpc.services.ProtoMGOContext) = new Context(
      dbName = msg.dbName,
      collName = msg.collName,
      commandType = msg.commandType,
      bsonParam = msg.bsonParam.map(protoToBson),
      resultOptions = msg.resultOptions.map(r => ResultOptions.fromProto(r)),
      options = msg.options.map(a => unmarshal[Any](a.value)),
      documents = msg.documents.map(doc => unmarshal[Document](doc.document)),
      targets = msg.targets,
      adminOptions = msg.adminOptions.map(ado => AdminContext.fromProto(ado))
    )
  }

  def bsonToProto(bson: Bson) =
    ProtoMGOBson(marshal(bson.toBsonDocument(
      classOf[org.mongodb.scala.bson.collection.immutable.Document],DEFAULT_CODEC_REGISTRY)))

  def protoToBson(proto: ProtoMGOBson): Bson = new Bson {
    val bsdoc = unmarshal[BsonDocument](proto.bson)
    override def toBsonDocument[TDocument](documentClass: Class[TDocument], codecRegistry: CodecRegistry): BsonDocument = bsdoc
  }

  def ctxFromProto(proto: ProtoMGOContext): MGOContext = proto.commandType match {
    case MGO_COMMAND_FIND => {
      var ctx = new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_QUERY,
        action = Some(Find())
      )
      def toResultOption(rts: Seq[ProtoMGOResultOption]): FindObservable[Document] => FindObservable[Document] = findObj =>
        rts.foldRight(findObj)((a,b) => ResultOptions.fromProto(a).toFindObservable(b))

      (proto.bsonParam, proto.resultOptions, proto.only) match {
        case (Nil, Nil, None) => ctx
        case (Nil, Nil, Some(b)) => ctx.setCommand(Find(firstOnly = b))
        case (bp,Nil,None) => ctx.setCommand(
          Find(filter = Some(protoToBson(bp.head))))
        case (bp,Nil,Some(b)) => ctx.setCommand(
          Find(filter = Some(protoToBson(bp.head)), firstOnly = b))
        case (bp,fo,None) => {
          ctx.setCommand(
            Find(filter = Some(protoToBson(bp.head)),
              andThen = fo.map(ResultOptions.fromProto)
            ))
        }
        case (bp,fo,Some(b)) => {
          ctx.setCommand(
            Find(filter = Some(protoToBson(bp.head)),
              andThen = fo.map(ResultOptions.fromProto),
              firstOnly = b))
        }
        case _ => ctx
      }
    }
    case MGO_COMMAND_COUNT => {
      var ctx = new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_QUERY,
        action = Some(Count())
      )
      (proto.bsonParam, proto.options) match {
        case (Nil, None) => ctx
        case (bp, None) => ctx.setCommand(
          Count(filter = Some(protoToBson(bp.head)))
        )
        case (Nil,Some(o)) => ctx.setCommand(
          Count(options = Some(unmarshal[Any](o.value)))
        )
        case _ => ctx
      }
    }
    case MGO_COMMAND_DISTICT => {
      var ctx = new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_QUERY,
        action = Some(Distict(fieldName = proto.targets.head))
      )
      (proto.bsonParam) match {
        case Nil => ctx
        case bp: Seq[ProtoMGOBson] => ctx.setCommand(
          Distict(fieldName = proto.targets.head,filter = Some(protoToBson(bp.head)))
        )
        case _ => ctx
      }
    }
    case MGO_COMMAND_AGGREGATE => {
      new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_QUERY,
        action = Some(Aggregate(proto.bsonParam.map(p => protoToBson(p))))
      )
    }
    case MGO_ADMIN_LISTCOLLECTION => {
      new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_QUERY,
        action = Some(ListCollection(proto.dbName)))
    }
    case MGO_COMMAND_INSERT => {
      var ctx = new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_UPDATE,
        action = Some(Insert(
          newdocs = proto.documents.map(doc => unmarshal[Document](doc.document))))
      )
      proto.options match {
        case None => ctx
        case Some(o) => ctx.setCommand(Insert(
          newdocs = proto.documents.map(doc => unmarshal[Document](doc.document)),
          options = Some(unmarshal[Any](o.value)))
        )
      }
    }
    case MGO_COMMAND_DELETE => {
      var ctx = new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_UPDATE,
        action = Some(Delete(
          filter = protoToBson(proto.bsonParam.head)))
      )
      (proto.options, proto.only) match {
        case (None,None) => ctx
        case (None,Some(b)) => ctx.setCommand(Delete(
          filter = protoToBson(proto.bsonParam.head),
          onlyOne = b))
        case (Some(o),None) => ctx.setCommand(Delete(
          filter = protoToBson(proto.bsonParam.head),
          options = Some(unmarshal[Any](o.value)))
        )
        case (Some(o),Some(b)) => ctx.setCommand(Delete(
          filter = protoToBson(proto.bsonParam.head),
          options = Some(unmarshal[Any](o.value)),
          onlyOne = b)
        )
      }
    }
    case MGO_COMMAND_REPLACE => {
      var ctx = new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_UPDATE,
        action = Some(Replace(
          filter = protoToBson(proto.bsonParam.head),
          replacement = unmarshal[Document](proto.documents.head.document)))
      )
      proto.options match {
        case None => ctx
        case Some(o) => ctx.setCommand(Replace(
          filter = protoToBson(proto.bsonParam.head),
          replacement = unmarshal[Document](proto.documents.head.document),
          options = Some(unmarshal[Any](o.value)))
        )
      }
    }
    case MGO_COMMAND_UPDATE => {
      var ctx = new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_UPDATE,
        action = Some(Update(
          filter = protoToBson(proto.bsonParam.head),
          update = protoToBson(proto.bsonParam.tail.head)))
      )
      (proto.options, proto.only) match {
        case (None,None) => ctx
        case (None,Some(b)) => ctx.setCommand(Update(
          filter = protoToBson(proto.bsonParam.head),
          update = protoToBson(proto.bsonParam.tail.head),
          onlyOne = b))
        case (Some(o),None) => ctx.setCommand(Update(
          filter = protoToBson(proto.bsonParam.head),
          update = protoToBson(proto.bsonParam.tail.head),
          options = Some(unmarshal[Any](o.value)))
        )
        case (Some(o),Some(b)) => ctx.setCommand(Update(
          filter = protoToBson(proto.bsonParam.head),
          update = protoToBson(proto.bsonParam.tail.head),
          options = Some(unmarshal[Any](o.value)),
          onlyOne = b)
        )
      }
    }
    case MGO_ADMIN_DROPCOLLECTION =>
      new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_ADMIN,
        action = Some(DropCollection(proto.collName))
      )
    case MGO_ADMIN_CREATECOLLECTION => {
      var ctx = new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_ADMIN,
        action = Some(CreateCollection(proto.collName))
      )
      proto.options match {
        case None => ctx
        case Some(o) => ctx.setCommand(CreateCollection(proto.collName,
          options = Some(unmarshal[Any](o.value)))
        )
      }
    }
    case MGO_ADMIN_CREATEVIEW => {
      var ctx = new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_ADMIN,
        action = Some(CreateView(viewName = proto.targets.head,
          viewOn = proto.targets.tail.head,
          pipeline = proto.bsonParam.map(p => protoToBson(p))))
      )
      proto.options match {
        case None => ctx
        case Some(o) => ctx.setCommand(CreateView(viewName = proto.targets.head,
          viewOn = proto.targets.tail.head,
          pipeline = proto.bsonParam.map(p => protoToBson(p)),
          options = Some(unmarshal[Any](o.value)))
        )
      }
    }
    case MGO_ADMIN_CREATEINDEX=> {
      var ctx = new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_ADMIN,
        action = Some(CreateIndex(key = protoToBson(proto.bsonParam.head)))
      )
      proto.options match {
        case None => ctx
        case Some(o) => ctx.setCommand(CreateIndex(key = protoToBson(proto.bsonParam.head),
          options = Some(unmarshal[Any](o.value)))
        )
      }
    }
    case MGO_ADMIN_DROPINDEXBYNAME=> {
      var ctx = new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_ADMIN,
        action = Some(DropIndexByName(indexName = proto.targets.head))
      )
      proto.options match {
        case None => ctx
        case Some(o) => ctx.setCommand(DropIndexByName(indexName = proto.targets.head,
          options = Some(unmarshal[Any](o.value)))
        )
      }
    }
    case MGO_ADMIN_DROPINDEXBYKEY=> {
      var ctx = new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_ADMIN,
        action = Some(DropIndexByKey(key = protoToBson(proto.bsonParam.head)))
      )
      proto.options match {
        case None => ctx
        case Some(o) => ctx.setCommand(DropIndexByKey(key = protoToBson(proto.bsonParam.head),
          options = Some(unmarshal[Any](o.value)))
        )
      }
    }
    case MGO_ADMIN_DROPALLINDEXES=> {
      var ctx = new MGOContext(
        dbName = proto.dbName,
        collName = proto.collName,
        actionType = MGO_ADMIN,
        action = Some(DropAllIndexes())
      )
      proto.options match {
        case None => ctx
        case Some(o) => ctx.setCommand(DropAllIndexes(
          options = Some(unmarshal[Any](o.value)))
        )
      }
    }

  }

  def ctxToProto(ctx: MGOContext): Option[sdp.grpc.services.ProtoMGOContext] = ctx.action match {
    case None => None
    case Some(act) => act match {
      case Count(filter, options) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          collName = ctx.collName,
          commandType = MGO_COMMAND_COUNT,
          bsonParam = { if (filter == None) Seq.empty[ProtoMGOBson]
                        else Seq(bsonToProto(filter.get))},
          options = { if(options == None) None  //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
                      else Some(ProtoAny(marshal(options.get))) }
      ))
      case Distict(fieldName, filter) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          collName = ctx.collName,
          commandType = MGO_COMMAND_DISTICT,
          bsonParam = { if (filter == None) Seq.empty[ProtoMGOBson]
                        else Seq(bsonToProto(filter.get))},
          targets = Seq(fieldName)

        ))

      case Find(filter, andThen, firstOnly) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          collName = ctx.collName,
          commandType = MGO_COMMAND_FIND,
          bsonParam = { if (filter == None) Seq.empty[ProtoMGOBson]
          else Seq(bsonToProto(filter.get))},
          resultOptions = andThen.map(_.toProto)
        ))

      case Aggregate(pipeLine) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          collName = ctx.collName,
          commandType = MGO_COMMAND_AGGREGATE,
          bsonParam = pipeLine.map(bsonToProto)
        ))

      case Insert(newdocs, options) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          collName = ctx.collName,
          commandType = MGO_COMMAND_INSERT,
          documents = newdocs.map(d => ProtoMGODocument(marshal(d))),
          options = { if(options == None) None      //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
          else Some(ProtoAny(marshal(options.get))) }
        ))

      case Delete(filter, options, onlyOne) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          collName = ctx.collName,
          commandType = MGO_COMMAND_DELETE,
          bsonParam = Seq(bsonToProto(filter)),
          options = { if(options == None) None  //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
          else Some(ProtoAny(marshal(options.get))) },
          only = Some(onlyOne)
        ))

      case Replace(filter, replacement, options) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          collName = ctx.collName,
          commandType = MGO_COMMAND_REPLACE,
          bsonParam = Seq(bsonToProto(filter)),
          options = { if(options == None) None //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
          else Some(ProtoAny(marshal(options.get))) },
          documents = Seq(ProtoMGODocument(marshal(replacement)))
        ))

      case Update(filter, update, options, onlyOne) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          collName = ctx.collName,
          commandType = MGO_COMMAND_UPDATE,
          bsonParam = Seq(bsonToProto(filter),bsonToProto(update)),
          options = { if(options == None) None //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
          else Some(ProtoAny(marshal(options.get))) },
          only = Some(onlyOne)
        ))


      case DropCollection(coll) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          collName = coll,
          commandType = MGO_ADMIN_DROPCOLLECTION
        ))

      case CreateCollection(coll, options) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          collName = coll,
          commandType = MGO_ADMIN_CREATECOLLECTION,
          options = { if(options == None) None  //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
          else Some(ProtoAny(marshal(options.get))) }
        ))

      case ListCollection(dbName) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          commandType = MGO_ADMIN_LISTCOLLECTION
        ))

      case CreateView(viewName, viewOn, pipeline, options) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          collName = ctx.collName,
          commandType = MGO_ADMIN_CREATEVIEW,
          bsonParam = pipeline.map(bsonToProto),
          options = { if(options == None) None  //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
          else Some(ProtoAny(marshal(options.get))) },
          targets = Seq(viewName,viewOn)
        ))

      case CreateIndex(key, options) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          collName = ctx.collName,
          commandType = MGO_ADMIN_CREATEINDEX,
          bsonParam = Seq(bsonToProto(key)),
          options = { if(options == None) None //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
          else Some(ProtoAny(marshal(options.get))) }
        ))


      case DropIndexByName(indexName, options) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          collName = ctx.collName,
          commandType = MGO_ADMIN_DROPINDEXBYNAME,
          targets = Seq(indexName),
          options = { if(options == None) None //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
          else Some(ProtoAny(marshal(options.get))) }
        ))

      case DropIndexByKey(key, options) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          collName = ctx.collName,
          commandType = MGO_ADMIN_DROPINDEXBYKEY,
          bsonParam = Seq(bsonToProto(key)),
          options = { if(options == None) None //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
          else Some(ProtoAny(marshal(options.get))) }
        ))


      case DropAllIndexes(options) =>
        Some(new sdp.grpc.services.ProtoMGOContext(
          dbName = ctx.dbName,
          collName = ctx.collName,
          commandType = MGO_ADMIN_DROPALLINDEXES,
          options = { if(options == None) None //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
          else Some(ProtoAny(marshal(options.get))) }
        ))

    }
  }

}

 

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

本站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号