restapi（5）- rest-mongo 應用實例：分佈式圖片管理系統之一，rest 服務

  最近有同事提起想把網頁上的圖片存在MongoDB裏，我十分贊同。比起把圖片以文件形式存放在硬盤子目錄的方式，MongoDB有太多的優點。首先，MongoDB是分佈式數據庫，圖片能夠跨服務器存儲。在一個集羣環境裏經過複製集、分片等技術能夠提升圖片讀取速度、實現數據的高可用和安全性。再就是對大量的圖片可用規範的記錄管理方式來進行處理，甚至在一個大流量環境裏還能夠用集羣節點負載平衡方式來助力圖片的存取。

我想了想看有沒有辦法讓這個圖片管理系統盡用分佈式集羣軟件的能力。MongoDB是一個分佈式數據庫，在一個集羣內任何節點均可以存取，也就是說在集羣全部節點上都部署統一的rest-mongo，這樣客戶端能夠用不一樣的ip地址來訪問不一樣的節點提交圖片存取請求。假設某一個節點接待比別的節點更多的客戶端，那麼咱們能夠把圖片存取的過程放到其它比較空閒的節點上去運行，即所謂負載均衡了。看來這個系統須要MongoDB,rest-mongo和akka-cluster這幾個組件。

咱們先從前端需求開始：頁面上每一個商品有n個圖片，客戶端提出存入系統請求時提供商品編號、描述、默認尺寸及圖片。對一個商品提出n個存寫請求，同一個商品編號，系統對每張圖片自動產生序號並在httprespose中返回給客戶端。客戶端取圖片時提供商品編號，系統先把這個商品的全部圖片序號返還客戶端，客戶端再按序號一張一張索取圖片，並指定輸出圖片的伸縮尺寸。

這篇咱們先跟着前幾篇的內容把有關圖片存取的rest服務實現了。在上篇rest-mongo的基礎上，針對新的系統需求作一些針對性的修改應該就好了。

首先是Model部分，以下：

case class WebPic( pid: String, seqno: Int, desc: Option[String], width: Option[Int], heigth: Option[Int], pic: Option[MGOBlob] ) extends ModelBase[Document] { self =>
    override def to: Document = { var doc = Document( "pid" -> self.pid, "seqno" -> self.seqno ) if (self.desc != None) doc = doc + ("desc" -> self.desc.get) if (self.width != None) doc = doc + ("width" -> self.width.get) if (self.heigth != None) doc = doc + ("heigth" -> self.heigth.get) if (self.pic != None) doc = doc + ("photo" -> self.pic.get) doc } } object WebPic { def fromDocument: Document => WebPic = doc => { WebPic( pid = doc.getString("pid"), seqno = doc.getInteger("seqno"), desc = mgoGetStringOrNone(doc,"desc"), width = mgoGetIntOrNone(doc,"width").asInstanceOf[Option[Int]], heigth = mgoGetIntOrNone(doc,"heigth").asInstanceOf[Option[Int]], pic = None ) } }

width,height字段是客戶端提供的默認寬高尺寸。若是客戶在請求圖片時沒有提供就用數據庫裏客戶端在提交存儲時提供的默認寬高。

在repo裏還要增長一個count功能，提供一個pid, 返回在該pid名下存寫的圖片數量：

 import org.mongodb.scala.model.Filters._ def count(pid: String):DBOResult[Int] = { val ctxCount = MGOContext(dbName = db,collName=coll) .setActionType(MGO_ACTION_TYPE.MGO_QUERY) .setCommand(Count(filter=Some(equal("pid",pid)))) mgoQuery[Int](ctxCount,None) }

返回類型是DBResult[Int]。還要加一個讀取第一條WebPic記錄的功能：

    def getOnePicture(pid: String, seqno: Int): DBOResult[R] = { val ctxFind = MGOContext(dbName = db, collName = coll) .setActionType(MGO_ACTION_TYPE.MGO_QUERY) .setCommand(Find(filter = Some(and(equal("pid",pid),equal("seqno",seqno))), firstOnly = true)) mgoQuery[R](ctxFind, converter) }

注意這個函數返回的是DBOResult[R]類型。這是由於咱們須要把整條記錄讀出來，特別是width,height字段，方便在用戶沒有指定寬高時提供默認值。由於涉及到具體的字段名稱，因此要在讀出Document時作一個WebPic轉換：

    def getOneDocument(filtr: Bson): DBOResult[Document] = { val ctxFind = MGOContext(dbName = db,collName=coll) .setActionType(MGO_ACTION_TYPE.MGO_QUERY) .setCommand(Find(filter = Some(filtr),firstOnly = true)) mgoQuery[Document](ctxFind,None) } def getOnePicture(pid: String, seqno: Int): DBOResult[R] = { val ctxFind = MGOContext(dbName = db, collName = coll) .setActionType(MGO_ACTION_TYPE.MGO_QUERY) .setCommand(Find(filter = Some(and(equal("pid",pid),equal("seqno",seqno))), firstOnly = true)) mgoQuery[R](ctxFind, converter) }

要用getOnPicture, getOnDocument是通用的。在編譯時沒法識別width,height。

好了，下面是Route部分的修改。先從用戶提交圖片存儲請求開始，用戶可能用下面這樣格式的url來請求：

(post &  parameters('pid,'desc.?,'width.as[Int].?,'heigth.as[Int].?)) 如：

http://example.com:50081/public/gms/pictures?pid=apple&width=128  圖片放在HttpRequest的Entity裏面。

咱們須要先獲取apple的數量seqno、把信息存入數據庫而後返回這個seqno：

     pathPrefix("pictures") { (post &  parameters('pid,'desc.?,'width.as[Int].?,'heigth.as[Int].?)) { (pid, optDesc, optWid, optHgh) => val futCount: Future[Int] = repository.count(pid).value.value.runToFuture.map { eoi => eoi match { case Right(oi) => oi match { case Some(i) => i case None => -1 } case Left(err) => -1 } } val count: Int = Await.result(futCount, 2 seconds) var doc = Document( "pid" -> pid, "seqno" -> count ) if (optDesc != None) doc = doc + ("desc" -> optDesc.get) if (optWid != None) doc = doc + ("desc" -> optWid.get) if (optHgh != None) doc = doc + ("desc" -> optHgh.get) withoutSizeLimit { decodeRequest { extractDataBytes { bytes => val fut = bytes.runFold(ByteString()) { case (hd, bs) => hd ++ bs } onComplete(fut) { case Success(b) => doc = doc + ("pic" -> b.toArray) val futmsg: Future[String] = repository.insert(doc).value.value.runToFuture.map { eoc => eoc match { case Right(oc) => oc match { case Some(c) => count.toString // c.toString()
                            case None => "insert may not complete!" } case Left(err) => err.getMessage } } complete(futmsg) case Failure(err) => complete(err) } } } }

注意，在Response裏的返回結果必定是ByteString類型的。

圖片讀取請求分兩步：先提供pid獲取一個不含圖片的記錄清單（注意Model裏WebPic的fromDocument函數裏pic=None），返還用戶，如：http://example.com:50081/public/pms/pictures?pid=apple

用戶獲得清單裏的seqno後組裝成完整url：http://example.com:50081/public/pms/pictures?pid=apple&seqno=2&height=64

系統讀取圖片並按用戶關於寬高要求或數據庫裏默認寬高數據輸出圖片：

        (get & parameters('pid, 'seqno.as[Int].?,'width.as[Int].?,'height.as[Int].?)) { (pid, optSeq, optWid,optHght) =>
            if (optSeq == None) { dbor = repository.query(equal("pid", pid)) val futRows = dbor.value.value.runToFuture.map { eolr => eolr match { case Right(olr) => olr match { case Some(lr) => lr case None => Seq[M]() } case Left(_) => Seq[M]() } } complete(futureToJson(futRows)) } else { val futOptPicRow: CancelableFuture[Option[WebPic]] = repository.getOnePicture(pid,optSeq.get) .value.value.runToFuture.map { eorow => eorow match { case Right(orow) => orow match { case Some(row) =>
                        if (row == null) None else Some(row.asInstanceOf[WebPic]) case None => None } case Left(_) => None } } onComplete(futOptPicRow) { case Success(optRow) => optRow match { case Some(row) => val width = if(optWid == None) row.width.getOrElse(128) else optWid.getOrElse(128) val height = if(optHght == None) row.heigth.getOrElse(128) else optHght.getOrElse(128) if (row.pic != None) { withoutSizeLimit { encodeResponseWith(Gzip) { complete( HttpEntity( ContentTypes.`application/octet-stream`, ByteArrayToSource(Imaging.setImageSize(row.pic.get.getData, width, height) )) ) } } } else complete(StatusCodes.NotFound) case None => complete(StatusCodes.NotFound) } case Failure(err) => complete(err) } } }

最後咱們把這個Route組裝在main route裏：

  implicit val webPicDao = new MongoRepo[WebPic]("testdb","pms", WebPic.fromDocument) ... pathPrefix("public") { (pathPrefix("crud")) { new MongoRoute[Person]("person")(personDao) .route ~
            new MongoRoute[Photo]("photo")(picDao) .route } ~
        new MongoRoute[WebPic]("pms")(webPicDao).route }

下面是本次示範的源代碼：

MongoModel.scala

package com.datatech.rest.mongo
import org.mongodb.scala._
import com.datatech.sdp.mongo.engine._
import MGOClasses._

object MongoModels {

  case class Person(
                   userid: String = "",
                   name: String = "",
                   age: Option[Int] = None,
                   dob: Option[MGODate] = None,
                   address: Option[String] = None
                   ) extends ModelBase[Document] {
    import org.mongodb.scala.bson._

    override def to: Document = {
      var doc = Document(
      "userid" -> this.userid,
      "name" -> this.name)


      if (this.age != None)
        doc = doc + ("age" -> this.age.get)

      if (this.dob != None)
        doc = doc + ("dob" -> this.dob.get)

      if (this.address != None)
        doc = doc + ("address" -> this.address.getOrElse(""))

      doc
    }

  }
  object Person {
    val fromDocument: Document => Person = doc => {
      val keyset = doc.keySet
      Person(
        userid = doc.getString("userid"),
        name = doc.getString("name"),
        age = mgoGetIntOrNone(doc,"age").asInstanceOf[Option[Int]],

        dob =  {if (keyset.contains("dob"))
          Some(doc.getDate("dob"))
        else None },

        address =  mgoGetStringOrNone(doc,"address")
      )
    }
  }

  case class Photo (
                     id: String,
                     loc: Option[String],
                     size: Option[Int],
                     credt: Option[MGODate],
                     photo: Option[MGOBlob]
                   ) extends ModelBase[Document] {
    override def to: Document = {
      var doc = Document("id" -> this.id)
      if (loc != None)
        doc = doc + ("loc" -> this.loc.get)
      if (size != None)
        doc = doc + ("size" -> this.size.get)
      if (credt != None)
        doc = doc + ("credt" -> this.credt.get)
      if (photo != None)
        doc = doc + ("photo" -> this.photo.get)
      doc
    }
  }

  object Photo {
    def fromDocument: Document => Photo = doc => {
      Photo(
        id = doc.getString("id"),
        loc = mgoGetStringOrNone(doc,"loc"),
        size = mgoGetIntOrNone(doc,"size").asInstanceOf[Option[Int]],
        credt = mgoGetDateOrNone(doc,"credt"),
        photo = mgoGetBlobOrNone(doc, "photo")
      )
    }
  }

  case class WebPic(
                     pid: String,
                     seqno: Int,
                     desc: Option[String],
                     width: Option[Int],
                     heigth: Option[Int],
                     pic: Option[MGOBlob]
                   ) extends ModelBase[Document] { self =>
    override def to: Document = {
      var doc = Document(
        "pid" -> self.pid,
        "seqno" -> self.seqno
      )
      if (self.desc != None)
        doc = doc + ("desc" -> self.desc.get)
      if (self.width != None)
        doc = doc + ("width" -> self.width.get)
      if (self.heigth != None)
        doc = doc + ("heigth" -> self.heigth.get)
      if (self.pic != None)
        doc = doc + ("photo" -> self.pic.get)
      doc
    }
  }
  object WebPic {
    def fromDocument: Document => WebPic = doc => {
      WebPic(
        pid = doc.getString("pid"),
        seqno = doc.getInteger("seqno"),
        desc = mgoGetStringOrNone(doc,"desc"),
        width = mgoGetIntOrNone(doc,"width").asInstanceOf[Option[Int]],
        heigth = mgoGetIntOrNone(doc,"heigth").asInstanceOf[Option[Int]],
        pic = None
      )
    }
  }
}

MongoRepo.scala

package com.datatech.rest.mongo
import org.mongodb.scala._
import org.bson.conversions.Bson
import org.mongodb.scala.result._
import com.datatech.sdp.mongo.engine._
import MGOClasses._
import MGOEngine._
import MGOCommands._
import com.datatech.sdp.result.DBOResult.DBOResult

object MongoRepo {

   class MongoRepo[R](db:String, coll: String, converter: Option[Document => R])(implicit client: MongoClient) {
    def getAll(next:Option[String],sort:Option[String],fields:Option[String],top:Option[Int]): DBOResult[Seq[R]] = {
      var res = Seq[ResultOptions]()
      next.foreach {b => res = res :+ ResultOptions(FOD_TYPE.FOD_FILTER,Some(Document(b)))}
      sort.foreach {b => res = res :+ ResultOptions(FOD_TYPE.FOD_SORT,Some(Document(b)))}
      fields.foreach {b => res = res :+ ResultOptions(FOD_TYPE.FOD_PROJECTION,Some(Document(b)))}
      top.foreach {b => res = res :+ ResultOptions(FOD_TYPE.FOD_LIMIT,None,b)}

      val ctxFind = MGOContext(dbName = db,collName=coll)
        .setActionType(MGO_ACTION_TYPE.MGO_QUERY)
        .setCommand(Find(andThen = res))
      mgoQuery[Seq[R]](ctxFind,converter)
    }

     def query(filtr: Bson, next:Option[String]=None,sort:Option[String]=None,fields:Option[String]=None,top:Option[Int]=None): DBOResult[Seq[R]] = {
       var res = Seq[ResultOptions]()
       next.foreach {b => res = res :+ ResultOptions(FOD_TYPE.FOD_FILTER,Some(Document(b)))}
       sort.foreach {b => res = res :+ ResultOptions(FOD_TYPE.FOD_SORT,Some(Document(b)))}
       fields.foreach {b => res = res :+ ResultOptions(FOD_TYPE.FOD_PROJECTION,Some(Document(b)))}
       top.foreach {b => res = res :+ ResultOptions(FOD_TYPE.FOD_LIMIT,None,b)}
       val ctxFind = MGOContext(dbName = db,collName=coll)
         .setActionType(MGO_ACTION_TYPE.MGO_QUERY)
         .setCommand(Find(filter = Some(filtr),andThen = res))
       mgoQuery[Seq[R]](ctxFind,converter)
    }

   import org.mongodb.scala.model.Filters._
    def count(pid: String):DBOResult[Int] = {
      val ctxCount = MGOContext(dbName = db,collName=coll)
        .setActionType(MGO_ACTION_TYPE.MGO_QUERY)
        .setCommand(Count(filter=Some(equal("pid",pid))))
      mgoQuery[Int](ctxCount,None)
    }

    def getOneDocument(filtr: Bson): DBOResult[Document] = {
          val ctxFind = MGOContext(dbName = db,collName=coll)
         .setActionType(MGO_ACTION_TYPE.MGO_QUERY)
         .setCommand(Find(filter = Some(filtr),firstOnly = true))
       mgoQuery[Document](ctxFind,None)
    }
    def getOnePicture(pid: String, seqno: Int): DBOResult[R] = {
      val ctxFind = MGOContext(dbName = db, collName = coll)
        .setActionType(MGO_ACTION_TYPE.MGO_QUERY)
        .setCommand(Find(filter = Some(and(equal("pid",pid),equal("seqno",seqno))), firstOnly = true))
      mgoQuery[R](ctxFind, converter)
    }
    def insert(doc: Document): DBOResult[Completed] = {
      val ctxInsert = MGOContext(dbName = db,collName=coll)
        .setActionType(MGO_ACTION_TYPE.MGO_UPDATE)
        .setCommand(Insert(Seq(doc)))
      mgoUpdate[Completed](ctxInsert)
    }

    def delete(filter: Bson): DBOResult[DeleteResult] = {
      val ctxDelete = MGOContext(dbName = db,collName=coll)
        .setActionType(MGO_ACTION_TYPE.MGO_UPDATE)
        .setCommand(Delete(filter))
      mgoUpdate[DeleteResult](ctxDelete)
    }

    def update(filter: Bson, update: Bson, many: Boolean): DBOResult[UpdateResult] = {
      val ctxUpdate = MGOContext(dbName = db,collName=coll)
        .setActionType(MGO_ACTION_TYPE.MGO_UPDATE)
        .setCommand(Update(filter,update,None,!many))
      mgoUpdate[UpdateResult](ctxUpdate)
    }

    def replace(filter: Bson, row: Document): DBOResult[UpdateResult] = {
       val ctxUpdate = MGOContext(dbName = db,collName=coll)
         .setActionType(MGO_ACTION_TYPE.MGO_UPDATE)
         .setCommand(Replace(filter,row))
       mgoUpdate[UpdateResult](ctxUpdate)
    }

  }

}

MongoRoute.scala

package com.datatech.rest.mongo
import akka.http.scaladsl.server.Directives
import com.datatech.sdp.file._

import scala.util._
import org.mongodb.scala._
import com.datatech.sdp.file.Streaming._
import org.mongodb.scala.result._
import MongoRepo._
import akka.stream.ActorMaterializer
import com.datatech.sdp.result.DBOResult._
import org.mongodb.scala.model.Filters._
import com.datatech.sdp.mongo.engine.MGOClasses._
import monix.execution.CancelableFuture
import akka.util._
import akka.http.scaladsl.model._
import akka.http.scaladsl.coding.Gzip
import com.datatech.rest.mongo.MongoModels.WebPic

import scala.concurrent._
import scala.concurrent.duration._
object MongoRoute {
  class MongoRoute[M <: ModelBase[Document]](val pathName: String)(repository: MongoRepo[M])(
    implicit c: MongoClient, m: Manifest[M], mat: ActorMaterializer) extends Directives with JsonConverter {
    import monix.execution.Scheduler.Implicits.global
    var dbor: DBOResult[Seq[M]] = _
    var dbou: DBOResult[UpdateResult] = _
    val route = pathPrefix(pathName) {
      pathPrefix("pictures") {
        (post &  parameters('pid,'desc.?,'width.as[Int].?,'heigth.as[Int].?)) { (pid, optDesc, optWid, optHgh) =>
          val futCount: Future[Int] = repository.count(pid).value.value.runToFuture.map {
            eoi =>
              eoi match {
                case Right(oi) => oi match {
                  case Some(i) => i
                  case None => -1
                }
                case Left(err) => -1
              }
          }
          val count: Int = Await.result(futCount, 2 seconds)
          var doc = Document(
            "pid" -> pid,
            "seqno" -> count
          )
          if (optDesc != None)
            doc = doc + ("desc" -> optDesc.get)
          if (optWid != None)
            doc = doc + ("desc" -> optWid.get)
          if (optHgh != None)
            doc = doc + ("desc" -> optHgh.get)

          withoutSizeLimit {
            decodeRequest {
              extractDataBytes { bytes =>
                val fut = bytes.runFold(ByteString()) { case (hd, bs) =>
                  hd ++ bs
                }
                onComplete(fut) {
                  case Success(b) =>
                    doc = doc + ("pic" -> b.toArray)
                    val futmsg: Future[String] = repository.insert(doc).value.value.runToFuture.map {
                      eoc =>
                        eoc match {
                          case Right(oc) => oc match {
                            case Some(c) => count.toString //   c.toString()
                            case None => "insert may not complete!"
                          }
                          case Left(err) => err.getMessage
                        }
                    }
                    complete(futmsg)
                  case Failure(err) => complete(err)
                }
              }
            }
          }
        } ~
        (get & parameters('pid, 'seqno.as[Int].?,'width.as[Int].?,'height.as[Int].?)) {
          (pid, optSeq, optWid,optHght) =>
            if (optSeq == None) {
              dbor = repository.query(equal("pid", pid))
              val futRows = dbor.value.value.runToFuture.map {
                eolr =>
                  eolr match {
                    case Right(olr) => olr match {
                      case Some(lr) => lr
                      case None => Seq[M]()
                    }
                    case Left(_) => Seq[M]()
                  }
              }
              complete(futureToJson(futRows))
            } else {
              val futOptPicRow: CancelableFuture[Option[WebPic]] = repository.getOnePicture(pid,optSeq.get)
                .value.value.runToFuture.map {
                eorow =>
                  eorow match {
                    case Right(orow) => orow match {
                      case Some(row) =>
                        if (row == null) None
                        else Some(row.asInstanceOf[WebPic])
                      case None => None
                    }
                    case Left(_) => None
                  }
              }
              onComplete(futOptPicRow) {
                case Success(optRow) => optRow match {
                  case Some(row) =>
                    val width  = if(optWid == None) row.width.getOrElse(128) else optWid.getOrElse(128)
                    val height = if(optHght == None) row.heigth.getOrElse(128) else optHght.getOrElse(128)
                    if (row.pic != None) {

                      withoutSizeLimit {
                        encodeResponseWith(Gzip) {
                          complete(
                            HttpEntity(
                              ContentTypes.`application/octet-stream`,
                              ByteArrayToSource(Imaging.setImageSize(row.pic.get.getData, width, height)
                              ))
                          )
                        }
                      }
                    } else complete(StatusCodes.NotFound)
                  case None => complete(StatusCodes.NotFound)
                }
                case Failure(err) => complete(err)
              }
            }
          }
      } ~
      pathPrefix("blob") {
        (get & parameter('filter)) { filter =>
          val filtr = Document(filter)
          val futOptPic: CancelableFuture[Option[MGOBlob]] = repository.getOneDocument(filtr).value.value.runToFuture.map {
            eodoc =>
              eodoc match {
                case Right(odoc) => odoc match {
                  case Some(doc) =>
                    if (doc == null) None
                    else mgoGetBlobOrNone(doc, "photo")
                  case None => None
                }
                case Left(_) => None
              }
          }
          onComplete(futOptPic) {
            case Success(optBlob) => optBlob match {
              case Some(blob) =>
                withoutSizeLimit {
                  encodeResponseWith(Gzip) {
                    complete(
                      HttpEntity(
                        ContentTypes.`application/octet-stream`,
                        ByteArrayToSource(blob.getData)
                      )
                    )
                  }
                }
              case None => complete(StatusCodes.NotFound)
            }
            case Failure(err) => complete(err)
          }
        } ~
        (post &  parameter('bson)) { bson =>
          val bdoc = Document(bson)
          withoutSizeLimit {
            decodeRequest {
              extractDataBytes { bytes =>
                val fut = bytes.runFold(ByteString()) { case (hd, bs) =>
                  hd ++ bs
                }
                onComplete(fut) {
                  case Success(b) =>
                    val doc = bdoc + ("photo" -> b.toArray)
                    val futmsg = repository.insert(doc).value.value.runToFuture.map {
                      eoc =>
                        eoc match {
                          case Right(oc) => oc match {
                            case Some(c) => c.toString()
                            case None => "insert may not complete!"
                          }
                          case Left(err) => err.getMessage
                        }
                    }
                    complete(futmsg)
                  case Failure(err) => complete(err)
                }
              }
            }
          }
        }
      } ~
      (get & parameters('filter.?,'fields.?,'sort.?,'top.as[Int].?,'next.?)) {
        (filter,fields,sort,top,next) => {
        dbor = {
          filter match {
            case Some(fltr) => repository.query(Document(fltr),next,sort,fields,top)
            case None => repository.getAll(next,sort,fields,top)
          }
        }
        val futRows = dbor.value.value.runToFuture.map {
          eolr =>
            eolr match {
              case Right(olr) => olr match {
                case Some(lr) => lr
                case None => Seq[M]()
              }
              case Left(_) => Seq[M]()
            }
        }
        complete(futureToJson(futRows))
       }
      } ~ post {
        entity(as[String]) { json =>
          val extractedEntity: M = fromJson[M](json)
          val doc: Document = extractedEntity.to
          val futmsg = repository.insert(doc).value.value.runToFuture.map {
            eoc =>
              eoc match {
                case Right(oc) => oc match {
                  case Some(c) => c.toString()
                  case None => "insert may not complete!"
                }
                case Left(err) => err.getMessage
              }
          }

          complete(futmsg)
        }
      } ~ (put & parameter('filter,'set.?, 'many.as[Boolean].?)) { (filter, set, many) =>
        val bson = Document(filter)
        if (set == None) {
          entity(as[String]) { json =>
            val extractedEntity: M = fromJson[M](json)
            val doc: Document = extractedEntity.to
            val futmsg = repository.replace(bson, doc).value.value.runToFuture.map {
              eoc =>
                eoc match {
                  case Right(oc) => oc match {
                    case Some(d) => s"${d.getMatchedCount} matched rows, ${d.getModifiedCount} rows updated."
                    case None => "update may not complete!"
                  }
                  case Left(err) => err.getMessage
                }
            }
            complete(futureToJson(futmsg))
          }
        } else {
          set match {
            case Some(u) =>
              val ubson = Document(u)
              dbou = repository.update(bson, ubson, many.getOrElse(true))
            case None =>
              dbou = Left(new IllegalArgumentException("missing set statement for update!"))
          }
          val futmsg = dbou.value.value.runToFuture.map {
            eoc =>
              eoc match {
                case Right(oc) => oc match {
                  case Some(d) => s"${d.getMatchedCount} matched rows, ${d.getModifiedCount} rows updated."
                  case None => "update may not complete!"
                }
                case Left(err) => err.getMessage
              }
          }
          complete(futureToJson(futmsg))
        }
      } ~ (delete & parameters('filter, 'many.as[Boolean].?)) { (filter,many) =>
        val bson = Document(filter)
        val futmsg = repository.delete(bson).value.value.runToFuture.map {
          eoc =>
            eoc match {
              case Right(oc) => oc match {
                case Some(d) => s"${d.getDeletedCount} rows deleted."
                case None => "delete may not complete!"
              }
              case Left(err) => err.getMessage
            }
        }
        complete(futureToJson(futmsg))
      }
    }
  }

}

PMSServer.scala

package com.datatech.rest.mongo

import akka.actor._
import akka.stream._
import akka.http.scaladsl.Http
import akka.http.scaladsl.server.Directives._
import pdi.jwt._
import AuthBase._
import MockUserAuthService._
import org.mongodb.scala._

import scala.collection.JavaConverters._
import MongoModels._
import MongoRepo._
import MongoRoute._


object PMSServer extends App {


  implicit val httpSys = ActorSystem("httpSystem")
  implicit val httpMat = ActorMaterializer()
  implicit val httpEC = httpSys.dispatcher

  val settings: MongoClientSettings = MongoClientSettings.builder()
    .applyToClusterSettings(b => b.hosts(List(new ServerAddress("localhost")).asJava))
    .build()
  implicit val client: MongoClient = MongoClient(settings)
  implicit val personDao = new MongoRepo[Person]("testdb","person", Some(Person.fromDocument))
  implicit val picDao = new MongoRepo[Photo]("testdb","photo", None)
  implicit val webPicDao = new MongoRepo[WebPic]("testdb","pms", WebPic.fromDocument)
  implicit val authenticator = new AuthBase()
    .withAlgorithm(JwtAlgorithm.HS256)
    .withSecretKey("OpenSesame")
    .withUserFunc(getValidUser)

  val route =
    path("auth") {
      authenticateBasic(realm = "auth", authenticator.getUserInfo) { userinfo =>
        post { complete(authenticator.issueJwt(userinfo))}
      }
    } ~
      pathPrefix("private") {
        authenticateOAuth2(realm = "private", authenticator.authenticateToken) { validToken =>
          FileRoute(validToken)
            .route
          // ~ ...
        }
      } ~
      pathPrefix("public") {
        (pathPrefix("crud")) {
          new MongoRoute[Person]("person")(personDao)
            .route ~
            new MongoRoute[Photo]("photo")(picDao)
              .route
        } ~
        new MongoRoute[WebPic]("pms")(webPicDao).route
      }

  val (port, host) = (50081,"192.168.11.189")

  val bindingFuture = Http().bindAndHandle(route,host,port)

  println(s"Server running at $host $port. Press any key to exit ...")

  scala.io.StdIn.readLine()


  bindingFuture.flatMap(_.unbind())
    .onComplete(_ => httpSys.terminate())


}

imaging.scala

package com.datatech.sdp.file
import javax.imageio.ImageIO
import java.awt.Graphics2D
import java.awt.image.BufferedImage
import java.io.ByteArrayInputStream
import java.io.ByteArrayOutputStream

object Imaging {
  def setImageSize(barr: Array[Byte], wth: Int, hth: Int): Array[Byte] = {
    val input = ImageIO.read(new ByteArrayInputStream(barr))
    val image = new BufferedImage(wth, hth, BufferedImage.TYPE_INT_BGR)
    val g = image.getGraphics.asInstanceOf[Graphics2D]
    g.drawImage(input, 0, 0, wth, hth, null) //畫圖
    g.dispose()
    image.flush()
    val barros = new ByteArrayOutputStream()
    ImageIO.write(image, "jpg", barros)
    barr
  }
}