Looking at the source code of the previous blog again, I found I couldn't understand even myself.Think about returning to the OOP line mode in order to make time for delivery. Look at this code:
(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)
}
}
}
}
Can anyone understand its function from this code?The original author's goal was simple: the front end submitted a picture and product number PID through httprequest, the system read MongoDB to find the number count of the same pid, and then wrote the picture and description, including count, to the database and returned count in reponse.It's my fault to make the implementation of a simple function so complex, probably too poisoned by OOP.This time I want to calm down and re-implement this code in functional programming mode, demonstrating the code refinement and elegance of functional programming.First, let's talk about DBResult[A]: This is a Monad designed to handle types like Future[Either[Option[R]]], which is a more comprehensive type of database operation, but at the same time it is the culprit of the code mess above.Now we can simplify code reuse by implicit conversion s:
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)
}
}
}
}
Add a function to Future[R] with this implicit conversion type for any DBOResult[R].The entire futCount formula can now be simplified to the following:
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
True simplicity.
Somehow, I used Await.result in the middle of this code.It's easy to understand from an OOP perspective that the next program needs the results of the previous one to continue running.In the example above, we need to get the count first, then insert the count into the Document and then save the Document into the database.There is no problem with the logic, but this is a typical line programming mode.In functional programming mode, the result of a phased operation is embedded in Manaud.Monad itself is an operation plan, and only a real operation can produce results.Monad itself is a function component that enables the combination of functions of multiple Monads.Here you can visually describe the combination of Monad functions as a database operation step: count first, then insert. The result of these two steps remains in Monad until the so-called end of the world, that is, the actual operation is completed, so Monad is a typical procedure operation process pipeline.If we write the insert program as addPicture(...): DBOResult[], as follows:
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)
}
Okay, now the whole code is like this:
(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)) } }
Is it easy to understand now?If this seems easier to understand, I suggest you start learning functional programming more now.
Then re-implement the entire project in the same way.The modified source code is as follows:
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)
}
}
}