Akka (41): Http: DBTable-rows streaming-database table row exchange

In the previous discussion, we introduced file exchange through http. Because the content of the file is represented by a bunch of bytes, and the data part of the HTTP message is also byte type, we can read the file directly with Source[ByteString,] and put it into HttpEntity. We also mentioned that if database data exchange is needed, the rows of database tables can be represented by Source[ROW,], but the ROW - > ByteString transformation must be performed first. In the last discussion, we mentioned that this transformation is actually ROW - > Json - > ByteString or the reverse transformation, which is called Marshalling and Unmarshalling in Akka-http. The Marshalling implementation of Akka-http adopts the type-class programming mode. It needs to provide an implicit example of Marshaller[A,B] type in the visual domain for each type of conversion to Json. Akka-http's default Json tool library is Spray-Json, which focuses on case class and provides JsonFormat?(case-class), where the number of parameters representing case class is slightly more complex to use. However, because it is an Akka-http library, it will have certain advantages in the future sustainable development of Akka-http, so we still use it for the following demonstration.

Now let's start writing some code. First, we use a case class to represent the database table row structure, and then use it as a flow element to construct a Source, as follows:

  case class County(id: Int, name: String)
  val source: Source[County, NotUsed] = Source(1 to 5).map { i => County(i, s"Regional Number of Guangdong Province, China #$i") }

First, we design the data download part of the server:

import akka.actor._
import akka.stream._
import akka.stream.scaladsl._
import akka.http.scaladsl.Http
import akka.http.scaladsl.server.Directives._
import akka._
import akka.http.scaladsl.common._
import spray.json.DefaultJsonProtocol
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport


trait MyFormats extends SprayJsonSupport with DefaultJsonProtocol
object Converters extends MyFormats {
  case class County(id: Int, name: String)
  val source: Source[County, NotUsed] = Source(1 to 5).map { i => County(i, s"Regional Number of Guangdong Province, China #$i") }
  implicit val countyFormat = jsonFormat2(County)
}

object HttpDBServer extends App {
  import Converters._

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


  implicit val jsonStreamingSupport = EntityStreamingSupport.json()
    .withParallelMarshalling(parallelism = 8, unordered = false)

  val route =
    path("rows") {
      get {
        complete {
          source
        }
      }
    }

  val (port, host) = (8011,"localhost")

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

}

In the above code, we put source directly into complete(), and then expect this directive to convert Source[County,NotUsed] into Source[ByteString,NotUsed] by Spray-Json through the ToEntity Marshaller [County] class instance, and then into HttpResponse's HTTP Entity. The conversion results can only be confirmed on the client side. We know that Entity.dataBytes in HttpResponse is a Source[ByteString,]. We can turn its Unmarshall into Source[County,], and then use Akka-stream to operate:

     case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) =>
          val futSource = Unmarshal(entity).to[Source[County,NotUsed]]
          futSource.onSuccess {
            case source => source.runForeach(println)
          }
 

The above Unmarshal invokes the following implicit instance of FromEntity Unmarshaller [County]:

  // support for as[Source[T, NotUsed]]
  implicit def sprayJsonSourceReader[T](implicit reader: RootJsonReader[T], support: EntityStreamingSupport): FromEntityUnmarshaller[Source[T, NotUsed]] =
    Unmarshaller.withMaterializer { implicit ec ⇒ implicit mat ⇒ e ⇒
      if (support.supported.matches(e.contentType)) {
        val frames = e.dataBytes.via(support.framingDecoder)
        val unmarshal = sprayJsonByteStringUnmarshaller(reader)(_)
        val unmarshallingFlow =
          if (support.unordered) Flow[ByteString].mapAsyncUnordered(support.parallelism)(unmarshal)
          else Flow[ByteString].mapAsync(support.parallelism)(unmarshal)
        val elements = frames.viaMat(unmarshallingFlow)(Keep.right)
        FastFuture.successful(elements)
      } else FastFuture.failed(Unmarshaller.UnsupportedContentTypeException(support.supported))
    }

This implicit instance is provided by Spray-Jason in prayJsonSupport. scala.
Here is the complete code for this part of the client:

import akka.actor._
import akka.stream._
import akka.stream.scaladsl._
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import scala.util._
import akka._
import akka.http.scaladsl.common._
import spray.json.DefaultJsonProtocol
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import akka.http.scaladsl.unmarshalling.Unmarshal

trait MyFormats extends SprayJsonSupport with DefaultJsonProtocol
object Converters extends MyFormats {
  case class County(id: Int, name: String)
  implicit val countyFormat = jsonFormat2(County)
}

object HttpDBClient extends App {
  import Converters._

  implicit val sys = ActorSystem("ClientSys")
  implicit val mat = ActorMaterializer()
  implicit val ec = sys.dispatcher

  implicit val jsonStreamingSupport: JsonEntityStreamingSupport = EntityStreamingSupport.json()

  def downloadRows(request: HttpRequest) = {
    val futResp = Http(sys).singleRequest(request)
    futResp
      .andThen {
        case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) =>
          val futSource = Unmarshal(entity).to[Source[County,NotUsed]]
          futSource.onSuccess {
            case source => source.runForeach(println)
          }
        case Success(r@HttpResponse(code, _, _, _)) =>
          println(s"download request failed, response code: $code")
          r.discardEntityBytes()
        case Success(_) => println("Unable to download rows!")
        case Failure(err) => println(s"download failed: ${err.getMessage}")

      }
  }
  downloadRows(HttpRequest(HttpMethods.GET,uri = s"http://localhost:8011/rows"))

  scala.io.StdIn.readLine()

  sys.terminate()

}

Above all, we have realized that the client downloads a database table row from the server, and then processes the download data in the way of Akka-stream operation. Then reverse switching, which uploads a table row from the client, requires converting a Source[T,] into a Source[ByteString,] and putting it into the HttpEntity of the HttpRequest. When the server receives the data, it needs to do the reverse transformation, that is, to transfer Request.Entity.dataBytes from Source[ByteString,] to Source[T,]. Akka-http does not provide such powerful automation functions as complete on the client side. We may need to customize and provide implicit instances like ToRequest Marshaller [Source[T,]]. But Markshalling-type-class of Akka-http is a very complex system. If our goal is simply to provide a Source[ByteString,], can we call Spray-Json's function directly to perform the ROW - > Son - > ByteString transformation? As follows:

  import akka.util.ByteString
  import akka.http.scaladsl.model.HttpEntity.limitableByteSource

  val source: Source[County,NotUsed] = Source(1 to 5).map {i => County(i, s"Number of prefectures, cities and counties in Guangxi Zhuang Autonomous Region #$i")}
  def countyToByteString(c: County) = {
    ByteString(c.toJson.toString)
  }
  val flowCountyToByteString : Flow[County,ByteString,NotUsed] = Flow.fromFunction(countyToByteString)

  val rowBytes = limitableByteSource(source via flowCountyToByteString)

  val request = HttpRequest(HttpMethods.POST,uri = s"http://localhost:8011/rows")
  val data = HttpEntity(
    ContentTypes.`application/json`,
    rowBytes
  )

//We use the toJson function directly for County->Json Conversion implementation flowCountyToByteString. toJason yes Spray-Json A function is provided:
package json {

  case class DeserializationException(msg: String, cause: Throwable = null, fieldNames: List[String] = Nil) extends RuntimeException(msg, cause)
  class SerializationException(msg: String) extends RuntimeException(msg)

  private[json] class PimpedAny[T](any: T) {
    def toJson(implicit writer: JsonWriter[T]): JsValue = writer.write(any)
  }

  private[json] class PimpedString(string: String) {
    @deprecated("deprecated in favor of parseJson", "1.2.6")
    def asJson: JsValue = parseJson
    def parseJson: JsValue = JsonParser(string)
  }
}

Assuming that the server receives the data and converts it to a List return in Akka-stream mode, we test the function in the following way:

  def uploadRows(request: HttpRequest, dataEntity: RequestEntity) = {
    val futResp = Http(sys).singleRequest(
      request.copy(entity = dataEntity)
    )
    futResp
      .andThen {
        case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) =>
          entity.dataBytes.map(_.utf8String).runForeach(println)
        case Success(r@HttpResponse(code, _, _, _)) =>
          println(s"Upload request failed, response code: $code")
          r.discardEntityBytes()
        case Success(_) => println("Unable to Upload file!")
        case Failure(err) => println(s"Upload failed: ${err.getMessage}")

      }
  }

The data processing method of the server receiving is as follows:

     post {
        withoutSizeLimit {
          entity(asSourceOf[County]) { source =>
            val futofNames: Future[List[String]] =
              source.runFold(List[String](""))((acc, b) => acc ++ List(b.name))
            complete {
              futofNames
            }
          }
        }
      }

Consider that exceptions may occur during data conversion. Exception handling is required to release backpressure:

  def postExceptionHandler: ExceptionHandler =
    ExceptionHandler {
      case _: RuntimeException =>
        extractRequest { req =>
          req.discardEntityBytes()
          complete((StatusCodes.InternalServerError.intValue,"Upload Failed!"))
        }
    }

      post {
        withoutSizeLimit {
          handleExceptions(postExceptionHandler) {
            entity(asSourceOf[County]) { source =>
              val futofNames: Future[List[String]] =
                source.runFold(List[String](""))((acc, b) => acc ++ List(b.name))
              complete {
                futofNames
              }
            }
          }
        }
      }

The results of the trial run on the client side show that:

  uploadRows(request,data)

["","Number of prefectures, cities and counties in Guangxi Zhuang Autonomous Region #1","Number of prefectures, cities and counties in Guangxi Zhuang Autonomous Region #2","Number of prefectures, cities and counties in Guangxi Zhuang Autonomous Region #3","Number of prefectures, cities and counties in Guangxi Zhuang Autonomous Region #4","Number of prefectures, cities and counties in Guangxi Zhuang Autonomous Region #5"]

It is the result we expect.

The following is a demonstration code for this discussion:

Server:

import akka.actor._
import akka.stream._
import akka.stream.scaladsl._
import akka.http.scaladsl.Http
import akka._
import akka.http.scaladsl.common._
import spray.json.DefaultJsonProtocol
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import scala.concurrent._
import akka.http.scaladsl.server._
import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.model._

trait MyFormats extends SprayJsonSupport with DefaultJsonProtocol
object Converters extends MyFormats {
  case class County(id: Int, name: String)
  val source: Source[County, NotUsed] = Source(1 to 5).map { i => County(i, s"Regional Number of Guangdong Province, China #$i") }
  implicit val countyFormat = jsonFormat2(County)
}

object HttpDBServer extends App {
  import Converters._

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


  implicit val jsonStreamingSupport = EntityStreamingSupport.json()
    .withParallelMarshalling(parallelism = 8, unordered = false)

  def postExceptionHandler: ExceptionHandler =
    ExceptionHandler {
      case _: RuntimeException =>
        extractRequest { req =>
          req.discardEntityBytes()
          complete((StatusCodes.InternalServerError.intValue,"Upload Failed!"))
        }
    }

  val route =
    path("rows") {
      get {
        complete {
          source
        }
      } ~
      post {
        withoutSizeLimit {
          handleExceptions(postExceptionHandler) {
            entity(asSourceOf[County]) { source =>
              val futofNames: Future[List[String]] =
                source.runFold(List[String](""))((acc, b) => acc ++ List(b.name))
              complete {
                futofNames
              }
            }
          }
        }
      }
    }

  val (port, host) = (8011,"localhost")

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

}

Client:

import akka.actor._
import akka.stream._
import akka.stream.scaladsl._
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import scala.util._
import akka._
import akka.http.scaladsl.common._
import spray.json.DefaultJsonProtocol
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import akka.http.scaladsl.unmarshalling._

trait MyFormats extends SprayJsonSupport with DefaultJsonProtocol
object Converters extends MyFormats {
  case class County(id: Int, name: String)
  implicit val countyFormat = jsonFormat2(County)
}

object HttpDBClient extends App {
  import Converters._

  implicit val sys = ActorSystem("ClientSys")
  implicit val mat = ActorMaterializer()
  implicit val ec = sys.dispatcher

  implicit val jsonStreamingSupport: JsonEntityStreamingSupport = EntityStreamingSupport.json()

  def downloadRows(request: HttpRequest) = {
    val futResp = Http(sys).singleRequest(request)
    futResp
      .andThen {
        case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) =>
          val futSource = Unmarshal(entity).to[Source[County,NotUsed]]
          futSource.onSuccess {
            case source => source.runForeach(println)
          }
        case Success(r@HttpResponse(code, _, _, _)) =>
          println(s"download request failed, response code: $code")
          r.discardEntityBytes()
        case Success(_) => println("Unable to download rows!")
        case Failure(err) => println(s"download failed: ${err.getMessage}")

      }
  }
  downloadRows(HttpRequest(HttpMethods.GET,uri = s"http://localhost:8011/rows"))

  
  import akka.util.ByteString
  import akka.http.scaladsl.model.HttpEntity.limitableByteSource

  val source: Source[County,NotUsed] = Source(1 to 5).map {i => County(i, s"Number of prefectures, cities and counties in Guangxi Zhuang Autonomous Region #$i")}
  def countyToByteString(c: County) = {
    ByteString(c.toJson.toString)
  }
  val flowCountyToByteString : Flow[County,ByteString,NotUsed] = Flow.fromFunction(countyToByteString)

  val rowBytes = limitableByteSource(source via flowCountyToByteString)

  val request = HttpRequest(HttpMethods.POST,uri = s"http://localhost:8011/rows")
  val data = HttpEntity(
    ContentTypes.`application/json`,
    rowBytes
  )

  def uploadRows(request: HttpRequest, dataEntity: RequestEntity) = {
    val futResp = Http(sys).singleRequest(
      request.copy(entity = dataEntity)
    )
    futResp
      .andThen {
        case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) =>
          entity.dataBytes.map(_.utf8String).runForeach(println)
        case Success(r@HttpResponse(code, _, _, _)) =>
          println(s"Upload request failed, response code: $code")
          r.discardEntityBytes()
        case Success(_) => println("Unable to Upload file!")
        case Failure(err) => println(s"Upload failed: ${err.getMessage}")

      }
  }

  uploadRows(request,data)

  scala.io.StdIn.readLine()

  sys.terminate()

}

Tags: Scala JSON Database Programming

Posted on Sat, 29 Dec 2018 09:36:07 -0800 by deadparrot