经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Scala » 查看文章
Akka-Cluster（3）- ClusterClient, 集群客户端
来源：cnblogs　　作者：雪川大虫　　时间：2018/12/11 9:26:00　　对本文有异议

  上篇我们介绍了distributed pub/sub消息传递机制。这是在同一个集群内的消息共享机制：发布者（publisher）和订阅者（subscriber）都在同一个集群的节点上，所有节点上的DistributedPubSubMediator通过集群内部的沟通机制在底层构建了消息流通渠道。在actor pub/sub层面可以实现对象位置透明化。在现实里很多前端都会作为某个集群的客户端但又与集群分离，又或者两个独立的集群之间可能会发生交互关系，这是也会出现客户端与服务端不在同一集群内的情况，ClusterClient就是为集群外部actor与集群内部actor进行沟通的解决方案。

实际上ClusterClient模式就代表一种依赖于消息发布订阅机制的服务方式：客户端通过消息来请求服务，服务端接收请求服务消息并提供相应运算服务。

我们可以把集群客户端模式分成集群客户端ClusterClient和集群服务端ClusterClientReceptionist，从字面理解这就是个接待员这么个角色，负责接待集群外客户端发起的服务请求。在集群所有节点上（或者选定角色role）都部署ClusterClientReceptionist,它们都与本节点的DistributedPubSubMediator对接组成更上一层的消息订阅方，ClusterClient与ClusterClientReceptionist的对接又组成了一种统一集群环境可以实现上集所讨论的distributed pub/sub机制。

ClusterClient就是消息发布方，它是在目标集群之外机器上的某个actor。这个机器上的actor如果需要向集群内部actor发送消息可以通过这个机器上的ClusterClient actor与集群内的ClusterClientReceptionist搭建的通道向集群内某个ClusterClientReceptionist连接的DistributedPubSubMediator所登记的actor进行消息发送。所以使用集群客户端的机器必须在本机启动ClusterClient服务（运行这个actor），这是通讯桥梁的一端。

ClusterClient在启动时用预先配置的地址（contact points）与ClusterClientReceptionist连接，然后通过ClusterClientReceptionist发布的联络点清单来维护内部的对接点清单，可以进行持久化，在发生系统重启时用这个名单来与集群连接。一旦连接，ClusterClient会监控对方运行情况，自动进行具体ClusterClientReceiptionist的替换。ClusterClient发布消息是包嵌在三种结构里的：

1、ClusterClient.Send

2、ClusterClient.SendAll

3、ClusterClient.Publish

这几种方法我们在上篇已经讨论过，这里就略去。

ClusterClientReceiptionist是集群内的消息接收接口。集群内需要接收消息的actor必须在本地的DistributedPubSubMediator上注册自己的地址，ClusterClientReceptionist由此获得集群内所有服务项目actor的地址清单。通过ClusterClient发布的消息内指定接收方类型信息来确定最终接收消息并提供服务的actor。服务注册示范如下：

  1. //注册服务A
  2.   val serviceA = system.actorOf(Props[Service], "serviceA")
  3.   ClusterClientReceptionist(system).registerService(serviceA)
  4. //注册服务B
  5.   val serviceB = system.actorOf(Props[Service], "serviceB")
  6.   ClusterClientReceptionist(system).registerService(serviceB)

ClusterClient调用服务示范：

  1.   val client = system.actorOf(ClusterClient.props(
  2.   ClusterClientSettings(system).withInitialContacts(initialContacts)), "client")
  3.   client ! ClusterClient.Send("/user/serviceA", DoThis, localAffinity = true)
  4.   client ! ClusterClient.SendToAll("/user/serviceB", DoThat)

注意：ClusterClientReceptionist需要接收DoThis,DoThat消息并实现相关的运算。

在具体应用中要注意sender()的具体意义：从提供服务的actor方面看，sender()代表ClusterClientReceptionist。从发布消息的actor角度看，sender()代表的是DeadLetter。如果服务actor需要知道请求者具体地址，发布方可以把自己的地址嵌在发布的消息结构里。

下面我们就通过一个简单的例子来进行示范。先设计两个服务actor：Cat,Dog 。假设它们会提供不同的叫声作为服务吧：

  1. class Cat extends Actor with ActorLogging {
  2.   //使用pub/sub方式设置
  3.   val mediator = DistributedPubSub(context.system).mediator
  4.   override def preStart() = {
  5.     mediator ! Subscribe("Shout", self)
  6.     super.preStart()
  7.   }
  9.   override def receive: Receive = {
  10.     case "Shout" =>
  11.       log.info("*******I am a cat, MIAOM ...******")
  12.   }
  13. }
  15. class Dog extends Actor with ActorLogging {
  16.   //使用pub/sub方式设置
  17.   val mediator = DistributedPubSub(context.system).mediator
  18.   override def preStart() = {
  19.     mediator ! Subscribe("Shout", self)
  20.     super.preStart()
  21.   }
  22.   override def receive: Receive = {
  23.     case "Shout" =>
  24.       log.info("*****I am a dog, WANG WANG...*****")
  25.   }
  26. }

我们看到，这就是两个很普通的actor。但我们还是可以和上一篇分布式pub/sub结合起来验证cluster-client是基于distributed-pub/sub的。然后我们分别把这两个actor（服务）放到不同的集群节点上：

  1. object Cat {
  2.   def props = Props[Cat]
  3.   def create(port: Int): ActorSystem  = {
  4.     val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=$port")
  5.       .withFallback(ConfigFactory.load())
  7.     val system = ActorSystem("ClusterSystem",config)
  8.     val catSound = system.actorOf(props,"CatSound")
  10.     ClusterClientReceptionist(system).registerService(catSound)
  11.     system
  12.   }
  13. }
  15. object Dog {
  16.   def props = Props(new Dog)
  17.   def create(port: Int): ActorSystem = {
  18.     val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=$port")
  19.       .withFallback(ConfigFactory.load())
  20.     val system = ActorSystem("ClusterSystem",config)
  21.     val dogSound = system.actorOf(props,"DogSound")
  22.     ClusterClientReceptionist(system).registerService(dogSound)
  23.     system
  24.   }
  25. }

注意：集群名称是ClusterSystem。我们分别在actor所在节点用ClusterClientReceptionist.registerService登记了服务。这个集群所使用的conf如下：

  1. akka.actor.warn-about-java-serializer-usage = off
  2. akka.log-dead-letters-during-shutdown = off
  3. akka.log-dead-letters = off
  5. akka {
  6.   loglevel = INFO
  7.   extensions = ["akka.cluster.client.ClusterClientReceptionist"]
  8.   actor {
  9.     provider = "cluster"
  10.     serializers {
  11.       java = "akka.serialization.JavaSerializer"
  12.       proto = "akka.remote.serialization.ProtobufSerializer"
  13.     }
  14.     serialization-bindings {
  15.       "java.lang.String" = java
  16.       "scalapb.GeneratedMessage" = proto
  17.     }
  18.   }
  20.   remote {
  21.     log-remote-lifecycle-events = off
  22.     netty.tcp {
  23.       hostname = "127.0.0.1"
  24.       port = 0
  25.     }
  26.   }
  28.   cluster {
  29.     seed-nodes = [
  30.       "akka.tcp://ClusterSystem@127.0.0.1:2551"]
  31.     log-info = off
  32.   }
  33. }

这是一个比较完整的集群配置文档，只有port需要再配置。然后运行这两个节点：

  1. object PetHouse extends App {
  3.   val sysCat = Cat.create(2551)
  4.   val sysDog = Dog.create(2552)
  6.   scala.io.StdIn.readLine()
  8.   sysCat.terminate()
  9.   sysDog.terminate()
  11. }

完成了在2551,2552节点上的Cat,Dog actor构建及ClusterClientReceptionist.registerService服务登记。现在看看客户端：

  1. object PetClient extends App {
  3.   val conf = ConfigFactory.load("client")
  4.   val clientSystem = ActorSystem("ClientSystem",conf)
  5. /* 从 conf 文件里读取 contact-points 地址
  6.   val initialContacts = immutableSeq(conf.getStringList("contact-points")).map {
  7.     case AddressFromURIString(addr) ? RootActorPath(addr) / "system" / "receptionist"
  8.   }.toSet
  9. */
  11.   //先放一个contact-point, 系统会自动增加其它的点
  12.   val initialContacts = Set(
  13.     ActorPaths.fromString("akka.tcp://ClusterSystem@127.0.0.1:2551/system/receptionist")
  14.   )
  16.   val clusterClient = clientSystem.actorOf(
  17.     ClusterClient.props(
  18.       ClusterClientSettings(clientSystem)
  19.         .withInitialContacts(initialContacts)),
  20.     "petClient")
  22.   clusterClient ! Send("/user/CatSound","Shout",localAffinity = true)
  23.   clusterClient ! Send("/user/DogSound","Shout",localAffinity = true)
  25.   println(s"sent shout messages ...")
  26.   scala.io.StdIn.readLine()
  28.   clusterClient ! Publish("Shout","Shout")
  29.   println(s"publish shout messages ...")
  31.   scala.io.StdIn.readLine()
  32.   clientSystem.terminate();
  33. }

客户端的ActorSystem名称为ClientSystem，是在ClusterSystem集群之外的。conf文件如下：

  1. akka {
  3.   actor.provider = remote
  5.   remote.netty.tcp.port= 2553
  6.   remote.netty.tcp.hostname=127.0.0.1
  8. }
  10. contact-points = [
  11.   "akka.tcp://ClusterSystem@127.0.0.1:2551",
  12.   "akka.tcp://ClusterSystem@127.0.0.1:2552"]

把它设成actor.provider=remote可以免去提供seednodes。运算结果：

  1.  [12/08/2018 09:32:51.432] [ClusterSystem-akka.actor.default-dispatcher-17] [akka.tcp://ClusterSystem@127.0.0.1:2551/user/CatSound] *******I am a cat, MIAOM ...******
  2. [INFO] [12/08/2018 09:32:51.435] [ClusterSystem-akka.actor.default-dispatcher-19] [akka.tcp://ClusterSystem@127.0.0.1:2552/user/DogSound] *****I am a dog, WANG WANG...*****
  3. [INFO] [12/08/2018 09:33:44.113] [ClusterSystem-akka.actor.default-dispatcher-4] [akka.tcp://ClusterSystem@127.0.0.1:2551/user/CatSound] *******I am a cat, MIAOM ...******
  4. [INFO] [12/08/2018 09:33:44.113] [ClusterSystem-akka.actor.default-dispatcher-23] [akka.tcp://ClusterSystem@127.0.0.1:2552/user/DogSound] *****I am a dog, WANG WANG...*****

无论ClusterClient或Receptionist都会针对自己的状态发送消息。我们可以截取这些消息来做些相应的工作。参考下面的截听器示范代码： 

  1. package petsound
  2. import akka.actor._
  3. import akka.cluster.client._
  4. class ClientListener(clusterClient: ActorRef) extends Actor with ActorLogging {
  5.   override def preStart(): Unit = {
  6.     clusterClient ! SubscribeContactPoints
  7.     super.preStart()
  8.   }
  10.   override def receive: Receive = {
  11.     case ContactPoints(cps) =>
  12.       cps.map {ap => log.info(s"*******ContactPoints:${ap.address.toString}******")}
  13.     case ContactPointAdded(cp) =>
  14.       log.info(s"*******ContactPointAdded: ${cp.address.toString}*******")
  15.     case ContactPointRemoved(cp) =>
  16.       log.info(s"*******ContactPointRemoved: ${cp.address.toString}*******")
  18.   }
  19. }
  21. class ReceptionistListener(receptionist: ActorRef) extends Actor with ActorLogging {
  22.   override def preStart(): Unit = {
  23.     receptionist ! SubscribeClusterClients
  24.     super.preStart()
  25.   }
  27.   override def receive: Receive = {
  28.     case ClusterClients(cs) =>
  29.       cs.map{aref => println(s"*******ClusterClients: ${aref.path.address.toString}*******")}
  30.     case ClusterClientUp(cc) =>
  31.       log.info(s"*******ClusterClientUp: ${cc.path.address.toString}*******")
  32.     case ClusterClientUnreachable(cc) =>
  33.       log.info(s"*******ClusterClientUnreachable: ${cc.path.address.toString}*******")
  35.   }
  36. }

这两个event-listener的安装方法如下：

  1.     val receptionist = ClusterClientReceptionist(system).underlying
  2.     system.actorOf(Props(classOf[ReceptionistListener],receptionist),"cat-event-listner")
  4.     val receptionist = ClusterClientReceptionist(system).underlying
  5.     system.actorOf(Props(classOf[ReceptionistListener],receptionist),"dog-event-listner")
  7.   val clusterClient = clientSystem.actorOf(
  8.     ClusterClient.props(
  9.       ClusterClientSettings(clientSystem)
  10.         .withInitialContacts(initialContacts)),
  11.     "petClient")
  13.   clientSystem.actorOf(Props(classOf[ClientListener],clusterClient),"client-event-listner")

看看运算结果：

  1. [INFO] [12/09/2018 09:42:40.838] [ClientSystem-akka.actor.default-dispatcher-3] [akka.tcp://ClientSystem@127.0.0.1:2553/user/client-event-listner] *******ContactPoints:akka.tcp://ClusterSystem@127.0.0.1:2551******
  2. [INFO] [12/09/2018 09:42:40.947] [ClientSystem-akka.actor.default-dispatcher-3] [akka.tcp://ClientSystem@127.0.0.1:2553/user/client-event-listner] *******ContactPointAdded: akka.tcp://ClusterSystem@127.0.0.1:2552*******
  3. [INFO] [12/09/2018 09:42:40.967] [ClientSystem-akka.actor.default-dispatcher-15] [akka.tcp://ClientSystem@127.0.0.1:2553/user/petClient] Connected to [akka.tcp://ClusterSystem@127.0.0.1:2551/system/receptionist]
  6. [INFO] [12/09/2018 09:42:40.979] [ClusterSystem-akka.actor.default-dispatcher-4] [akka.tcp://ClusterSystem@127.0.0.1:2551/user/cat-event-listner] *******ClusterClientUp: akka.tcp://ClientSystem@127.0.0.1:2553*******
  8. [INFO] [12/09/2018 09:54:34.363] [ClusterSystem-akka.actor.default-dispatcher-19] [akka.tcp://ClusterSystem@127.0.0.1:2551/user/cat-event-listner] *******ClusterClientUnreachable: akka.tcp://ClientSystem@127.0.0.1:2553*******

下面我们再做个示范，还是与上篇讨论一样：由集群客户端发送MongoDB指令至某个在集群里用ClusterClientReceptionist注册的MongoDB操作服务actor。服务方接收指令后在MongoDB上进行运算。下面是MongoDB的服务actor： 

  1. package petsound
  2. import akka.actor._
  3. import com.typesafe.config._
  4. import akka.actor.ActorSystem
  5. import org.mongodb.scala._
  6. import sdp.grpc.services.ProtoMGOContext
  7. import sdp.mongo.engine.MGOClasses._
  8. import sdp.mongo.engine.MGOEngine._
  9. import sdp.result.DBOResult._
  10. import akka.cluster.client._
  12. import scala.collection.JavaConverters._
  13. import scala.util._
  15. class MongoAdder extends Actor with ActorLogging {
  16.   import monix.execution.Scheduler.Implicits.global
  17.   implicit val mgosys = context.system
  18.   implicit val ec = mgosys.dispatcher
  20.   val clientSettings: MongoClientSettings = MongoClientSettings.builder()
  21.     .applyToClusterSettings {b =>
  22.       b.hosts(List(new ServerAddress("localhost:27017")).asJava)
  23.     }.build()
  25.   implicit val client: MongoClient = MongoClient(clientSettings)
  27.   val ctx = MGOContext("testdb","friends")
  29.   override def receive: Receive = {
  31.     case someProto @ Some(proto:ProtoMGOContext) =>
  32.       val ctx = MGOContext.fromProto(proto)
  33.       log.info(s"****** received MGOContext: $someProto *********")
  35.       val task = mgoUpdate[Completed](ctx).toTask
  36.       task.runOnComplete {
  37.         case Success(s) => println("operations completed successfully.")
  38.         case Failure(exception) => println(s"error: ${exception.getMessage}")
  39.       }
  41.   }
  42. }
  44. object MongoAdder {
  46.   def create(port: Int): ActorSystem = {
  47.     val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=$port")
  48.       .withFallback(ConfigFactory.load())
  49.     val system = ActorSystem("ClusterSystem", config)
  51.     val mongoAdder = system.actorOf(Props[MongoAdder],"MongoAdder")
  52.     ClusterClientReceptionist(system).registerService(mongoAdder)
  54.     val receptionist = ClusterClientReceptionist(system).underlying
  55.     system.actorOf(Props(classOf[ReceptionistListener],receptionist),"mongo-event-listner")
  57.     system
  59.   }
  61. }

MongoAdder处于同一个集群ClusterSystem中。代码里已经包括了服务注册部分。客户端发送MongoDB指令的示范如下：

  1.  //MongoDB 操作示范
  2.   import org.mongodb.scala._
  3.   import sdp.mongo.engine.MGOClasses._
  5.   val ctx = MGOContext("testdb","friends")
  7.   val chen = Document("姓" -> "陈", "名" -> "大文","age" -> 28)
  8.   val zhang = Document("姓" -> "张", "名" -> "小海","age" -> 7)
  9.   val lee = Document("姓" -> "李", "名" -> "四","age" -> 45)
  10.   val ouyang = Document("姓" -> "欧阳", "名" -> "锋","age" -> 120)
  12.   val c = ctx.setCommand(MGOCommands.Insert(Seq(chen,zhang,lee,ouyang)))
  13.   clusterClient ! Send("/user/MongoAdder",c.toSomeProto,localAffinity = true)

由于MongoDB指令是通过protobuffer方式进行序列化的，所以需要修改client.conf通知akka使用protobuf格式的消息：

  1. akka {
  3.   actor {
  4.     provider = remote
  5.     serializers {
  6.       java = "akka.serialization.JavaSerializer"
  7.       proto = "akka.remote.serialization.ProtobufSerializer"
  8.     }
  9.     serialization-bindings {
  10.       "java.lang.String" = java
  11.       "scalapb.GeneratedMessage" = proto
  12.     }
  13.   }
  14.   remote.netty.tcp.port= 2553
  15.   remote.netty.tcp.hostname=127.0.0.1
  17. }
  19. contact-points = [
  20.   "akka.tcp://ClusterSystem@127.0.0.1:2551",
  21.   "akka.tcp://ClusterSystem@127.0.0.1:2552"]

下面是本次讨论完整示范源代码：

build.sbt

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

project/scalapb.sbt

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

resouces/application.conf

  1. akka.actor.warn-about-java-serializer-usage = off
  2. akka.log-dead-letters-during-shutdown = off
  3. akka.log-dead-letters = off
  5. akka {
  6.   loglevel = INFO
  7.   extensions = ["akka.cluster.client.ClusterClientReceptionist"]
  8.   actor {
  9.     provider = "cluster"
  10.     serializers {
  11.       java = "akka.serialization.JavaSerializer"
  12.       proto = "akka.remote.serialization.ProtobufSerializer"
  13.     }
  14.     serialization-bindings {
  15.       "java.lang.String" = java
  16.       "scalapb.GeneratedMessage" = proto
  17.     }
  18.   }
  20.   remote {
  21.     log-remote-lifecycle-events = off
  22.     netty.tcp {
  23.       hostname = "127.0.0.1"
  24.       port = 0
  25.     }
  26.   }
  28.   cluster {
  29.     seed-nodes = [
  30.       "akka.tcp://ClusterSystem@127.0.0.1:2551"]
  31.     log-info = off
  32.   }
  33. }

resources/client.conf

  1. akka {
  3.   actor {
  4.     provider = remote
  5.     serializers {
  6.       java = "akka.serialization.JavaSerializer"
  7.       proto = "akka.remote.serialization.ProtobufSerializer"
  8.     }
  9.     serialization-bindings {
  10.       "java.lang.String" = java
  11.       "scalapb.GeneratedMessage" = proto
  12.     }
  13.   }
  14.   remote.netty.tcp.port= 2553
  15.   remote.netty.tcp.hostname=127.0.0.1
  17. }
  19. contact-points = [
  20.   "akka.tcp://ClusterSystem@127.0.0.1:2551",
  21.   "akka.tcp://ClusterSystem@127.0.0.1:2552"]

protobuf/spd.proto

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

protobuf/mgo.proto

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

converters/ByteConverter.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.   }
  22. }

converters/DBOResultType.scala

  1. package sdp.result
  3. import cats._
  4. import cats.data.EitherT
  5. import cats.data.OptionT
  6. import monix.eval.Task
  7. import cats.implicits._
  9. import scala.concurrent._
  11. import scala.collection.TraversableOnce
  13. object DBOResult {
  16.   type DBOError[A] = EitherT[Task,Throwable,A]
  17.   type DBOResult[A] = OptionT[DBOError,A]
  19.   implicit def valueToDBOResult[A](a: A): DBOResult[A] =
  20.          Applicative[DBOResult].pure(a)
  21.   implicit def optionToDBOResult[A](o: Option[A]): DBOResult[A] =
  22.          OptionT((o: Option[A]).pure[DBOError])
  23.   implicit def eitherToDBOResult[A](e: Either[Throwable,A]): DBOResult[A] = {
  24.  //   val error: DBOError[A] = EitherT[Task,Throwable, A](Task.eval(e))
  25.          OptionT.liftF(EitherT.fromEither[Task](e))
  26.   }
  27.   implicit def futureToDBOResult[A](fut: Future[A]): DBOResult[A] = {
  28.        val task = Task.fromFuture[A](fut)
  29.        val et = EitherT.liftF[Task,Throwable,A](task)
  30.        OptionT.liftF(et)
  31.   }
  33.   implicit class DBOResultToTask[A](r: DBOResult[A]) {
  34.     def toTask = r.value.value
  35.   }
  37.   implicit class DBOResultToOption[A](r:Either[Throwable,Option[A]]) {
  38.     def someValue: Option[A] = r match {
  39.       case Left(err) => (None: Option[A])
  40.       case Right(oa) => oa
  41.     }
  42.   }
  44.   def wrapCollectionInOption[A, C[_] <: TraversableOnce[_]](coll: C[A]): DBOResult[C[A]] =
  45.     if (coll.isEmpty)
  46.       optionToDBOResult(None: Option[C[A]])
  47.     else
  48.       optionToDBOResult(Some(coll): Option[C[A]])
  49. }

filestream/FileStreaming.scala

  1. package sdp.file
  3. import java.io.{ByteArrayInputStream, InputStream}
  4. import java.nio.ByteBuffer
  5. import java.nio.file.Paths
  7. import akka.stream.Materializer
  8. import akka.stream.scaladsl.{FileIO, StreamConverters}
  9. import akka.util._
  11. import scala.concurrent.Await
  12. import scala.concurrent.duration._
  14. object Streaming {
  15.   def FileToByteBuffer(fileName: String, timeOut: FiniteDuration = 60 seconds)(
  16.     implicit mat: Materializer):ByteBuffer = {
  17.     val fut = FileIO.fromPath(Paths.get(fileName)).runFold(ByteString()) { case (hd, bs) =>
  18.       hd ++ bs
  19.     }
  20.     (Await.result(fut, timeOut)).toByteBuffer
  21.   }
  24.   def FileToByteArray(fileName: String, timeOut: FiniteDuration = 60 seconds)(
  25.     implicit mat: Materializer): Array[Byte] = {
  26.     val fut = FileIO.fromPath(Paths.get(fileName)).runFold(ByteString()) { case (hd, bs) =>
  27.       hd ++ bs
  28.     }
  29.     (Await.result(fut, timeOut)).toArray
  30.   }
  32.   def FileToInputStream(fileName: String, timeOut: FiniteDuration = 60 seconds)(
  33.     implicit mat: Materializer): InputStream = {
  34.     val fut = FileIO.fromPath(Paths.get(fileName)).runFold(ByteString()) { case (hd, bs) =>
  35.       hd ++ bs
  36.     }
  37.     val buf = (Await.result(fut, timeOut)).toArray
  38.     new ByteArrayInputStream(buf)
  39.   }
  41.   def ByteBufferToFile(byteBuf: ByteBuffer, fileName: String)(
  42.     implicit mat: Materializer) = {
  43.     val ba = new Array[Byte](byteBuf.remaining())
  44.     byteBuf.get(ba,0,ba.length)
  45.     val baInput = new ByteArrayInputStream(ba)
  46.     val source = StreamConverters.fromInputStream(() => baInput)  //ByteBufferInputStream(bytes))
  47.     source.runWith(FileIO.toPath(Paths.get(fileName)))
  48.   }
  50.   def ByteArrayToFile(bytes: Array[Byte], fileName: String)(
  51.     implicit mat: Materializer) = {
  52.     val bb = ByteBuffer.wrap(bytes)
  53.     val baInput = new ByteArrayInputStream(bytes)
  54.     val source = StreamConverters.fromInputStream(() => baInput) //ByteBufferInputStream(bytes))
  55.     source.runWith(FileIO.toPath(Paths.get(fileName)))
  56.   }
  58.   def InputStreamToFile(is: InputStream, fileName: String)(
  59.     implicit mat: Materializer) = {
  60.     val source = StreamConverters.fromInputStream(() => is)
  61.     source.runWith(FileIO.toPath(Paths.get(fileName)))
  62.   }
  64. }

logging/Log.scala

  1. package sdp.logging
  3. import org.slf4j.Logger
  5. /**
  6.   * Logger which just wraps org.slf4j.Logger internally.
  7.   *
  8.   * @param logger logger
  9.   */
  10. class Log(logger: Logger) {
  12.   // use var consciously to enable squeezing later
  13.   var isDebugEnabled: Boolean = logger.isDebugEnabled
  14.   var isInfoEnabled: Boolean = logger.isInfoEnabled
  15.   var isWarnEnabled: Boolean = logger.isWarnEnabled
  16.   var isErrorEnabled: Boolean = logger.isErrorEnabled
  18.   def withLevel(level: Symbol)(msg: => String, e: Throwable = null): Unit = {
  19.     level match {
  20.       case 'debug | 'DEBUG => debug(msg)
  21.       case 'info | 'INFO => info(msg)
  22.       case 'warn | 'WARN => warn(msg)
  23.       case 'error | 'ERROR => error(msg)
  24.       case _ => // nothing to do
  25.     }
  26.   }
  28.   def debug(msg: => String): Unit = {
  29.     if (isDebugEnabled && logger.isDebugEnabled) {
  30.       logger.debug(msg)
  31.     }
  32.   }
  34.   def debug(msg: => String, e: Throwable): Unit = {
  35.     if (isDebugEnabled && logger.isDebugEnabled) {
  36.       logger.debug(msg, e)
  37.     }
  38.   }
  40.   def info(msg: => String): Unit = {
  41.     if (isInfoEnabled && logger.isInfoEnabled) {
  42.       logger.info(msg)
  43.     }
  44.   }
  46.   def info(msg: => String, e: Throwable): Unit = {
  47.     if (isInfoEnabled && logger.isInfoEnabled) {
  48.       logger.info(msg, e)
  49.     }
  50.   }
  52.   def warn(msg: => String): Unit = {
  53.     if (isWarnEnabled && logger.isWarnEnabled) {
  54.       logger.warn(msg)
  55.     }
  56.   }
  58.   def warn(msg: => String, e: Throwable): Unit = {
  59.     if (isWarnEnabled && logger.isWarnEnabled) {
  60.       logger.warn(msg, e)
  61.     }
  62.   }
  64.   def error(msg: => String): Unit = {
  65.     if (isErrorEnabled && logger.isErrorEnabled) {
  66.       logger.error(msg)
  67.     }
  68.   }
  70.   def error(msg: => String, e: Throwable): Unit = {
  71.     if (isErrorEnabled && logger.isErrorEnabled) {
  72.       logger.error(msg, e)
  73.     }
  74.   }
  76. }

logging/LogSupport.scala

  1. package sdp.logging
  3. import org.slf4j.LoggerFactory
  5. trait LogSupport {
  7.   /**
  8.     * Logger
  9.     */
  10.   protected val log = new Log(LoggerFactory.getLogger(this.getClass))
  12. }

mgo/engine/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 MGOClasses._
  7. import MGOAdmins._
  8. import MGOCommands._
  9. import org.bson.BsonDocument
  10. import org.bson.codecs.configuration.CodecRegistry
  11. import org.mongodb.scala.bson.codecs.DEFAULT_CODEC_REGISTRY
  12. import org.mongodb.scala.FindObservable
  14. object MGOProtoConversion {
  16.   type MGO_COMMAND_TYPE = Int
  17.   val MGO_COMMAND_FIND            = 0
  18.   val MGO_COMMAND_COUNT           = 20
  19.   val MGO_COMMAND_DISTICT         = 21
  20.   val MGO_COMMAND_DOCUMENTSTREAM  = 1
  21.   val MGO_COMMAND_AGGREGATE       = 2
  22.   val MGO_COMMAND_INSERT          = 3
  23.   val MGO_COMMAND_DELETE          = 4
  24.   val MGO_COMMAND_REPLACE         = 5
  25.   val MGO_COMMAND_UPDATE          = 6
  28.   val MGO_ADMIN_DROPCOLLECTION    = 8
  29.   val MGO_ADMIN_CREATECOLLECTION  = 9
  30.   val MGO_ADMIN_LISTCOLLECTION    = 10
  31.   val MGO_ADMIN_CREATEVIEW        = 11
  32.   val MGO_ADMIN_CREATEINDEX       = 12
  33.   val MGO_ADMIN_DROPINDEXBYNAME   = 13
  34.   val MGO_ADMIN_DROPINDEXBYKEY    = 14
  35.   val MGO_ADMIN_DROPALLINDEXES    = 15
  36.  
  37.  
  38.   case class AdminContext(
  39.                            tarName: String = "",
  40.                            bsonParam: Seq[Bson] = Nil,
  41.                            options: Option[Any] = None,
  42.                            objName: String = ""
  43.                          ){
  44.     def toProto = sdp.grpc.services.ProtoMGOAdmin(
  45.       tarName = this.tarName,
  46.       bsonParam = this.bsonParam.map {b => sdp.grpc.services.ProtoMGOBson(marshal(b))},
  47.       objName = this.objName,
  48.       options = this.options.map(b => ProtoAny(marshal(b)))
  50.     )
  51.   }
  53.   object AdminContext {
  54.     def fromProto(msg: sdp.grpc.services.ProtoMGOAdmin) = new AdminContext(
  55.       tarName = msg.tarName,
  56.       bsonParam = msg.bsonParam.map(b => unmarshal[Bson](b.bson)),
  57.       objName = msg.objName,
  58.       options = msg.options.map(b => unmarshal[Any](b.value))
  59.     )
  60.   }
  62.   case class Context(
  63.                       dbName: String = "",
  64.                       collName: String = "",
  65.                       commandType: MGO_COMMAND_TYPE,
  66.                       bsonParam: Seq[Bson] = Nil,
  67.                       resultOptions: Seq[ResultOptions] = Nil,
  68.                       options: Option[Any] = None,
  69.                       documents: Seq[Document] = Nil,
  70.                       targets: Seq[String] = Nil,
  71.                       only: Boolean = false,
  72.                       adminOptions: Option[AdminContext] = None
  73.                     ){
  75.     def toProto = new sdp.grpc.services.ProtoMGOContext(
  76.       dbName = this.dbName,
  77.       collName = this.collName,
  78.       commandType = this.commandType,
  79.       bsonParam = this.bsonParam.map(bsonToProto),
  80.       resultOptions = this.resultOptions.map(_.toProto),
  81.       options = { if(this.options == None)
  82.         None //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
  83.       else
  84.         Some(ProtoAny(marshal(this.options.get))) },
  85.       documents = this.documents.map(d => sdp.grpc.services.ProtoMGODocument(marshal(d))),
  86.       targets = this.targets,
  87.       only = Some(this.only),
  88.       adminOptions = this.adminOptions.map(_.toProto)
  89.     )
  91.   }
  93.   object MGODocument {
  94.     def fromProto(msg: sdp.grpc.services.ProtoMGODocument): Document =
  95.       unmarshal[Document](msg.document)
  96.     def toProto(doc: Document): sdp.grpc.services.ProtoMGODocument =
  97.       new ProtoMGODocument(marshal(doc))
  98.   }
  100.   object MGOProtoMsg {
  101.     def fromProto(msg: sdp.grpc.services.ProtoMGOContext) = new Context(
  102.       dbName = msg.dbName,
  103.       collName = msg.collName,
  104.       commandType = msg.commandType,
  105.       bsonParam = msg.bsonParam.map(protoToBson),
  106.       resultOptions = msg.resultOptions.map(r => ResultOptions.fromProto(r)),
  107.       options = msg.options.map(a => unmarshal[Any](a.value)),
  108.       documents = msg.documents.map(doc => unmarshal[Document](doc.document)),
  109.       targets = msg.targets,
  110.       adminOptions = msg.adminOptions.map(ado => AdminContext.fromProto(ado))
  111.     )
  112.   }
  114.   def bsonToProto(bson: Bson) =
  115.     ProtoMGOBson(marshal(bson.toBsonDocument(
  116.       classOf[org.mongodb.scala.bson.collection.immutable.Document],DEFAULT_CODEC_REGISTRY)))
  118.   def protoToBson(proto: ProtoMGOBson): Bson = new Bson {
  119.     val bsdoc = unmarshal[BsonDocument](proto.bson)
  120.     override def toBsonDocument[TDocument](documentClass: Class[TDocument], codecRegistry: CodecRegistry): BsonDocument = bsdoc
  121.   }
  123.   def ctxFromProto(proto: ProtoMGOContext): MGOContext = proto.commandType match {
  124.     case MGO_COMMAND_FIND => {
  125.       var ctx = new MGOContext(
  126.         dbName = proto.dbName,
  127.         collName = proto.collName,
  128.         actionType = MGO_QUERY,
  129.         action = Some(Find())
  130.       )
  131.       def toResultOption(rts: Seq[ProtoMGOResultOption]): FindObservable[Document] => FindObservable[Document] = findObj =>
  132.         rts.foldRight(findObj)((a,b) => ResultOptions.fromProto(a).toFindObservable(b))
  134.       (proto.bsonParam, proto.resultOptions, proto.only) match {
  135.         case (Nil, Nil, None) => ctx
  136.         case (Nil, Nil, Some(b)) => ctx.setCommand(Find(firstOnly = b))
  137.         case (bp,Nil,None) => ctx.setCommand(
  138.           Find(filter = Some(protoToBson(bp.head))))
  139.         case (bp,Nil,Some(b)) => ctx.setCommand(
  140.           Find(filter = Some(protoToBson(bp.head)), firstOnly = b))
  141.         case (bp,fo,None) => {
  142.           ctx.setCommand(
  143.             Find(filter = Some(protoToBson(bp.head)),
  144.               andThen = fo.map(ResultOptions.fromProto)
  145.             ))
  146.         }
  147.         case (bp,fo,Some(b)) => {
  148.           ctx.setCommand(
  149.             Find(filter = Some(protoToBson(bp.head)),
  150.               andThen = fo.map(ResultOptions.fromProto),
  151.               firstOnly = b))
  152.         }
  153.         case _ => ctx
  154.       }
  155.     }
  156.     case MGO_COMMAND_COUNT => {
  157.       var ctx = new MGOContext(
  158.         dbName = proto.dbName,
  159.         collName = proto.collName,
  160.         actionType = MGO_QUERY,
  161.         action = Some(Count())
  162.       )
  163.       (proto.bsonParam, proto.options) match {
  164.         case (Nil, None) => ctx
  165.         case (bp, None) => ctx.setCommand(
  166.           Count(filter = Some(protoToBson(bp.head)))
  167.         )
  168.         case (Nil,Some(o)) => ctx.setCommand(
  169.           Count(options = Some(unmarshal[Any](o.value)))
  170.         )
  171.         case _ => ctx
  172.       }
  173.     }
  174.     case MGO_COMMAND_DISTICT => {
  175.       var ctx = new MGOContext(
  176.         dbName = proto.dbName,
  177.         collName = proto.collName,
  178.         actionType = MGO_QUERY,
  179.         action = Some(Distict(fieldName = proto.targets.head))
  180.       )
  181.       (proto.bsonParam) match {
  182.         case Nil => ctx
  183.         case bp: Seq[ProtoMGOBson] => ctx.setCommand(
  184.           Distict(fieldName = proto.targets.head,filter = Some(protoToBson(bp.head)))
  185.         )
  186.         case _ => ctx
  187.       }
  188.     }
  189.     case MGO_COMMAND_AGGREGATE => {
  190.       new MGOContext(
  191.         dbName = proto.dbName,
  192.         collName = proto.collName,
  193.         actionType = MGO_QUERY,
  194.         action = Some(Aggregate(proto.bsonParam.map(p => protoToBson(p))))
  195.       )
  196.     }
  197.     case MGO_ADMIN_LISTCOLLECTION => {
  198.       new MGOContext(
  199.         dbName = proto.dbName,
  200.         collName = proto.collName,
  201.         actionType = MGO_QUERY,
  202.         action = Some(ListCollection(proto.dbName)))
  203.     }
  204.     case MGO_COMMAND_INSERT => {
  205.       var ctx = new MGOContext(
  206.         dbName = proto.dbName,
  207.         collName = proto.collName,
  208.         actionType = MGO_UPDATE,
  209.         action = Some(Insert(
  210.           newdocs = proto.documents.map(doc => unmarshal[Document](doc.document))))
  211.       )
  212.       proto.options match {
  213.         case None => ctx
  214.         case Some(o) => ctx.setCommand(Insert(
  215.           newdocs = proto.documents.map(doc => unmarshal[Document](doc.document)),
  216.           options = Some(unmarshal[Any](o.value)))
  217.         )
  218.       }
  219.     }
  220.     case MGO_COMMAND_DELETE => {
  221.       var ctx = new MGOContext(
  222.         dbName = proto.dbName,
  223.         collName = proto.collName,
  224.         actionType = MGO_UPDATE,
  225.         action = Some(Delete(
  226.           filter = protoToBson(proto.bsonParam.head)))
  227.       )
  228.       (proto.options, proto.only) match {
  229.         case (None,None) => ctx
  230.         case (None,Some(b)) => ctx.setCommand(Delete(
  231.           filter = protoToBson(proto.bsonParam.head),
  232.           onlyOne = b))
  233.         case (Some(o),None) => ctx.setCommand(Delete(
  234.           filter = protoToBson(proto.bsonParam.head),
  235.           options = Some(unmarshal[Any](o.value)))
  236.         )
  237.         case (Some(o),Some(b)) => ctx.setCommand(Delete(
  238.           filter = protoToBson(proto.bsonParam.head),
  239.           options = Some(unmarshal[Any](o.value)),
  240.           onlyOne = b)
  241.         )
  242.       }
  243.     }
  244.     case MGO_COMMAND_REPLACE => {
  245.       var ctx = new MGOContext(
  246.         dbName = proto.dbName,
  247.         collName = proto.collName,
  248.         actionType = MGO_UPDATE,
  249.         action = Some(Replace(
  250.           filter = protoToBson(proto.bsonParam.head),
  251.           replacement = unmarshal[Document](proto.documents.head.document)))
  252.       )
  253.       proto.options match {
  254.         case None => ctx
  255.         case Some(o) => ctx.setCommand(Replace(
  256.           filter = protoToBson(proto.bsonParam.head),
  257.           replacement = unmarshal[Document](proto.documents.head.document),
  258.           options = Some(unmarshal[Any](o.value)))
  259.         )
  260.       }
  261.     }
  262.     case MGO_COMMAND_UPDATE => {
  263.       var ctx = new MGOContext(
  264.         dbName = proto.dbName,
  265.         collName = proto.collName,
  266.         actionType = MGO_UPDATE,
  267.         action = Some(Update(
  268.           filter = protoToBson(proto.bsonParam.head),
  269.           update = protoToBson(proto.bsonParam.tail.head)))
  270.       )
  271.       (proto.options, proto.only) match {
  272.         case (None,None) => ctx
  273.         case (None,Some(b)) => ctx.setCommand(Update(
  274.           filter = protoToBson(proto.bsonParam.head),
  275.           update = protoToBson(proto.bsonParam.tail.head),
  276.           onlyOne = b))
  277.         case (Some(o),None) => ctx.setCommand(Update(
  278.           filter = protoToBson(proto.bsonParam.head),
  279.           update = protoToBson(proto.bsonParam.tail.head),
  280.           options = Some(unmarshal[Any](o.value)))
  281.         )
  282.         case (Some(o),Some(b)) => ctx.setCommand(Update(
  283.           filter = protoToBson(proto.bsonParam.head),
  284.           update = protoToBson(proto.bsonParam.tail.head),
  285.           options = Some(unmarshal[Any](o.value)),
  286.           onlyOne = b)
  287.         )
  288.       }
  289.     }
  290.     case MGO_ADMIN_DROPCOLLECTION =>
  291.       new MGOContext(
  292.         dbName = proto.dbName,
  293.         collName = proto.collName,
  294.         actionType = MGO_ADMIN,
  295.         action = Some(DropCollection(proto.collName))
  296.       )
  297.     case MGO_ADMIN_CREATECOLLECTION => {
  298.       var ctx = new MGOContext(
  299.         dbName = proto.dbName,
  300.         collName = proto.collName,
  301.         actionType = MGO_ADMIN,
  302.         action = Some(CreateCollection(proto.collName))
  303.       )
  304.       proto.options match {
  305.         case None => ctx
  306.         case Some(o) => ctx.setCommand(CreateCollection(proto.collName,
  307.           options = Some(unmarshal[Any](o.value)))
  308.         )
  309.       }
  310.     }
  311.     case MGO_ADMIN_CREATEVIEW => {
  312.       var ctx = new MGOContext(
  313.         dbName = proto.dbName,
  314.         collName = proto.collName,
  315.         actionType = MGO_ADMIN,
  316.         action = Some(CreateView(viewName = proto.targets.head,
  317.           viewOn = proto.targets.tail.head,
  318.           pipeline = proto.bsonParam.map(p => protoToBson(p))))
  319.       )
  320.       proto.options match {
  321.         case None => ctx
  322.         case Some(o) => ctx.setCommand(CreateView(viewName = proto.targets.head,
  323.           viewOn = proto.targets.tail.head,
  324.           pipeline = proto.bsonParam.map(p => protoToBson(p)),
  325.           options = Some(unmarshal[Any](o.value)))
  326.         )
  327.       }
  328.     }
  329.     case MGO_ADMIN_CREATEINDEX=> {
  330.       var ctx = new MGOContext(
  331.         dbName = proto.dbName,
  332.         collName = proto.collName,
  333.         actionType = MGO_ADMIN,
  334.         action = Some(CreateIndex(key = protoToBson(proto.bsonParam.head)))
  335.       )
  336.       proto.options match {
  337.         case None => ctx
  338.         case Some(o) => ctx.setCommand(CreateIndex(key = protoToBson(proto.bsonParam.head),
  339.           options = Some(unmarshal[Any](o.value)))
  340.         )
  341.       }
  342.     }
  343.     case MGO_ADMIN_DROPINDEXBYNAME=> {
  344.       var ctx = new MGOContext(
  345.         dbName = proto.dbName,
  346.         collName = proto.collName,
  347.         actionType = MGO_ADMIN,
  348.         action = Some(DropIndexByName(indexName = proto.targets.head))
  349.       )
  350.       proto.options match {
  351.         case None => ctx
  352.         case Some(o) => ctx.setCommand(DropIndexByName(indexName = proto.targets.head,
  353.           options = Some(unmarshal[Any](o.value)))
  354.         )
  355.       }
  356.     }
  357.     case MGO_ADMIN_DROPINDEXBYKEY=> {
  358.       var ctx = new MGOContext(
  359.         dbName = proto.dbName,
  360.         collName = proto.collName,
  361.         actionType = MGO_ADMIN,
  362.         action = Some(DropIndexByKey(key = protoToBson(proto.bsonParam.head)))
  363.       )
  364.       proto.options match {
  365.         case None => ctx
  366.         case Some(o) => ctx.setCommand(DropIndexByKey(key = protoToBson(proto.bsonParam.head),
  367.           options = Some(unmarshal[Any](o.value)))
  368.         )
  369.       }
  370.     }
  371.     case MGO_ADMIN_DROPALLINDEXES=> {
  372.       var ctx = new MGOContext(
  373.         dbName = proto.dbName,
  374.         collName = proto.collName,
  375.         actionType = MGO_ADMIN,
  376.         action = Some(DropAllIndexes())
  377.       )
  378.       proto.options match {
  379.         case None => ctx
  380.         case Some(o) => ctx.setCommand(DropAllIndexes(
  381.           options = Some(unmarshal[Any](o.value)))
  382.         )
  383.       }
  384.     }
  386.   }
  388.   def ctxToProto(ctx: MGOContext): Option[sdp.grpc.services.ProtoMGOContext] = ctx.action match {
  389.     case None => None
  390.     case Some(act) => act match {
  391.       case Count(filter, options) =>
  392.         Some(new sdp.grpc.services.ProtoMGOContext(
  393.           dbName = ctx.dbName,
  394.           collName = ctx.collName,
  395.           commandType = MGO_COMMAND_COUNT,
  396.           bsonParam = { if (filter == None) Seq.empty[ProtoMGOBson]
  397.                         else Seq(bsonToProto(filter.get))},
  398.           options = { if(options == None) None  //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
  399.                       else Some(ProtoAny(marshal(options.get))) }
  400.       ))
  401.       case Distict(fieldName, filter) =>
  402.         Some(new sdp.grpc.services.ProtoMGOContext(
  403.           dbName = ctx.dbName,
  404.           collName = ctx.collName,
  405.           commandType = MGO_COMMAND_DISTICT,
  406.           bsonParam = { if (filter == None) Seq.empty[ProtoMGOBson]
  407.                         else Seq(bsonToProto(filter.get))},
  408.           targets = Seq(fieldName)
  410.         ))
  412.       case Find(filter, andThen, firstOnly) =>
  413.         Some(new sdp.grpc.services.ProtoMGOContext(
  414.           dbName = ctx.dbName,
  415.           collName = ctx.collName,
  416.           commandType = MGO_COMMAND_FIND,
  417.           bsonParam = { if (filter == None) Seq.empty[ProtoMGOBson]
  418.           else Seq(bsonToProto(filter.get))},
  419.           resultOptions = andThen.map(_.toProto)
  420.         ))
  422.       case Aggregate(pipeLine) =>
  423.         Some(new sdp.grpc.services.ProtoMGOContext(
  424.           dbName = ctx.dbName,
  425.           collName = ctx.collName,
  426.           commandType = MGO_COMMAND_AGGREGATE,
  427.           bsonParam = pipeLine.map(bsonToProto)
  428.         ))
  430.       case Insert(newdocs, options) =>
  431.         Some(new sdp.grpc.services.ProtoMGOContext(
  432.           dbName = ctx.dbName,
  433.           collName = ctx.collName,
  434.           commandType = MGO_COMMAND_INSERT,
  435.           documents = newdocs.map(d => ProtoMGODocument(marshal(d))),
  436.           options = { if(options == None) None      //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
  437.           else Some(ProtoAny(marshal(options.get))) }
  438.         ))
  440.       case Delete(filter, options, onlyOne) =>
  441.         Some(new sdp.grpc.services.ProtoMGOContext(
  442.           dbName = ctx.dbName,
  443.           collName = ctx.collName,
  444.           commandType = MGO_COMMAND_DELETE,
  445.           bsonParam = Seq(bsonToProto(filter)),
  446.           options = { if(options == None) None  //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
  447.           else Some(ProtoAny(marshal(options.get))) },
  448.           only = Some(onlyOne)
  449.         ))
  451.       case Replace(filter, replacement, options) =>
  452.         Some(new sdp.grpc.services.ProtoMGOContext(
  453.           dbName = ctx.dbName,
  454.           collName = ctx.collName,
  455.           commandType = MGO_COMMAND_REPLACE,
  456.           bsonParam = Seq(bsonToProto(filter)),
  457.           options = { if(options == None) None //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
  458.           else Some(ProtoAny(marshal(options.get))) },
  459.           documents = Seq(ProtoMGODocument(marshal(replacement)))
  460.         ))
  462.       case Update(filter, update, options, onlyOne) =>
  463.         Some(new sdp.grpc.services.ProtoMGOContext(
  464.           dbName = ctx.dbName,
  465.           collName = ctx.collName,
  466.           commandType = MGO_COMMAND_UPDATE,
  467.           bsonParam = Seq(bsonToProto(filter),bsonToProto(update)),
  468.           options = { if(options == None) None //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
  469.           else Some(ProtoAny(marshal(options.get))) },
  470.           only = Some(onlyOne)
  471.         ))
  474.       case DropCollection(coll) =>
  475.         Some(new sdp.grpc.services.ProtoMGOContext(
  476.           dbName = ctx.dbName,
  477.           collName = coll,
  478.           commandType = MGO_ADMIN_DROPCOLLECTION
  479.         ))
  481.       case CreateCollection(coll, options) =>
  482.         Some(new sdp.grpc.services.ProtoMGOContext(
  483.           dbName = ctx.dbName,
  484.           collName = coll,
  485.           commandType = MGO_ADMIN_CREATECOLLECTION,
  486.           options = { if(options == None) None  //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
  487.           else Some(ProtoAny(marshal(options.get))) }
  488.         ))
  490.       case ListCollection(dbName) =>
  491.         Some(new sdp.grpc.services.ProtoMGOContext(
  492.           dbName = ctx.dbName,
  493.           commandType = MGO_ADMIN_LISTCOLLECTION
  494.         ))
  496.       case CreateView(viewName, viewOn, pipeline, options) =>
  497.         Some(new sdp.grpc.services.ProtoMGOContext(
  498.           dbName = ctx.dbName,
  499.           collName = ctx.collName,
  500.           commandType = MGO_ADMIN_CREATEVIEW,
  501.           bsonParam = pipeline.map(bsonToProto),
  502.           options = { if(options == None) None  //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
  503.           else Some(ProtoAny(marshal(options.get))) },
  504.           targets = Seq(viewName,viewOn)
  505.         ))
  507.       case CreateIndex(key, options) =>
  508.         Some(new sdp.grpc.services.ProtoMGOContext(
  509.           dbName = ctx.dbName,
  510.           collName = ctx.collName,
  511.           commandType = MGO_ADMIN_CREATEINDEX,
  512.           bsonParam = Seq(bsonToProto(key)),
  513.           options = { if(options == None) None //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
  514.           else Some(ProtoAny(marshal(options.get))) }
  515.         ))
  518.       case DropIndexByName(indexName, options) =>
  519.         Some(new sdp.grpc.services.ProtoMGOContext(
  520.           dbName = ctx.dbName,
  521.           collName = ctx.collName,
  522.           commandType = MGO_ADMIN_DROPINDEXBYNAME,
  523.           targets = Seq(indexName),
  524.           options = { if(options == None) None //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
  525.           else Some(ProtoAny(marshal(options.get))) }
  526.         ))
  528.       case DropIndexByKey(key, options) =>
  529.         Some(new sdp.grpc.services.ProtoMGOContext(
  530.           dbName = ctx.dbName,
  531.           collName = ctx.collName,
  532.           commandType = MGO_ADMIN_DROPINDEXBYKEY,
  533.           bsonParam = Seq(bsonToProto(key)),
  534.           options = { if(options == None) None //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
  535.           else Some(ProtoAny(marshal(options.get))) }
  536.         ))
  539.       case DropAllIndexes(options) =>
  540.         Some(new sdp.grpc.services.ProtoMGOContext(
  541.           dbName = ctx.dbName,
  542.           collName = ctx.collName,
  543.           commandType = MGO_ADMIN_DROPALLINDEXES,
  544.           options = { if(options == None) None //Some(ProtoAny(com.google.protobuf.ByteString.EMPTY))
  545.           else Some(ProtoAny(marshal(options.get))) }
  546.         ))
  548.     }
  549.   }
  551. }

mgo/engine/MongoDBEngine.scala

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

PetSound.scala

  1. package petsound
  2. import akka.actor._
  3. import akka.cluster.client._
  4. import com.typesafe.config.ConfigFactory
  5. import akka.cluster.pubsub.DistributedPubSubMediator._
  6. import akka.cluster.pubsub._
  8. object Cat {
  9.   def props = Props[Cat]
  10.   def create(port: Int): ActorSystem  = {
  11.     val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=$port")
  12.       .withFallback(ConfigFactory.load())
  14.     val system = ActorSystem("ClusterSystem",config)
  15.     val catSound = system.actorOf(props,"CatSound")
  17.     ClusterClientReceptionist(system).registerService(catSound)
  19.     val receptionist = ClusterClientReceptionist(system).underlying
  20.     system.actorOf(Props(classOf[ReceptionistListener],receptionist),"cat-event-listner")
  22.     system
  23.   }
  24. }
  26. object Dog {
  27.   def props = Props(new Dog)
  28.   def create(port: Int): ActorSystem = {
  29.     val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=$port")
  30.       .withFallback(ConfigFactory.load())
  31.     val system = ActorSystem("ClusterSystem",config)
  32.     val dogSound = system.actorOf(props,"DogSound")
  33.     ClusterClientReceptionist(system).registerService(dogSound)
  35.     val receptionist = ClusterClientReceptionist(system).underlying
  36.     system.actorOf(Props(classOf[ReceptionistListener],receptionist),"dog-event-listner")
  38.     system
  39.   }
  40. }
  42. class Cat extends Actor with ActorLogging {
  43.   //使用pub/sub方式设置
  44.   val mediator = DistributedPubSub(context.system).mediator
  45.   override def preStart() = {
  46.     mediator ! Subscribe("Shout", self)
  47.     super.preStart()
  48.   }
  50.   override def receive: Receive = {
  51.     case "Shout" =>
  52.       log.info("*******I am a cat, MIAOM ...******")
  53.   }
  54. }
  56. class Dog extends Actor with ActorLogging {
  57.   //使用pub/sub方式设置
  58.   val mediator = DistributedPubSub(context.system).mediator
  59.   override def preStart() = {
  60.     mediator ! Subscribe("Shout", self)
  61.     super.preStart()
  62.   }
  63.   override def receive: Receive = {
  64.     case "Shout" =>
  65.       log.info("*****I am a dog, WANG WANG...*****")
  66.   }
  67. }

EventListener.scala

  1. package petsound
  2. import akka.actor._
  3. import akka.cluster.client._
  4. class ClientListener(clusterClient: ActorRef) extends Actor with ActorLogging {
  5.   override def preStart(): Unit = {
  6.     clusterClient ! SubscribeContactPoints
  7.     super.preStart()
  8.   }
  10.   override def receive: Receive = {
  11.     case ContactPoints(cps) =>
  12.       cps.map {ap => log.info(s"*******ContactPoints:${ap.address.toString}******")}
  13.     case ContactPointAdded(cp) =>
  14.       log.info(s"*******ContactPointAdded: ${cp.address.toString}*******")
  15.     case ContactPointRemoved(cp) =>
  16.       log.info(s"*******ContactPointRemoved: ${cp.address.toString}*******")
  18.   }
  19. }
  21. class ReceptionistListener(receptionist: ActorRef) extends Actor with ActorLogging {
  22.   override def preStart(): Unit = {
  23.     receptionist ! SubscribeClusterClients
  24.     super.preStart()
  25.   }
  27.   override def receive: Receive = {
  28.     case ClusterClients(cs) =>
  29.       cs.map{aref => println(s"*******ClusterClients: ${aref.path.address.toString}*******")}
  30.     case ClusterClientUp(cc) =>
  31.       log.info(s"*******ClusterClientUp: ${cc.path.address.toString}*******")
  32.     case ClusterClientUnreachable(cc) =>
  33.       log.info(s"*******ClusterClientUnreachable: ${cc.path.address.toString}*******")
  35.   }
  36. }

MongoAdder.scala

  1. package petsound
  2. import akka.actor._
  3. import com.typesafe.config._
  4. import akka.actor.ActorSystem
  5. import org.mongodb.scala._
  6. import sdp.grpc.services.ProtoMGOContext
  7. import sdp.mongo.engine.MGOClasses._
  8. import sdp.mongo.engine.MGOEngine._
  9. import sdp.result.DBOResult._
  10. import akka.cluster.client._
  12. import scala.collection.JavaConverters._
  13. import scala.util._
  15. class MongoAdder extends Actor with ActorLogging {
  16.   import monix.execution.Scheduler.Implicits.global
  17.   implicit val mgosys = context.system
  18.   implicit val ec = mgosys.dispatcher
  20.   val clientSettings: MongoClientSettings = MongoClientSettings.builder()
  21.     .applyToClusterSettings {b =>
  22.       b.hosts(List(new ServerAddress("localhost:27017")).asJava)
  23.     }.build()
  25.   implicit val client: MongoClient = MongoClient(clientSettings)
  27.   val ctx = MGOContext("testdb","friends")
  29.   override def receive: Receive = {
  31.     case someProto @ Some(proto:ProtoMGOContext) =>
  32.       val ctx = MGOContext.fromProto(proto)
  33.       log.info(s"****** received MGOContext: $someProto *********")
  35.       val task = mgoUpdate[Completed](ctx).toTask
  36.       task.runOnComplete {
  37.         case Success(s) => println("operations completed successfully.")
  38.         case Failure(exception) => println(s"error: ${exception.getMessage}")
  39.       }
  41.   }
  42. }
  44. object MongoAdder {
  46.   def create(port: Int): ActorSystem = {
  47.     val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=$port")
  48.       .withFallback(ConfigFactory.load())
  49.     val system = ActorSystem("ClusterSystem", config)
  51.     val mongoAdder = system.actorOf(Props[MongoAdder],"MongoAdder")
  52.     ClusterClientReceptionist(system).registerService(mongoAdder)
  54.     val receptionist = ClusterClientReceptionist(system).underlying
  55.     system.actorOf(Props(classOf[ReceptionistListener],receptionist),"mongo-event-listner")
  57.     system
  59.   }
  61. }

PetHouse.scala

  1. package petsound
  2. import akka.actor._
  3. import akka.japi.Util.immutableSeq
  4. import akka.actor.AddressFromURIString
  5. import com.typesafe.config.ConfigFactory
  6. import akka.cluster.client._
  7. import akka.cluster.client.ClusterClient._
  9. object PetHouse extends App {
  11.   val sysCat = Cat.create(2551)
  12.   val sysDog = Dog.create(2552)
  13.   val mongo = MongoAdder.create(2555)
  15.   scala.io.StdIn.readLine()
  17.   sysCat.terminate()
  18.   sysDog.terminate()
  19.   mongo.terminate()
  21. }
  24. object PetClient extends App {
  26.   val conf = ConfigFactory.load("client")
  27.   val clientSystem = ActorSystem("ClientSystem",conf)
  28. /* 从 conf 文件里读取 contact-points 地址
  29.   val initialContacts = immutableSeq(conf.getStringList("contact-points")).map {
  30.     case AddressFromURIString(addr) ? RootActorPath(addr) / "system" / "receptionist"
  31.   }.toSet
  32. */
  33.  
  34.   //先放一个contact-point, 系统会自动增加其它的点
  35.   val initialContacts = Set(
  36.     ActorPaths.fromString("akka.tcp://ClusterSystem@127.0.0.1:2551/system/receptionist")
  37.   )
  39.   val clusterClient = clientSystem.actorOf(
  40.     ClusterClient.props(
  41.       ClusterClientSettings(clientSystem)
  42.         .withInitialContacts(initialContacts)), "petClient")
  44.   clientSystem.actorOf(Props(classOf[ClientListener],clusterClient),"client-event-listner")
  46.   clusterClient ! Send("/user/CatSound","Shout",localAffinity = true)
  47.   clusterClient ! Send("/user/DogSound","Shout",localAffinity = true)
  49.   println(s"sent shout messages ...")
  50.   scala.io.StdIn.readLine()
  52.   clusterClient ! Publish("Shout","Shout")
  53.   println(s"publish shout messages ...")
  56.   //MongoDB 操作示范
  57.   import org.mongodb.scala._
  58.   import sdp.mongo.engine.MGOClasses._
  60.   val ctx = MGOContext("testdb","friends")
  62.   val chen = Document("姓" -> "陈", "名" -> "大文","age" -> 28)
  63.   val zhang = Document("姓" -> "张", "名" -> "小海","age" -> 7)
  64.   val lee = Document("姓" -> "李", "名" -> "四","age" -> 45)
  65.   val ouyang = Document("姓" -> "欧阳", "名" -> "锋","age" -> 120)
  67.   val c = ctx.setCommand(MGOCommands.Insert(Seq(chen,zhang,lee,ouyang)))
  68.   clusterClient ! Send("/user/MongoAdder",c.toSomeProto,localAffinity = true)
  71.   scala.io.StdIn.readLine()
  72.   clientSystem.terminate()
  73. }

 

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

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号