最近刚好有同事在学习MongoDB，我们讨论过MongoDB应该置于服务器端然后通过web-service为客户端提供数据的上传下载服务。我们可以用上节讨论的respapi框架来实现针对MongoDB的CRUD操作。在谈到restapi之前我在这篇讨论先介绍一下MongoDB数据库操作的scala编程，因为与传统的SQL数据库操作编程有比较大的差别。
在前面有关sdp (streaming-data-processor)系列的博文中有一段是关于MongoDBEngine的。刚好把这套工具的使用在这里介绍一下。
MongoDBEngine是基于mongodb-scala-driver上开发的一套MongoDB数据库CRUD Scala编程工具，其主要功能可以从下面这三个函数中反映出来：
代码语言：javascript复制 def mgoUpdate[T](ctx: MGOContext)(implicit client: MongoClient): DBOResult[T]
// T => FindIterable e.g List[Document]
def mgoQuery[T](ctx: MGOContext, Converter: Option[Document => Any] = None)(implicit client: MongoClient): DBOResult[T]
def mgoAdmin(ctx: MGOContext)(implicit client: MongoClient): DBOResult[Completed]
其中： mgoUpdate功能包括：insert,update,delete,replace ...
mgoQuery: find,count,distinct ...
mgoAdmin: dropCollection, createCollection ...
首先需要注意的是它们的返回结果类型： DBOResult[T]，实质上是 Future[Either[String,Option[T]]]
代码语言：javascript复制 type DBOError[A] = EitherT[Task,Throwable,A]
type DBOResult[A] = OptionT[DBOError,A]看起来很复杂，实际容易解释：设计这个类型的初衷是针对数据库操作的，所以：
1、异步操作，所以用Future （Task即Future, 如：Task.runToFuture）
2、返回结果可能为空，所以用Option
3、发生错误结果也为空，但需要知道空值是由错误产生的，所以用了Either
把所有返回结果类型统一成DBOResult[T]是为了方便进行函数组合，如：
代码语言：javascript复制for {
a <- mgoQuery(...)
_ <- mgoUpdate(a, ...)
b <- mgoQuery(...)
} yield b但另一方面也为写代码带来一些麻烦，如从结构中抽出运算结果值：
代码语言：javascript复制 mgoQuery[List[Document]](ctxFind).value.value.runToFuture {
case Success(eold) => eold match {
case Right(old) => old match {
case Some(ld) => ld.map(toPO(_)).foreach(showPO)
case None => println(s"Empty document found!")
}
case Left(err) => println(s"Error: ${err.getMessage}")
}是有些麻烦，不过能更详细的了解命令执行过程，而且是统一标准的写法（ctlr-c, ctlr-v 就可以了）。
上面三个函数都有一个同样的MGOContext类型的入参数，这是一个命令类型：
代码语言：javascript复制 case class MGOContext(
dbName: String,
collName: String,
actionType: MGO_ACTION_TYPE = MGO_QUERY,
action: Option[MGOCommands] = None,
actionOptions: Option[Any] = None,
actionTargets: Seq[String] = Nil
) {
ctx =>
def setDbName(name: String): MGOContext = ctx.copy(dbName = name)
def setCollName(name: String): MGOContext = ctx.copy(collName = name)
def setActionType(at: MGO_ACTION_TYPE): MGOContext = ctx.copy(actionType = at)
def setCommand(cmd: MGOCommands): MGOContext = ctx.copy(action = Some(cmd))
def toSomeProto = MGOProtoConversion.ctxToProto(this)
}
object MGOContext {
def apply(db: String, coll: String) = new MGOContext(db, coll)
def fromProto(proto: sdp.grpc.services.ProtoMGOContext): MGOContext =
MGOProtoConversion.ctxFromProto(proto)
}可以看到MGOContext.action就是具体的操作命令。下面是一个mgoAdmin命令的示范：
代码语言：javascript复制 val ctx = MGOContext("testdb","po").setCommand(
DropCollection("po"))
import monix.execution.Scheduler.Implicits.global
println(getResult(mgoAdmin(ctx).value.value.runToFuture))mgoUpdate示范：
代码语言：javascript复制 val optInsert = new InsertManyOptions().ordered(true)
val ctxInsert = ctx.setCommand(
Insert(Seq(po1,po2),Some(optInsert))
)
println(getResult(mgoUpdate(ctxInsert).value.value.runToFuture))我们选择MongoDB的主要目的是因为它分布式特性，适合大数据模式。但同时MongoDB具备强大的query功能，与传统SQL数据库模式相近，甚至还可以用索引。虽然MongoDB不支持数据关系，但对于我们这样的传统SQL老兵还是必然之选。MongoDB支持逗点查询组合，如：
代码语言：javascript复制 val resultDocType = FindIterable[Document]
val resultOption = FindObservable(resultDocType)
.maxScan(...)
.limit(...)
.sort(...)
.project(...) 比如对查询结果进行排序，同时又抽选几个返回的字段可以写成：FindObservable(...).sort(...).project(...)。MongoEngine提供了一个ResultOptions类型：
代码语言：javascript复制 case class ResultOptions(
optType: FOD_TYPE,
bson: Option[Bson] = None,
value: Int = 0 ){
def toFindObservable: FindObservable[Document] => FindObservable[Document] = find => {
optType match {
case FOD_FIRST => find
case FOD_FILTER => find.filter(bson.get)
case FOD_LIMIT => find.limit(value)
case FOD_SKIP => find.skip(value)
case FOD_PROJECTION => find.projection(bson.get)
case FOD_SORT => find.sort(bson.get)
case FOD_PARTIAL => find.partial(value != 0)
case FOD_CURSORTYPE => find
case FOD_HINT => find.hint(bson.get)
case FOD_MAX => find.max(bson.get)
case FOD_MIN => find.min(bson.get)
case FOD_RETURNKEY => find.returnKey(value != 0)
case FOD_SHOWRECORDID => find.showRecordId(value != 0)
}
}这个类型也是MGOContext类型的一个参数。 下面是一些用例：
代码语言：javascript复制 val ctxFilter = Find(filter=Some(equal("podtl.qty",100)))
val sort: Bson = (descending("ponum"))
val proj: Bson = (and(include("ponum","podate")
,include("vendor"),excludeId()))
val resSort = ResultOptions(FOD_SORT,Some(sort))
val resProj = ResultOptions(FOD_PROJECTION,Some(proj))
val ctxFind = ctx.setCommand(Find(andThen=Seq(resProj,resSort)))
val ctxFindFirst = ctx.setCommand(Find(firstOnly=true))
val ctxFindArrayItem = ctx.setCommand(
Find(filter = Some(equal("podtl.qty",100)))
)下面是一个完整的例子：
代码语言：javascript复制import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import org.mongodb.scala._
import scala.collection.JavaConverters._
import com.mongodb.client.model._
import com.datatech.sdp.mongo.engine._
import MGOClasses._
import scala.util._
object TestMongoEngine extends App {
import MGOEngine._
import MGOHelpers._
import MGOCommands._
import MGOAdmins._
// or provide custom MongoClientSettings
val settings: MongoClientSettings = MongoClientSettings.builder()
.applyToClusterSettings(b => b.hosts(List(new ServerAddress("localhost")).asJava))
.build()
implicit val client: MongoClient = MongoClient(settings)
implicit val system = ActorSystem()
implicit val mat = ActorMaterializer()
// implicit val ec = system.dispatcher
val ctx = MGOContext("testdb","po").setCommand(
DropCollection("po"))
import monix.execution.Scheduler.Implicits.global
println(getResult(mgoAdmin(ctx).value.value.runToFuture))
scala.io.StdIn.readLine()
val pic = fileToMGOBlob("/users/tiger/cert/MeTiger.png")
val po1 = Document (
"ponum" -> "po18012301",
"vendor" -> "The smartphone compay",
"podate" -> mgoDate(2017,5,13),
"remarks" -> "urgent, rush order",
"handler" -> pic,
"podtl" -> Seq(
Document("item" -> "sony smartphone", "price" -> 2389.00, "qty" -> 1239, "packing" -> "standard"),
Document("item" -> "ericson smartphone", "price" -> 897.00, "qty" -> 1000, "payterm" -> "30 days")
)
)
val po2 = Document (
"ponum" -> "po18022002",
"vendor" -> "The Samsung compay",
"podate" -> mgoDate(2015,11,6),
"podtl" -> Seq(
Document("item" -> "samsung galaxy s8", "price" -> 2300.00, "qty" -> 100, "packing" -> "standard"),
Document("item" -> "samsung galaxy s7", "price" -> 1897.00, "qty" -> 1000, "payterm" -> "30 days"),
Document("item" -> "apple iphone7", "price" -> 6500.00, "qty" -> 100, "packing" -> "luxury")
)
)
val optInsert = new InsertManyOptions().ordered(true)
val ctxInsert = ctx.setCommand(
Insert(Seq(po1,po2),Some(optInsert))
)
println(getResult(mgoUpdate(ctxInsert).value.value.runToFuture))
scala.io.StdIn.readLine()
case class PO (
ponum: String,
podate: MGODate,
vendor: String,
remarks: Option[String],
podtl: Option[MGOArray],
handler: Option[MGOBlob]
)
def toPO(doc: Document): PO = {
PO(
ponum = doc.getString("ponum"),
podate = doc.getDate("podate"),
vendor = doc.getString("vendor"),
remarks = mgoGetStringOrNone(doc,"remarks"),
podtl = mgoGetArrayOrNone(doc,"podtl"),
handler = mgoGetBlobOrNone(doc,"handler")
)
}
case class PODTL(
item: String,
price: Double,
qty: Int,
packing: Option[String],
payTerm: Option[String]
)
def toPODTL(podtl: Document): PODTL = {
PODTL(
item = podtl.getString("item"),
price = podtl.getDouble("price"),
qty = podtl.getInteger("qty"),
packing = mgoGetStringOrNone(podtl,"packing"),
payTerm = mgoGetStringOrNone(podtl,"payterm")
)
}
def showPO(po: PO) = {
println(s"po number: ${po.ponum}")
println(s"po date: ${mgoDateToString(po.podate,"yyyy-MM-dd")}")
println(s"vendor: ${po.vendor}")
if (po.remarks != None)
println(s"remarks: ${po.remarks.get}")
po.podtl match {
case Some(barr) =>
mgoArrayToDocumentList(barr)
.map { dc => toPODTL(dc)}
.foreach { doc: PODTL =>
print(s"==>Item: ${doc.item} ")
print(s"price: ${doc.price} ")
print(s"qty: ${doc.qty} ")
doc.packing.foreach(pk => print(s"packing: ${pk} "))
doc.payTerm.foreach(pt => print(s"payTerm: ${pt} "))
println("")
}
case _ =>
}
po.handler match {
case Some(blob) =>
val fileName = s"/users/tiger/${po.ponum}.png"
mgoBlobToFile(blob,fileName)
println(s"picture saved to ${fileName}")
case None => println("no picture provided")
}
}
import org.mongodb.scala.model.Projections._
import org.mongodb.scala.model.Filters._
import org.mongodb.scala.model.Sorts._
import org.mongodb.scala.bson.conversions._
import org.mongodb.scala.bson.Document
val ctxFilter = Find(filter=Some(equal("podtl.qty",100)))
val sort: Bson = (descending("ponum"))
val proj: Bson = (and(include("ponum","podate")
,include("vendor"),excludeId()))
val resSort = ResultOptions(FOD_TYPE.FOD_SORT,Some(sort))
val resProj = ResultOptions(FOD_TYPE.FOD_PROJECTION,Some(proj))
val ctxFind = ctx.setCommand(Find(andThen=Seq(resProj,resSort)))
val ctxFindFirst = ctx.setCommand(Find(firstOnly=true))
val ctxFindArrayItem = ctx.setCommand(
Find(filter = Some(equal("podtl.qty",100)))
)
for {
_ <- mgoQuery[List[Document]](ctxFind).value.value.runToFuture.andThen {
case Success(eold) => eold match {
case Right(old) => old match {
case Some(ld) => ld.map(toPO(_)).foreach(showPO)
case None => println(s"Empty document found!")
}
case Left(err) => println(s"Error: ${err.getMessage}")
}
println("-------------------------------")
case Failure(e) => println(e.getMessage)
}
_ <- mgoQuery[PO](ctxFindFirst,Some(toPO _)).value.value.runToFuture.andThen {
case Success(eop) => eop match {
case Right(op) => op match {
case Some(p) => showPO(_)
case None => println(s"Empty document found!")
}
case Left(err) => println(s"Error: ${err.getMessage}")
}
println("-------------------------------")
case Failure(e) => println(e.getMessage)
}
_ <- mgoQuery[List[PO]](ctxFindArrayItem,Some(toPO _)).value.value.runToFuture.andThen {
case Success(eops) => eops match {
case Right(ops) => ops match {
case Some(lp) => lp.foreach(showPO)
case None => println(s"Empty document found!")
}
case Left(err) => println(s"Error: ${err.getMessage}")
}
println("-------------------------------")
case Failure(e) => println(e.getMessage)
}
} yield()
scala.io.StdIn.readLine()
system.terminate()
}运行程序后结果如下：
代码语言：javascript复制Right(Some(The operation completed successfully))
Right(Some(The operation completed successfully))
po number: po18022002
po date: 2015-12-06
vendor: The Samsung compay
no picture provided
po number: po18012301
po date: 2017-06-13
vendor: The smartphone compay
no picture provided
-------------------------------
-------------------------------
po number: po18022002
po date: 2015-12-06
vendor: The Samsung compay
==>Item: samsung galaxy s8 price: 2300.0 qty: 100 packing: standard
==>Item: samsung galaxy s7 price: 1897.0 qty: 1000 payTerm: 30 days
==>Item: apple iphone7 price: 6500.0 qty: 100 packing: luxury
no picture provided
-------------------------------以下是本次讨论涉及的全部源代码：
build.sbt
代码语言：javascript复制name := "dt-dal"
version := "0.2"
scalaVersion := "2.12.8"
scalacOptions = "-Ypartial-unification"
val akkaVersion = "2.5.23"
val akkaHttpVersion = "10.1.8"
libraryDependencies := Seq(
// for scalikejdbc
"org.scalikejdbc" %% "scalikejdbc" % "3.2.1",
"org.scalikejdbc" %% "scalikejdbc-test" % "3.2.1" % "test",
"org.scalikejdbc" %% "scalikejdbc-config" % "3.2.1",
"org.scalikejdbc" %% "scalikejdbc-streams" % "3.2.1",
"org.scalikejdbc" %% "scalikejdbc-joda-time" % "3.2.1",
"com.h2database" % "h2" % "1.4.199",
"com.zaxxer" % "HikariCP" % "2.7.4",
"com.jolbox" % "bonecp" % "0.8.0.RELEASE",
"com.typesafe.slick" %% "slick" % "3.3.2",
//for cassandra 3.6.0
"com.datastax.cassandra" % "cassandra-driver-core" % "3.6.0",
"com.datastax.cassandra" % "cassandra-driver-extras" % "3.6.0",
"com.lightbend.akka" %% "akka-stream-alpakka-cassandra" % "1.1.0",
//for mongodb 4.0
"org.mongodb.scala" %% "mongo-scala-driver" % "2.6.0",
"com.lightbend.akka" %% "akka-stream-alpakka-mongodb" % "1.1.0",
"ch.qos.logback" % "logback-classic" % "1.2.3",
"io.monix" %% "monix" % "3.0.0-RC3",
"org.typelevel" %% "cats-core" % "2.0.0-M4"
)converters/DBOResultType.scala
代码语言：javascript复制package com.datatech.sdp.result
import cats._
import cats.data.EitherT
import cats.data.OptionT
import monix.eval.Task
import cats.implicits._
import scala.concurrent._
import scala.collection.TraversableOnce
object DBOResult {
type DBOError[A] = EitherT[Task,Throwable,A]
type DBOResult[A] = OptionT[DBOError,A]
implicit def valueToDBOResult[A](a: A): DBOResult[A] =
Applicative[DBOResult].pure(a)
implicit def optionToDBOResult[A](o: Option[A]): DBOResult[A] =
OptionT((o: Option[A]).pure[DBOError])
implicit def eitherToDBOResult[A](e: Either[Throwable,A]): DBOResult[A] = {
// val error: DBOError[A] = EitherT[Task,Throwable, A](Task.eval(e))
OptionT.liftF(EitherT.fromEither[Task](e))
}
implicit def futureToDBOResult[A](fut: Future[A]): DBOResult[A] = {
val task = Task.fromFuture[A](fut)
val et = EitherT.liftF[Task,Throwable,A](task)
OptionT.liftF(et)
}
implicit class DBOResultToTask[A](r: DBOResult[A]) {
def toTask = r.value.value
}
implicit class DBOResultToOption[A](r:Either[Throwable,Option[A]]) {
def someValue: Option[A] = r match {
case Left(err) => (None: Option[A])
case Right(oa) => oa
}
}
def wrapCollectionInOption[A, C[_] <: TraversableOnce[_]](coll: C[A]): DBOResult[C[A]] =
if (coll.isEmpty)
optionToDBOResult(None: Option[C[A]])
else
optionToDBOResult(Some(coll): Option[C[A]])
}filestream/FileStreaming.scala
代码语言：javascript复制package com.datatech.sdp.file
import java.io.{ByteArrayInputStream, InputStream}
import java.nio.ByteBuffer
import java.nio.file.Paths
import akka.stream.Materializer
import akka.stream.scaladsl.{FileIO, StreamConverters}
import akka.util._
import scala.concurrent.Await
import scala.concurrent.duration._
object Streaming {
def FileToByteBuffer(fileName: String, timeOut: FiniteDuration = 60 seconds)(
implicit mat: Materializer):ByteBuffer = {
val fut = FileIO.fromPath(Paths.get(fileName)).runFold(ByteString()) { case (hd, bs) =>
hd bs
}
(Await.result(fut, timeOut)).toByteBuffer
}
def FileToByteArray(fileName: String, timeOut: FiniteDuration = 60 seconds)(
implicit mat: Materializer): Array[Byte] = {
val fut = FileIO.fromPath(Paths.get(fileName)).runFold(ByteString()) { case (hd, bs) =>
hd bs
}
(Await.result(fut, timeOut)).toArray
}
def FileToInputStream(fileName: String, timeOut: FiniteDuration = 60 seconds)(
implicit mat: Materializer): InputStream = {
val fut = FileIO.fromPath(Paths.get(fileName)).runFold(ByteString()) { case (hd, bs) =>
hd bs
}
val buf = (Await.result(fut, timeOut)).toArray
new ByteArrayInputStream(buf)
}
def ByteBufferToFile(byteBuf: ByteBuffer, fileName: String)(
implicit mat: Materializer) = {
val ba = new Array[Byte](byteBuf.remaining())
byteBuf.get(ba,0,ba.length)
val baInput = new ByteArrayInputStream(ba)
val source = StreamConverters.fromInputStream(() => baInput) //ByteBufferInputStream(bytes))
source.runWith(FileIO.toPath(Paths.get(fileName)))
}
def ByteArrayToFile(bytes: Array[Byte], fileName: String)(
implicit mat: Materializer) = {
val bb = ByteBuffer.wrap(bytes)
val baInput = new ByteArrayInputStream(bytes)
val source = StreamConverters.fromInputStream(() => baInput) //ByteBufferInputStream(bytes))
source.runWith(FileIO.toPath(Paths.get(fileName)))
}
def InputStreamToFile(is: InputStream, fileName: String)(
implicit mat: Materializer) = {
val source = StreamConverters.fromInputStream(() => is)
source.runWith(FileIO.toPath(Paths.get(fileName)))
}
}logging/Log.scala
代码语言：javascript复制package com.datatech.sdp.logging
import org.slf4j.Logger
/**
* Logger which just wraps org.slf4j.Logger internally.
*
* @param logger logger
*/
class Log(logger: Logger) {
// use var consciously to enable squeezing later
var isDebugEnabled: Boolean = logger.isDebugEnabled
var isInfoEnabled: Boolean = logger.isInfoEnabled
var isWarnEnabled: Boolean = logger.isWarnEnabled
var isErrorEnabled: Boolean = logger.isErrorEnabled
var isTraceEnabled: Boolean = logger.isTraceEnabled
def withLevel(level: Symbol)(msg: => String, e: Throwable = null): Unit = {
level match {
case 'debug | 'DEBUG => debug(msg)
case 'info | 'INFO => info(msg)
case 'warn | 'WARN => warn(msg)
case 'error | 'ERROR => error(msg)
case 'trace | 'TRACE => trace(msg)
case _ => // nothing to do
}
}
var stepOn: Boolean = false
def step(msg: => String): Unit = {
if(stepOn)
logger.info("n****** {} ******n",msg)
}
def trace(msg: => String): Unit = {
if (isTraceEnabled && logger.isTraceEnabled) {
logger.trace(msg)
}
}
def trace(msg: => String, e: Throwable): Unit = {
if (isTraceEnabled && logger.isTraceEnabled) {
logger.trace(msg, e)
}
}
def debug(msg: => String): Unit = {
if (isDebugEnabled && logger.isDebugEnabled) {
logger.debug(msg)
}
}
def debug(msg: => String, e: Throwable): Unit = {
if (isDebugEnabled && logger.isDebugEnabled) {
logger.debug(msg, e)
}
}
def info(msg: => String): Unit = {
if (isInfoEnabled && logger.isInfoEnabled) {
logger.info(msg)
}
}
def info(msg: => String, e: Throwable): Unit = {
if (isInfoEnabled && logger.isInfoEnabled) {
logger.info(msg, e)
}
}
def warn(msg: => String): Unit = {
if (isWarnEnabled && logger.isWarnEnabled) {
logger.warn(msg)
}
}
def warn(msg: => String, e: Throwable): Unit = {
if (isWarnEnabled && logger.isWarnEnabled) {
logger.warn(msg, e)
}
}
def error(msg: => String): Unit = {
if (isErrorEnabled && logger.isErrorEnabled) {
logger.error(msg)
}
}
def error(msg: => String, e: Throwable): Unit = {
if (isErrorEnabled && logger.isErrorEnabled) {
logger.error(msg, e)
}
}
}logging/LogSupport.scala
代码语言：javascript复制package com.datatech.sdp.logging
import org.slf4j.LoggerFactory
trait LogSupport {
/**
* Logger
*/
protected val log = new Log(LoggerFactory.getLogger(this.getClass))
}mgo/engine/ObservableToPublisher.scala
代码语言：javascript复制package com.datatech.sdp.mongo.engine
import java.util.concurrent.atomic.AtomicBoolean
import org.mongodb.{scala => mongoDB}
import org.{reactivestreams => rxStreams}
final case class ObservableToPublisher[T](observable: mongoDB.Observable[T])
extends rxStreams.Publisher[T] {
def subscribe(subscriber: rxStreams.Subscriber[_ >: T]): Unit =
observable.subscribe(new mongoDB.Observer[T]() {
override def onSubscribe(subscription: mongoDB.Subscription): Unit =
subscriber.onSubscribe(new rxStreams.Subscription() {
private final val cancelled: AtomicBoolean = new AtomicBoolean
override def request(n: Long): Unit =
if (!subscription.isUnsubscribed && !cancelled.get() && n < 1) {
subscriber.onError(
new IllegalArgumentException(
s"Demand from publisher should be a positive amount. Current amount is:$n"
)
)
} else {
subscription.request(n)
}
override def cancel(): Unit =
if (!cancelled.getAndSet(true)) subscription.unsubscribe()
})
def onNext(result: T): Unit = subscriber.onNext(result)
def onError(e: Throwable): Unit = subscriber.onError(e)
def onComplete(): Unit = subscriber.onComplete()
})
}mgo/engine/MongoDBEngine.scala
代码语言：javascript复制package com.datatech.sdp.mongo.engine
import java.text.SimpleDateFormat
import java.util.Calendar
import akka.NotUsed
import akka.stream.Materializer
import akka.stream.alpakka.mongodb.scaladsl._
import akka.stream.scaladsl.{Flow, Source}
import org.bson.conversions.Bson
import org.mongodb.scala.bson.collection.immutable.Document
import org.mongodb.scala.bson.{BsonArray, BsonBinary}
import org.mongodb.scala.model._
import org.mongodb.scala.{MongoClient, _}
import com.datatech.sdp
import sdp.file.Streaming._
import sdp.logging.LogSupport
import scala.collection.JavaConverters._
import scala.concurrent._
import scala.concurrent.duration._
object MGOClasses {
type MGO_ACTION_TYPE = Int
object MGO_ACTION_TYPE {
val MGO_QUERY = 0
val MGO_UPDATE = 1
val MGO_ADMIN = 2
}
/* org.mongodb.scala.FindObservable
import com.mongodb.async.client.FindIterable
val resultDocType = FindIterable[Document]
val resultOption = FindObservable(resultDocType)
.maxScan(...)
.limit(...)
.sort(...)
.project(...) */
type FOD_TYPE = Int
object FOD_TYPE {
val FOD_FIRST = 0 //def first(): SingleObservable[TResult], return the first item
val FOD_FILTER = 1 //def filter(filter: Bson): FindObservable[TResult]
val FOD_LIMIT = 2 //def limit(limit: Int): FindObservable[TResult]
val FOD_SKIP = 3 //def skip(skip: Int): FindObservable[TResult]
val FOD_PROJECTION = 4 //def projection(projection: Bson): FindObservable[TResult]
//Sets a document describing the fields to return for all matching documents
val FOD_SORT = 5 //def sort(sort: Bson): FindObservable[TResult]
val FOD_PARTIAL = 6 //def partial(partial: Boolean): FindObservable[TResult]
//Get partial results from a sharded cluster if one or more shards are unreachable (instead of throwing an error)
val FOD_CURSORTYPE = 7 //def cursorType(cursorType: CursorType): FindObservable[TResult]
//Sets the cursor type
val FOD_HINT = 8 //def hint(hint: Bson): FindObservable[TResult]
//Sets the hint for which index to use. A null value means no hint is set
val FOD_MAX = 9 //def max(max: Bson): FindObservable[TResult]
//Sets the exclusive upper bound for a specific index. A null value means no max is set
val FOD_MIN = 10 //def min(min: Bson): FindObservable[TResult]
//Sets the minimum inclusive lower bound for a specific index. A null value means no max is set
val FOD_RETURNKEY = 11 //def returnKey(returnKey: Boolean): FindObservable[TResult]
//Sets the returnKey. If true the find operation will return only the index keys in the resulting documents
val FOD_SHOWRECORDID = 12 //def showRecordId(showRecordId: Boolean): FindObservable[TResult]
//Sets the showRecordId. Set to true to add a field `$recordId` to the returned documents
}
case class ResultOptions(
optType: FOD_TYPE,
bson: Option[Bson] = None,
value: Int = 0 ){
import FOD_TYPE._
def toFindObservable: FindObservable[Document] => FindObservable[Document] = find => {
optType match {
case FOD_FIRST => find
case FOD_FILTER => find.filter(bson.get)
case FOD_LIMIT => find.limit(value)
case FOD_SKIP => find.skip(value)
case FOD_PROJECTION => find.projection(bson.get)
case FOD_SORT => find.sort(bson.get)
case FOD_PARTIAL => find.partial(value != 0)
case FOD_CURSORTYPE => find
case FOD_HINT => find.hint(bson.get)
case FOD_MAX => find.max(bson.get)
case FOD_MIN => find.min(bson.get)
case FOD_RETURNKEY => find.returnKey(value != 0)
case FOD_SHOWRECORDID => find.showRecordId(value != 0)
}
}
}
trait MGOCommands
object MGOCommands {
case class Count(filter: Option[Bson] = None, options: Option[Any] = None) extends MGOCommands
case class Distict(fieldName: String, filter: Option[Bson] = None) extends MGOCommands
/* org.mongodb.scala.FindObservable
import com.mongodb.async.client.FindIterable
val resultDocType = FindIterable[Document]
val resultOption = FindObservable(resultDocType)
.maxScan(...)
.limit(...)
.sort(...)
.project(...) */
case class Find(filter: Option[Bson] = None,
andThen: Seq[ResultOptions] = Seq.empty[ResultOptions],
firstOnly: Boolean = false) extends MGOCommands
case class Aggregate(pipeLine: Seq[Bson]) extends MGOCommands
case class MapReduce(mapFunction: String, reduceFunction: String) extends MGOCommands
case class Insert(newdocs: Seq[Document], options: Option[Any] = None) extends MGOCommands
case class Delete(filter: Bson, options: Option[Any] = None, onlyOne: Boolean = false) extends MGOCommands
case class Replace(filter: Bson, replacement: Document, options: Option[Any] = None) extends MGOCommands
case class Update(filter: Bson, update: Bson, options: Option[Any] = None, onlyOne: Boolean = false) extends MGOCommands
case class BulkWrite(commands: List[WriteModel[Document]], options: Option[Any] = None) extends MGOCommands
}
object MGOAdmins {
case class DropCollection(collName: String) extends MGOCommands
case class CreateCollection(collName: String, options: Option[Any] = None) extends MGOCommands
case class ListCollection(dbName: String) extends MGOCommands
case class CreateView(viewName: String, viewOn: String, pipeline: Seq[Bson], options: Option[Any] = None) extends MGOCommands
case class CreateIndex(key: Bson, options: Option[Any] = None) extends MGOCommands
case class DropIndexByName(indexName: String, options: Option[Any] = None) extends MGOCommands
case class DropIndexByKey(key: Bson, options: Option[Any] = None) extends MGOCommands
case class DropAllIndexes(options: Option[Any] = None) extends MGOCommands
}
case class MGOContext(
dbName: String,
collName: String,
actionType: MGO_ACTION_TYPE = MGO_ACTION_TYPE.MGO_QUERY,
action: Option[MGOCommands] = None,
actionOptions: Option[Any] = None,
actionTargets: Seq[String] = Nil
) {
ctx =>
def setDbName(name: String): MGOContext = ctx.copy(dbName = name)
def setCollName(name: String): MGOContext = ctx.copy(collName = name)
def setActionType(at: MGO_ACTION_TYPE): MGOContext = ctx.copy(actionType = at)
def setCommand(cmd: MGOCommands): MGOContext = ctx.copy(action = Some(cmd))
}
object MGOContext {
def apply(db: String, coll: String) = new MGOContext(db, coll)
}
case class MGOBatContext(contexts: Seq[MGOContext], tx: Boolean = false) {
ctxs =>
def setTx(txopt: Boolean): MGOBatContext = ctxs.copy(tx = txopt)
def appendContext(ctx: MGOContext): MGOBatContext =
ctxs.copy(contexts = contexts : ctx)
}
object MGOBatContext {
def apply(ctxs: Seq[MGOContext], tx: Boolean = false) = new MGOBatContext(ctxs,tx)
}
type MGODate = java.util.Date
def mgoDate(yyyy: Int, mm: Int, dd: Int): MGODate = {
val ca = Calendar.getInstance()
ca.set(yyyy,mm,dd)
ca.getTime()
}
def mgoDateTime(yyyy: Int, mm: Int, dd: Int, hr: Int, min: Int, sec: Int): MGODate = {
val ca = Calendar.getInstance()
ca.set(yyyy,mm,dd,hr,min,sec)
ca.getTime()
}
def mgoDateTimeNow: MGODate = {
val ca = Calendar.getInstance()
ca.getTime
}
def mgoDateToString(dt: MGODate, formatString: String): String = {
val fmt= new SimpleDateFormat(formatString)
fmt.format(dt)
}
type MGOBlob = BsonBinary
type MGOArray = BsonArray
def fileToMGOBlob(fileName: String, timeOut: FiniteDuration = 60 seconds)(
implicit mat: Materializer) = FileToByteArray(fileName,timeOut)
def mgoBlobToFile(blob: MGOBlob, fileName: String)(
implicit mat: Materializer) = ByteArrayToFile(blob.getData,fileName)
def mgoGetStringOrNone(doc: Document, fieldName: String) = {
if (doc.keySet.contains(fieldName))
Some(doc.getString(fieldName))
else None
}
def mgoGetIntOrNone(doc: Document, fieldName: String) = {
if (doc.keySet.contains(fieldName))
Some(doc.getInteger(fieldName))
else None
}
def mgoGetLonggOrNone(doc: Document, fieldName: String) = {
if (doc.keySet.contains(fieldName))
Some(doc.getLong(fieldName))
else None
}
def mgoGetDoubleOrNone(doc: Document, fieldName: String) = {
if (doc.keySet.contains(fieldName))
Some(doc.getDouble(fieldName))
else None
}
def mgoGetBoolOrNone(doc: Document, fieldName: String) = {
if (doc.keySet.contains(fieldName))
Some(doc.getBoolean(fieldName))
else None
}
def mgoGetDateOrNone(doc: Document, fieldName: String) = {
if (doc.keySet.contains(fieldName))
Some(doc.getDate(fieldName))
else None
}
def mgoGetBlobOrNone(doc: Document, fieldName: String) = {
if (doc.keySet.contains(fieldName))
doc.get(fieldName).asInstanceOf[Option[MGOBlob]]
else None
}
def mgoGetArrayOrNone(doc: Document, fieldName: String) = {
if (doc.keySet.contains(fieldName))
doc.get(fieldName).asInstanceOf[Option[MGOArray]]
else None
}
def mgoArrayToDocumentList(arr: MGOArray): scala.collection.immutable.List[org.bson.BsonDocument] = {
(arr.getValues.asScala.toList)
.asInstanceOf[scala.collection.immutable.List[org.bson.BsonDocument]]
}
type MGOFilterResult = FindObservable[Document] => FindObservable[Document]
}
object MGOEngine extends LogSupport {
import MGOClasses._
import MGOAdmins._
import MGOCommands._
import sdp.result.DBOResult._
import com.mongodb.reactivestreams.client.MongoClients
object TxUpdateMode {
private def mgoTxUpdate(ctxs: MGOBatContext, observable: SingleObservable[ClientSession])(
implicit client: MongoClient, ec: ExecutionContext): SingleObservable[ClientSession] = {
log.info(s"mgoTxUpdate> calling ...")
observable.map(clientSession => {
val transactionOptions =
TransactionOptions.builder()
.readConcern(ReadConcern.SNAPSHOT)
.writeConcern(WriteConcern.MAJORITY).build()
clientSession.startTransaction(transactionOptions)
/*
val fut = Future.traverse(ctxs.contexts) { ctx =>
mgoUpdateObservable[Completed](ctx).map(identity).toFuture()
}
Await.ready(fut, 3 seconds) */
ctxs.contexts.foreach { ctx =>
mgoUpdateObservable[Completed](ctx).map(identity).toFuture()
}
clientSession
})
}
private def commitAndRetry(observable: SingleObservable[Completed]): SingleObservable[Completed] = {
log.info(s"commitAndRetry> calling ...")
observable.recoverWith({
case e: MongoException if e.hasErrorLabel(MongoException.UNKNOWN_TRANSACTION_COMMIT_RESULT_LABEL) => {
log.warn("commitAndRetry> UnknownTransactionCommitResult, retrying commit operation ...")
commitAndRetry(observable)
}
case e: Exception => {
log.error(s"commitAndRetry> Exception during commit ...: $e")
throw e
}
})
}
private def runTransactionAndRetry(observable: SingleObservable[Completed]): SingleObservable[Completed] = {
log.info(s"runTransactionAndRetry> calling ...")
observable.recoverWith({
case e: MongoException if e.hasErrorLabel(MongoException.TRANSIENT_TRANSACTION_ERROR_LABEL) => {
log.warn("runTransactionAndRetry> TransientTransactionError, aborting transaction and retrying ...")
runTransactionAndRetry(observable)
}
})
}
def mgoTxBatch(ctxs: MGOBatContext)(
implicit client: MongoClient, ec: ExecutionContext): DBOResult[Completed] = {
log.info(s"mgoTxBatch> MGOBatContext: ${ctxs}")
val updateObservable: Observable[ClientSession] = mgoTxUpdate(ctxs, client.startSession())
val commitTransactionObservable: SingleObservable[Completed] =
updateObservable.flatMap(clientSession => clientSession.commitTransaction())
val commitAndRetryObservable: SingleObservable[Completed] = commitAndRetry(commitTransactionObservable)
runTransactionAndRetry(commitAndRetryObservable)
valueToDBOResult(Completed())
}
}
def mgoUpdateBatch(ctxs: MGOBatContext)(implicit client: MongoClient, ec: ExecutionContext): DBOResult[Completed] = {
log.info(s"mgoUpdateBatch> MGOBatContext: ${ctxs}")
if (ctxs.tx) {
TxUpdateMode.mgoTxBatch(ctxs)
} else {
/*
val fut = Future.traverse(ctxs.contexts) { ctx =>
mgoUpdate[Completed](ctx).map(identity) }
Await.ready(fut, 3 seconds)
FastFastFuture.successful(new Completed) */
ctxs.contexts.foreach { ctx =>
mgoUpdate[Completed](ctx).map(identity) }
valueToDBOResult(Completed())
}
}
def mongoStream(ctx: MGOContext)(
implicit client: MongoClient, ec: ExecutionContextExecutor): Source[Document, NotUsed] = {
log.info(s"mongoStream> MGOContext: ${ctx}")
def toResultOption(rts: Seq[ResultOptions]): FindObservable[Document] => FindObservable[Document] = findObj =>
rts.foldRight(findObj)((a,b) => a.toFindObservable(b))
val db = client.getDatabase(ctx.dbName)
val coll = db.getCollection(ctx.collName)
if ( ctx.action == None) {
log.error(s"mongoStream> uery action cannot be null!")
throw new IllegalArgumentException("query action cannot be null!")
}
try {
ctx.action.get match {
case Find(None, Nil, false) => //FindObservable
MongoSource(ObservableToPublisher(coll.find()))
case Find(None, Nil, true) => //FindObservable
MongoSource(ObservableToPublisher(coll.find().first()))
case Find(Some(filter), Nil, false) => //FindObservable
MongoSource(ObservableToPublisher(coll.find(filter)))
case Find(Some(filter), Nil, true) => //FindObservable
MongoSource(ObservableToPublisher(coll.find(filter).first()))
case Find(None, sro, _) => //FindObservable
val next = toResultOption(sro)
MongoSource(ObservableToPublisher(next(coll.find[Document]())))
case Find(Some(filter), sro, _) => //FindObservable
val next = toResultOption(sro)
MongoSource(ObservableToPublisher(next(coll.find[Document](filter))))
case _ =>
log.error(s"mongoStream> unsupported streaming query [${ctx.action.get}]")
throw new RuntimeException(s"mongoStream> unsupported streaming query [${ctx.action.get}]")
}
}
catch { case e: Exception =>
log.error(s"mongoStream> runtime error: ${e.getMessage}")
throw new RuntimeException(s"mongoStream> Error: ${e.getMessage}")
}
}
// T => FindIterable e.g List[Document]
def mgoQuery[T](ctx: MGOContext, Converter: Option[Document => Any] = None)(implicit client: MongoClient): DBOResult[T] = {
log.info(s"mgoQuery> MGOContext: ${ctx}")
val db = client.getDatabase(ctx.dbName)
val coll = db.getCollection(ctx.collName)
def toResultOption(rts: Seq[ResultOptions]): FindObservable[Document] => FindObservable[Document] = findObj =>
rts.foldRight(findObj)((a,b) => a.toFindObservable(b))
if ( ctx.action == None) {
log.error(s"mgoQuery> uery action cannot be null!")
Left(new IllegalArgumentException("query action cannot be null!"))
}
try {
ctx.action.get match {
/* count */
case Count(Some(filter), Some(opt)) => //SingleObservable
coll.countDocuments(filter, opt.asInstanceOf[CountOptions])
.toFuture().asInstanceOf[Future[T]]
case Count(Some(filter), None) => //SingleObservable
coll.countDocuments(filter).toFuture()
.asInstanceOf[Future[T]]
case Count(None, None) => //SingleObservable
coll.countDocuments().toFuture()
.asInstanceOf[Future[T]]
/* distinct */
case Distict(field, Some(filter)) => //DistinctObservable
coll.distinct(field, filter).toFuture()
.asInstanceOf[Future[T]]
case Distict(field, None) => //DistinctObservable
coll.distinct((field)).toFuture()
.asInstanceOf[Future[T]]
/* find */
case Find(None, Nil, false) => //FindObservable
if (Converter == None) coll.find().toFuture().asInstanceOf[Future[T]]
else coll.find().map(Converter.get).toFuture().asInstanceOf[Future[T]]
case Find(None, Nil, true) => //FindObservable
if (Converter == None) coll.find().first().head().asInstanceOf[Future[T]]
else coll.find().first().map(Converter.get).head().asInstanceOf[Future[T]]
case Find(Some(filter), Nil, false) => //FindObservable
if (Converter == None) coll.find(filter).toFuture().asInstanceOf[Future[T]]
else coll.find(filter).map(Converter.get).toFuture().asInstanceOf[Future[T]]
case Find(Some(filter), Nil, true) => //FindObservable
if (Converter == None) coll.find(filter).first().head().asInstanceOf[Future[T]]
else coll.find(filter).first().map(Converter.get).head().asInstanceOf[Future[T]]
case Find(None, sro, _) => //FindObservable
val next = toResultOption(sro)
if (Converter == None) next(coll.find[Document]()).toFuture().asInstanceOf[Future[T]]
else next(coll.find[Document]()).map(Converter.get).toFuture().asInstanceOf[Future[T]]
case Find(Some(filter), sro, _) => //FindObservable
val next = toResultOption(sro)
if (Converter == None) next(coll.find[Document](filter)).toFuture().asInstanceOf[Future[T]]
else next(coll.find[Document](filter)).map(Converter.get).toFuture().asInstanceOf[Future[T]]
/* aggregate AggregateObservable*/
case Aggregate(pline) => coll.aggregate(pline).toFuture().asInstanceOf[Future[T]]
/* mapReduce MapReduceObservable*/
case MapReduce(mf, rf) => coll.mapReduce(mf, rf).toFuture().asInstanceOf[Future[T]]
/* list collection */
case ListCollection(dbName) => //ListConllectionObservable
client.getDatabase(dbName).listCollections().toFuture().asInstanceOf[Future[T]]
}
}
catch { case e: Exception =>
log.error(s"mgoQuery> runtime error: ${e.getMessage}")
Left(new RuntimeException(s"mgoQuery> Error: ${e.getMessage}"))
}
}
//T => Completed, result.UpdateResult, result.DeleteResult
def mgoUpdate[T](ctx: MGOContext)(implicit client: MongoClient): DBOResult[T] =
try {
mgoUpdateObservable[T](ctx).toFuture()
}
catch { case e: Exception =>
log.error(s"mgoUpdate> runtime error: ${e.getMessage}")
Left(new RuntimeException(s"mgoUpdate> Error: ${e.getMessage}"))
}
def mgoUpdateObservable[T](ctx: MGOContext)(implicit client: MongoClient): SingleObservable[T] = {
log.info(s"mgoUpdateObservable> MGOContext: ${ctx}")
val db = client.getDatabase(ctx.dbName)
val coll = db.getCollection(ctx.collName)
if ( ctx.action == None) {
log.error(s"mgoUpdateObservable> uery action cannot be null!")
throw new IllegalArgumentException("mgoUpdateObservable> query action cannot be null!")
}
try {
ctx.action.get match {
/* insert */
case Insert(docs, Some(opt)) => //SingleObservable[Completed]
if (docs.size > 1)
coll.insertMany(docs, opt.asInstanceOf[InsertManyOptions]).asInstanceOf[SingleObservable[T]]
else coll.insertOne(docs.head, opt.asInstanceOf[InsertOneOptions]).asInstanceOf[SingleObservable[T]]
case Insert(docs, None) => //SingleObservable
if (docs.size > 1) coll.insertMany(docs).asInstanceOf[SingleObservable[T]]
else coll.insertOne(docs.head).asInstanceOf[SingleObservable[T]]
/* delete */
case Delete(filter, None, onlyOne) => //SingleObservable
if (onlyOne) coll.deleteOne(filter).asInstanceOf[SingleObservable[T]]
else coll.deleteMany(filter).asInstanceOf[SingleObservable[T]]
case Delete(filter, Some(opt), onlyOne) => //SingleObservable
if (onlyOne) coll.deleteOne(filter, opt.asInstanceOf[DeleteOptions]).asInstanceOf[SingleObservable[T]]
else coll.deleteMany(filter, opt.asInstanceOf[DeleteOptions]).asInstanceOf[SingleObservable[T]]
/* replace */
case Replace(filter, replacement, None) => //SingleObservable
coll.replaceOne(filter, replacement).asInstanceOf[SingleObservable[T]]
case Replace(filter, replacement, Some(opt)) => //SingleObservable
coll.replaceOne(filter, replacement, opt.asInstanceOf[ReplaceOptions]).asInstanceOf[SingleObservable[T]]
/* update */
case Update(filter, update, None, onlyOne) => //SingleObservable
if (onlyOne) coll.updateOne(filter, update).asInstanceOf[SingleObservable[T]]
else coll.updateMany(filter, update).asInstanceOf[SingleObservable[T]]
case Update(filter, update, Some(opt), onlyOne) => //SingleObservable
if (onlyOne) coll.updateOne(filter, update, opt.asInstanceOf[UpdateOptions]).asInstanceOf[SingleObservable[T]]
else coll.updateMany(filter, update, opt.asInstanceOf[UpdateOptions]).asInstanceOf[SingleObservable[T]]
/* bulkWrite */
case BulkWrite(commands, None) => //SingleObservable
coll.bulkWrite(commands).asInstanceOf[SingleObservable[T]]
case BulkWrite(commands, Some(opt)) => //SingleObservable
coll.bulkWrite(commands, opt.asInstanceOf[BulkWriteOptions]).asInstanceOf[SingleObservable[T]]
}
}
catch { case e: Exception =>
log.error(s"mgoUpdateObservable> runtime error: ${e.getMessage}")
throw new RuntimeException(s"mgoUpdateObservable> Error: ${e.getMessage}")
}
}
def mgoAdmin(ctx: MGOContext)(implicit client: MongoClient): DBOResult[Completed] = {
log.info(s"mgoAdmin> MGOContext: ${ctx}")
val db = client.getDatabase(ctx.dbName)
val coll = db.getCollection(ctx.collName)
if ( ctx.action == None) {
log.error(s"mgoAdmin> uery action cannot be null!")
Left(new IllegalArgumentException("mgoAdmin> query action cannot be null!"))
}
try {
ctx.action.get match {
/* drop collection */
case DropCollection(collName) => //SingleObservable
val coll = db.getCollection(collName)
coll.drop().toFuture()
/* create collection */
case CreateCollection(collName, None) => //SingleObservable
db.createCollection(collName).toFuture()
case CreateCollection(collName, Some(opt)) => //SingleObservable
db.createCollection(collName, opt.asInstanceOf[CreateCollectionOptions]).toFuture()
/* list collection
case ListCollection(dbName) => //ListConllectionObservable
client.getDatabase(dbName).listCollections().toFuture().asInstanceOf[Future[T]]
*/
/* create view */
case CreateView(viewName, viewOn, pline, None) => //SingleObservable
db.createView(viewName, viewOn, pline).toFuture()
case CreateView(viewName, viewOn, pline, Some(opt)) => //SingleObservable
db.createView(viewName, viewOn, pline, opt.asInstanceOf[CreateViewOptions]).toFuture()
/* create index */
case CreateIndex(key, None) => //SingleObservable
coll.createIndex(key).toFuture().asInstanceOf[Future[Completed]] // asInstanceOf[SingleObservable[Completed]]
case CreateIndex(key, Some(opt)) => //SingleObservable
coll.createIndex(key, opt.asInstanceOf[IndexOptions]).asInstanceOf[Future[Completed]] // asInstanceOf[SingleObservable[Completed]]
/* drop index */
case DropIndexByName(indexName, None) => //SingleObservable
coll.dropIndex(indexName).toFuture()
case DropIndexByName(indexName, Some(opt)) => //SingleObservable
coll.dropIndex(indexName, opt.asInstanceOf[DropIndexOptions]).toFuture()
case DropIndexByKey(key, None) => //SingleObservable
coll.dropIndex(key).toFuture()
case DropIndexByKey(key, Some(opt)) => //SingleObservable
coll.dropIndex(key, opt.asInstanceOf[DropIndexOptions]).toFuture()
case DropAllIndexes(None) => //SingleObservable
coll.dropIndexes().toFuture()
case DropAllIndexes(Some(opt)) => //SingleObservable
coll.dropIndexes(opt.asInstanceOf[DropIndexOptions]).toFuture()
}
}
catch { case e: Exception =>
log.error(s"mgoAdmin> runtime error: ${e.getMessage}")
throw new RuntimeException(s"mgoAdmin> Error: ${e.getMessage}")
}
}
}
object MongoActionStream {
import MGOClasses._
case class StreamingInsert[A](dbName: String,
collName: String,
converter: A => Document,
parallelism: Int = 1
) extends MGOCommands
case class StreamingDelete[A](dbName: String,
collName: String,
toFilter: A => Bson,
parallelism: Int = 1,
justOne: Boolean = false
) extends MGOCommands
case class StreamingUpdate[A](dbName: String,
collName: String,
toFilter: A => Bson,
toUpdate: A => Bson,
parallelism: Int = 1,
justOne: Boolean = false
) extends MGOCommands
case class InsertAction[A](ctx: StreamingInsert[A])(
implicit mongoClient: MongoClient) {
val database = mongoClient.getDatabase(ctx.dbName)
val collection = database.getCollection(ctx.collName)
def performOnRow(implicit ec: ExecutionContext): Flow[A, Document, NotUsed] =
Flow[A].map(ctx.converter)
.mapAsync(ctx.parallelism)(doc => collection.insertOne(doc).toFuture().map(_ => doc))
}
case class UpdateAction[A](ctx: StreamingUpdate[A])(
implicit mongoClient: MongoClient) {
val database = mongoClient.getDatabase(ctx.dbName)
val collection = database.getCollection(ctx.collName)
def performOnRow(implicit ec: ExecutionContext): Flow[A, A, NotUsed] =
if (ctx.justOne) {
Flow[A]
.mapAsync(ctx.parallelism)(a =>
collection.updateOne(ctx.toFilter(a), ctx.toUpdate(a)).toFuture().map(_ => a))
} else
Flow[A]
.mapAsync(ctx.parallelism)(a =>
collection.updateMany(ctx.toFilter(a), ctx.toUpdate(a)).toFuture().map(_ => a))
}
case class DeleteAction[A](ctx: StreamingDelete[A])(
implicit mongoClient: MongoClient) {
val database = mongoClient.getDatabase(ctx.dbName)
val collection = database.getCollection(ctx.collName)
def performOnRow(implicit ec: ExecutionContext): Flow[A, A, NotUsed] =
if (ctx.justOne) {
Flow[A]
.mapAsync(ctx.parallelism)(a =>
collection.deleteOne(ctx.toFilter(a)).toFuture().map(_ => a))
} else
Flow[A]
.mapAsync(ctx.parallelism)(a =>
collection.deleteMany(ctx.toFilter(a)).toFuture().map(_ => a))
}
}
object MGOHelpers {
implicit class DocumentObservable[C](val observable: Observable[Document]) extends ImplicitObservable[Document] {
override val converter: (Document) => String = (doc) => doc.toJson
}
implicit class GenericObservable[C](val observable: Observable[C]) extends ImplicitObservable[C] {
override val converter: (C) => String = (doc) => doc.toString
}
trait ImplicitObservable[C] {
val observable: Observable[C]
val converter: (C) => String
def results(): Seq[C] = Await.result(observable.toFuture(), 10 seconds)
def headResult() = Await.result(observable.head(), 10 seconds)
def printResults(initial: String = ""): Unit = {
if (initial.length > 0) print(initial)
results().foreach(res => println(converter(res)))
}
def printHeadResult(initial: String = ""): Unit = println(s"${initial}${converter(headResult())}")
}
def getResult[T](fut: Future[T], timeOut: Duration = 1 second): T = {
Await.result(fut, timeOut)
}
def getResults[T](fut: Future[Iterable[T]], timeOut: Duration = 1 second): Iterable[T] = {
Await.result(fut, timeOut)
}
import monix.eval.Task
import monix.execution.Scheduler.Implicits.global
final class FutureToTask[A](x: => Future[A]) {
def asTask: Task[A] = Task.deferFuture[A](x)
}
final class TaskToFuture[A](x: => Task[A]) {
def asFuture: Future[A] = x.runToFuture
}
}TestMongoEngine.scala
代码语言：javascript复制import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import org.mongodb.scala._
import scala.collection.JavaConverters._
import com.mongodb.client.model._
import com.datatech.sdp.mongo.engine._
import MGOClasses._
import scala.util._
object TestMongoEngine extends App {
import MGOEngine._
import MGOHelpers._
import MGOCommands._
import MGOAdmins._
// or provide custom MongoClientSettings
val settings: MongoClientSettings = MongoClientSettings.builder()
.applyToClusterSettings(b => b.hosts(List(new ServerAddress("localhost")).asJava))
.build()
implicit val client: MongoClient = MongoClient(settings)
implicit val system = ActorSystem()
implicit val mat = ActorMaterializer()
// implicit val ec = system.dispatcher
val ctx = MGOContext("testdb","po").setCommand(
DropCollection("po"))
import monix.execution.Scheduler.Implicits.global
println(getResult(mgoAdmin(ctx).value.value.runToFuture))
scala.io.StdIn.readLine()
val pic = fileToMGOBlob("/users/tiger/cert/MeTiger.png")
val po1 = Document (
"ponum" -> "po18012301",
"vendor" -> "The smartphone compay",
"podate" -> mgoDate(2017,5,13),
"remarks" -> "urgent, rush order",
"handler" -> pic,
"podtl" -> Seq(
Document("item" -> "sony smartphone", "price" -> 2389.00, "qty" -> 1239, "packing" -> "standard"),
Document("item" -> "ericson smartphone", "price" -> 897.00, "qty" -> 1000, "payterm" -> "30 days")
)
)
val po2 = Document (
"ponum" -> "po18022002",
"vendor" -> "The Samsung compay",
"podate" -> mgoDate(2015,11,6),
"podtl" -> Seq(
Document("item" -> "samsung galaxy s8", "price" -> 2300.00, "qty" -> 100, "packing" -> "standard"),
Document("item" -> "samsung galaxy s7", "price" -> 1897.00, "qty" -> 1000, "payterm" -> "30 days"),
Document("item" -> "apple iphone7", "price" -> 6500.00, "qty" -> 100, "packing" -> "luxury")
)
)
val optInsert = new InsertManyOptions().ordered(true)
val ctxInsert = ctx.setCommand(
Insert(Seq(po1,po2),Some(optInsert))
)
println(getResult(mgoUpdate(ctxInsert).value.value.runToFuture))
scala.io.StdIn.readLine()
case class PO (
ponum: String,
podate: MGODate,
vendor: String,
remarks: Option[String],
podtl: Option[MGOArray],
handler: Option[MGOBlob]
)
def toPO(doc: Document): PO = {
PO(
ponum = doc.getString("ponum"),
podate = doc.getDate("podate"),
vendor = doc.getString("vendor"),
remarks = mgoGetStringOrNone(doc,"remarks"),
podtl = mgoGetArrayOrNone(doc,"podtl"),
handler = mgoGetBlobOrNone(doc,"handler")
)
}
case class PODTL(
item: String,
price: Double,
qty: Int,
packing: Option[String],
payTerm: Option[String]
)
def toPODTL(podtl: Document): PODTL = {
PODTL(
item = podtl.getString("item"),
price = podtl.getDouble("price"),
qty = podtl.getInteger("qty"),
packing = mgoGetStringOrNone(podtl,"packing"),
payTerm = mgoGetStringOrNone(podtl,"payterm")
)
}
def showPO(po: PO) = {
println(s"po number: ${po.ponum}")
println(s"po date: ${mgoDateToString(po.podate,"yyyy-MM-dd")}")
println(s"vendor: ${po.vendor}")
if (po.remarks != None)
println(s"remarks: ${po.remarks.get}")
po.podtl match {
case Some(barr) =>
mgoArrayToDocumentList(barr)
.map { dc => toPODTL(dc)}
.foreach { doc: PODTL =>
print(s"==>Item: ${doc.item} ")
print(s"price: ${doc.price} ")
print(s"qty: ${doc.qty} ")
doc.packing.foreach(pk => print(s"packing: ${pk} "))
doc.payTerm.foreach(pt => print(s"payTerm: ${pt} "))
println("")
}
case _ =>
}
po.handler match {
case Some(blob) =>
val fileName = s"/users/tiger/${po.ponum}.png"
mgoBlobToFile(blob,fileName)
println(s"picture saved to ${fileName}")
case None => println("no picture provided")
}
}
import org.mongodb.scala.model.Projections._
import org.mongodb.scala.model.Filters._
import org.mongodb.scala.model.Sorts._
import org.mongodb.scala.bson.conversions._
import org.mongodb.scala.bson.Document
val ctxFilter = Find(filter=Some(equal("podtl.qty",100)))
val sort: Bson = (descending("ponum"))
val proj: Bson = (and(include("ponum","podate")
,include("vendor"),excludeId()))
val resSort = ResultOptions(FOD_TYPE.FOD_SORT,Some(sort))
val resProj = ResultOptions(FOD_TYPE.FOD_PROJECTION,Some(proj))
val ctxFind = ctx.setCommand(Find(andThen=Seq(resProj,resSort)))
val ctxFindFirst = ctx.setCommand(Find(firstOnly=true))
val ctxFindArrayItem = ctx.setCommand(
Find(filter = Some(equal("podtl.qty",100)))
)
for {
_ <- mgoQuery[List[Document]](ctxFind).value.value.runToFuture.andThen {
case Success(eold) => eold match {
case Right(old) => old match {
case Some(ld) => ld.map(toPO(_)).foreach(showPO)
case None => println(s"Empty document found!")
}
case Left(err) => println(s"Error: ${err.getMessage}")
}
println("-------------------------------")
case Failure(e) => println(e.getMessage)
}
_ <- mgoQuery[PO](ctxFindFirst,Some(toPO _)).value.value.runToFuture.andThen {
case Success(eop) => eop match {
case Right(op) => op match {
case Some(p) => showPO(_)
case None => println(s"Empty document found!")
}
case Left(err) => println(s"Error: ${err.getMessage}")
}
println("-------------------------------")
case Failure(e) => println(e.getMessage)
}
_ <- mgoQuery[List[PO]](ctxFindArrayItem,Some(toPO _)).value.value.runToFuture.andThen {
case Success(eops) => eops match {
case Right(ops) => ops match {
case Some(lp) => lp.foreach(showPO)
case None => println(s"Empty document found!")
}
case Left(err) => println(s"Error: ${err.getMessage}")
}
println("-------------------------------")
case Failure(e) => println(e.getMessage)
}
} yield()
scala.io.StdIn.readLine()
system.terminate()
}
