restapi（6）- do it the functional way, 重溫函數式編程

  再次看了看上篇博客的源代碼，發現連本身都看不懂了。想是爲了趕時間交貨不知不覺又回到OOP行令模式了，看看下面這段代碼：

       (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) } } } }

有人能從這段代碼裏理解它的功能嗎？原本做者的目的很簡單：前端經過httprequest提交了一張圖片及產品編號pid、系統讀取MongoDB查找相同pid的數量count，而後將圖片和描述包括count寫入數據庫並在reponse裏返回count。把一個簡單功能的實現搞的這麼複雜都是個人錯，可能受OOP荼毒太深。此次但願靜下心來用函數式編程模式把這段代碼重新實現一次，示範一下函數式編程的代碼精煉和高雅特色。首先介紹一下DBResult[A]這個類型：這是一個Monad，爲了應付Future[Either[Option[R]]]這樣的類型而設計的，是一個表現數據庫操做比較全面的類型，但同時它又是形成上面這段代碼混亂的元兇。如今咱們能夠用隱式轉換implicit conversion方式進行代碼簡化重用：

  import monix.execution.Scheduler.Implicits.global
  implicit class DBResultToFuture(dbr: DBOResult[_]){ def toFuture[R] = { dbr.value.value.runToFuture.map { eor => eor match { case Right(or) => or match { case Some(r) => r.asInstanceOf[R] case None => throw new RuntimeException("Operation produced None result!") } case Left(err) => throw new RuntimeException(err) } } } }

 用這個隱式轉換類型爲任何DBOResult[R]增長一個函數toFuture[R]。如今整個futCount算式能夠簡化成下面這樣：

          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 } } futCount:Future[Int]=repository.count(pid).toFuture

真正的簡單易明。

不知怎麼搞的，我盡然在這段代碼中間使用了Await.result。從OOP角度分析這很容易理解，下一段程序須要上一段程序的結果來繼續運行。在上面的例子裏咱們須要先獲取count而後把count塞進Document再把Document存入數據庫。邏輯思路上沒問題，不過這樣的作法是典型的行令式編程模式。在函數式編程模式裏，階段性的運算結果是在包嵌在Monad中的。Monad自己只是一個運算計劃，只有真正運算時才能獲取結果。Monad自己是函數組件，能夠實現多個Monad的函數組合。在這裏能夠形象的把Monad函數組合描述爲數據庫操做步驟：先count、再insert，這兩個步驟產生的結果仍是留在Monad裏的，直到所謂的世界末日，即實際運算完成後才取出，因此Monad是一種典型的程序運算流程管道。假如咱們再把insert這段程序寫成addPicture(...): DBOResult[_], 以下：

 def addPicuture(pid: String,seqno: Int, optDesc: Option[String] ,optWid:Option[Int],optHgh:Option[Int], bytes: Array[Byte]):DBOResult[Completed] ={ var doc = Document( "pid" -> pid, "seqno" -> seqno, "pic" -> bytes ) if (optDesc != None) doc = doc + ("desc" -> optDesc.get) if (optWid != None) doc = doc + ("desc" -> optWid.get) if (optHgh != None) doc = doc + ("desc" -> optHgh.get) repository.insert(doc) }

好了，如今整篇代碼變成了下面這樣：

       (post &  parameters('pid,'desc.?,'width.as[Int].?,'heigth.as[Int].?)) { (pid, optDesc, optWid, optHgh) => withoutSizeLimit { decodeRequest { extractDataBytes { bytes => val futBytes = bytes.runFold(ByteString()) { case (hd, bs) => hd ++ bs } val futSeqno = for { cnt <- repository.count(pid).toFuture barr <- futBytes _ <- addPicuture(pid, cnt, optDesc, optWid, optHgh, barr.toArray).toFuture } yield cnt complete(futSeqno.map(_.toString)) } }  

如今是否是變得簡單易明瞭？若是你覺着這樣看起來更加容易理解，那麼我建議你如今開始多點接觸瞭解函數式編程。

接着用一樣方式把整個項目從新實現一次。修改後的源代碼以下：

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[R](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[R](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[R](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)
    }

  }
  import monix.execution.Scheduler.Implicits.global
  implicit class DBResultToFuture(dbr: DBOResult[_]){
      def toFuture[R] = {
        dbr.value.value.runToFuture.map {
          eor =>
            eor match {
              case Right(or) => or match {
                case Some(r) => r.asInstanceOf[R]
                case None => throw new RuntimeException("Operation produced None result!")
              }
              case Left(err) => throw new RuntimeException(err)
            }
        }
      }
  }

}

MongoRoute.scala

package com.datatech.rest.mongoimport akka.http.scaladsl.server.Directivesimport 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.ActorMaterializerimport com.datatech.sdp.result.DBOResult._import org.mongodb.scala.model.Filters._import com.datatech.sdp.mongo.engine.MGOClasses._import monix.execution.CancelableFutureimport akka.util._import akka.http.scaladsl.model._import akka.http.scaladsl.coding.Gzipimport akka.stream.scaladsl._import MongoModels.WebPicimport 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) =>          withoutSizeLimit {            decodeRequest {              extractDataBytes { bytes =>                val futBytes = bytes.runFold(ByteString()) { case (hd, bs) =>                  hd ++ bs                }                val futSeqno = for {                  cnt <- repository.count(pid).toFuture[Int]                  barr <- futBytes                  _ <- addPicuture(pid, cnt, optDesc, optWid, optHgh, barr.toArray).toFuture[Completed]                } yield cnt                complete(futSeqno.map(_.toString))              }            }          } ~            (get & parameters('pid, 'seqno.as[Int].?, 'width.as[Int].?, 'height.as[Int].?)) {              (pid, optSeq, optWid, optHght) =>                if (optSeq == None) {                  val futRows = repository.query(equal("pid", pid)).toFuture                  complete(futureToJson(futRows))                } else {                  val futPicRow = repository.getOnePicture(pid, optSeq.get).toFuture[WebPic]                  onComplete(futPicRow) {                    case Success(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 Failure(err) => complete(err)                  }                }            }        }      } ~        pathPrefix("blob") {          (get & parameter('filter)) { filter =>            val filtr = Document(filter)            val futOptPic: CancelableFuture[Option[MGOBlob]] = repository.getOneDocument(filtr).toFuture            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 futbytes = bytes.runFold(ByteString()) { case (hd, bs) =>                      hd ++ bs                    }                    val futmsg:Future[Completed] = for {                      bytes <- futbytes                      doc = Document(bson) + ("photo" -> bytes.toArray)                      c <- repository.insert(doc).toFuture[Completed]                    } yield c                    complete(futmsg.map(_.toString))                  }                }              }            }        } ~        (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:Future[Seq[WebPic]] = dbor.toFuture[Seq[WebPic]]            complete(futureToJson(futRows))          }        } ~ post {        entity(as[String]) { json =>          val extractedEntity: M = fromJson[M](json)          val doc: Document = extractedEntity.to          val futmsg = repository.insert(doc).toFuture[Completed]          complete(futmsg.map(_.toString))        }      } ~ (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).toFuture            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:Future[UpdateResult] = dbou.toFuture[UpdateResult]          complete(futureToJson(futmsg.map(_.toString)))        }      } ~ (delete & parameters('filter, 'many.as[Boolean].?)) { (filter,many) =>        val bson = Document(filter)        val futmsg:Future[DeleteResult] = repository.delete(bson).toFuture[DeleteResult]        complete(futureToJson(futmsg.map(_.toString)))      }    }    def addPicuture(pid: String,seqno: Int, optDesc: Option[String]                         ,optWid:Option[Int],optHgh:Option[Int],                         bytes: Array[Byte]):DBOResult[Completed] ={      var doc = Document(        "pid" -> pid,        "seqno" -> seqno,        "pic" -> bytes      )      if (optDesc != None)        doc = doc + ("desc" -> optDesc.get)      if (optWid != None)        doc = doc + ("desc" -> optWid.get)      if (optHgh != None)        doc = doc + ("desc" -> optHgh.get)      repository.insert(doc)    }  }}