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

阿新 • • 發佈：2019-08-22

最近有同事提起想把網頁上的圖片存在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
  }
}

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

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

RabbitMQ學習總結（5）——釋出和訂閱例項詳解

訊息佇列入門（四）ActiveMQ的應用例項

深度相機---（5）深度相機應用

JavaWeb學習（七）HttpServletResponse基本應用——生成驗證碼圖片並提交到伺服器（6）

計算機網路（5）-傳輸層--TCP傳輸：可靠資料傳輸、流量控制、3次握手過程

帶你深入AI（5）- 自然語言處理領域：RNN LSTM GRU

glib學習筆記之二（續）——GLib核心應用支援：The Main Event Loop

golang（5）使用beego 開發 api server 和前端同學拆分開發，使用swagger

量化分析（5）——Python應用tushare資料計算單資產CAPM例項

restapi（4）- rest-mongo : MongoDB資料庫前端的httpserver

網絡構建入門技術（5）——子網劃分應用

springCloud（5）：Eureka的元數據與Eureka Server的rest端點

第9章應用層（5）_文件傳輸協議FTP

例項學習ansible系列（5）常用模組之copy

SparkStreaming（5）：例項-處理socket源資料

SparkStreaming（5）：例項-SparkStreaming處理本地或者HDFS檔案

第9章應用層（5）_檔案傳輸協議FTP

Redis系列-生產應用篇-分散式鎖（5）-單程序Redis分散式鎖的Java實現（Redisson使用與底層實現）-原子鎖類

「Odoo 基礎教程系列」第六篇——從 Todo 應用開始（5）

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

相關推薦