restapi（6）- do it the functional way, 重溫函數式編程
- 2019 年 10 月 3 日
- 筆記
再次看了看上篇部落格的源程式碼，發現連自己都看不懂了。想是為了趕時間交貨不知不覺又回到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.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 akka.stream.scaladsl._
import 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) =>
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)
}
}
}
