ByteString或然反方向的更换,在后边一篇啄磨里大家介绍了通过http进行文件的交流澳门永利备用网址

地方那些Unmarshal调用了上边那个FromEntityUnmarshaller[County]隐式实例:

幸亏我们希望的结果。

 

 

  uploadRows(request,data)

["","广西壮族自治区地市县编号 #1","广西壮族自治区地市县编号 #2","广西壮族自治区地市县编号 #3","广西壮族自治区地市县编号 #4","广西壮族自治区地市县编号 #5"]

在客户端试运行重返结果突显:

在地点的代码里大家直接把source放进了complete(),然后希望以此directive能通过ToEntity马尔斯haller[County]类实例用Spray-Json把Source[County,NotUsed]转换成Source[ByteString,NotUsed]接下来放入HttpResponse的HttpEntity里。转换结果不得不在客户端拿到验证。大家领略HttpResponse里的Entity.dataBytes就是一个Source[ByteString,_],我们得以把它Unmarshall成Source[County,_],然后用Akka-stream来操作:

 

 

 

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"中国广东省地区编号 #$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())

}

服务端接收数据处理格局如下:

 

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

幸好大家期待的结果。

设想到在数额转换的经过中或然会出现极度。须求分外处理办法来刑释解教backpressure:

在客户端试运行重返结果突显:

 

  case class County(id: Int, name: String)
  val source: Source[County, NotUsed] = Source(1 to 5).map { i => County(i, s"中国广东省地区编号 #$i") }

 

 

 

服务端接收数据处理措施如下:

     case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) =>
          val futSource = Unmarshal(entity).to[Source[County,NotUsed]]
          futSource.onSuccess {
            case source => source.runForeach(println)
          }
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"广西壮族自治区地市县编号 #$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()

}

 

大家先规划服务端的多少下载部分:

以上大家已经达成了客户端从服务端下载一段数据库表行,然后以Akka-stream的操作办法来拍卖下载数据。那么反向沟通即从客户端上传一段表行的话就须求把一个Source[T,_]转换成Source[ByteString,_]然后放进HttpRequest的HttpEntity里。服务端收到数量后又要开展反向的转换即把Request.Entity.dataBytes从Source[ByteString,_]转回Source[T,_]。Akka-http在客户端从未提供像complete那样的强有力的自动化成效。我们兴许须要自定义并提供像ToRequest马尔斯haller[Source[T,_]]这么的隐式实例。但Akka-http的马尔斯halling-type-class是个卓殊复杂的系统。如若大家的目的是概括提供一个Source[ByteString,_],大家是否可以从来调用Spray-Json的函数来拓展ROW->Son->ByteString转换呢?如下:

下边就让我们开始写些代码吧。首先,我们用一个case
class代表数据库表行结构,然后用它作为流成分来创设一个Source,如下:

地点那几个Unmarshal调用了上面那些FromEntityUnmarshaller[County]隐式实例:

 

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

这几个隐式实例是由Spray-杰森提供的,在SprayJsonSupport.scala里。
上面是这一部分客户端的完好代码: 

 

 

服务端:

  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"广西壮族自治区地市县编号 #$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
  )

我们直接用toJson函数进行County->Json转换实现了flowCountyToByteString。toJason是Spray-Json提供的一个函数:
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)
  }
}

 

 

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

      }
  }

 

 

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

}

 

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

}

 

客户端:

服务端:

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"中国广东省地区编号 #$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())

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

在下边的代码里大家直接把source放进了complete(),然后希望以此directive能通过ToEntity马尔斯haller[County]类实例用Spray-Json把Source[County,NotUsed]转换成Source[ByteString,NotUsed]接下来放入HttpResponse的HttpEntity里。转换结果不得不在客户端得到验证。大家了然HttpResponse里的Entity.dataBytes就是一个Source[ByteString,_],大家能够把它Unmarshall成Source[County,_],然后用Akka-stream来操作:

 

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"广西壮族自治区地市县编号 #$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()

}

 

  case class County(id: Int, name: String)
  val source: Source[County, NotUsed] = Source(1 to 5).map { i => County(i, s"中国广东省地区编号 #$i") }

上述我们早已已毕了客户端从服务端下载一段数据库表行,然后以Akka-stream的操作办法来拍卖下载数据。那么反向交流即从客户端上传一段表行的话就必要把一个Source[T,_]转换成Source[ByteString,_]接下来放进HttpRequest的HttpEntity里。服务端收到多少后又要拓展反向的更换即把Request.Entity.dataBytes从Source[ByteString,_]转回Source[T,_]。Akka-http在客户端从未提供像complete那样的无敌的自动化成效。大家或许需求自定义并提供像ToRequestMarshaller[Source[T,_]]如此的隐式实例。但Akka-http的马尔斯halling-type-class是个万分复杂的种类。即便大家的目标是大约提供一个Source[ByteString,_],大家是或不是能够直接调用Spray-Json的函数来进展ROW->Son->ByteString转换呢?如下:

 

 

 

客户端:

上面就让大家起始写些代码吧。首先,大家用一个case
class代表数据库表行结构,然后用它看成流成分来创设一个Source,如下:

下边是本次探讨的以身作则代码:

假如服务端收到数量后以Akka-stream格局再转换成一个List重回,大家用下边的措施来测试功用:

 

 

  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)

["","广西壮族自治区地市县编号 #1","广西壮族自治区地市县编号 #2","广西壮族自治区地市县编号 #3","广西壮族自治区地市县编号 #4","广西壮族自治区地市县编号 #5"]

 

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"中国广东省地区编号 #$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())

}

 

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

假如服务端收到多少后以Akka-stream格局再转换成一个List重返,大家用上面的法子来测试功效:

大家先规划服务端的多寡下载部分:

 

  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"广西壮族自治区地市县编号 #$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
  )

我们直接用toJson函数进行County->Json转换实现了flowCountyToByteString。toJason是Spray-Json提供的一个函数:
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)
  }
}
  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
              }
            }
          }
        }
      }

 

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"中国广东省地区编号 #$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())

}

 

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

本条隐式实例是由Spray-杰森提供的,在SprayJsonSupport.scala里。
上面是那有的客户端的全体代码: 

 
在前面一篇探究里大家介绍了经过http进行文件的交流。因为文件内容是以一堆bytes来表示的,而http音信的多寡部分也是byte类型的,所以大家可以直接用Source[ByteString,_]来读取文件然后放进HttpEntity中。大家还论及:如果必要进行数据库数据互换的话,可以用Source[ROW,_]来代表库表行,但首先必须开展ROW
->
ByteString的转移。在上期议论大家关系过那种转移其实是ROW->Json->ByteString可能反方向的转移,在Akka-http里称之为马尔斯halling和Unmarshalling。Akka-http的马尔斯halling完结利用了type-class编程方式,需求为每一系列型与Json的转移在可视域内提供Marshaller[A,B]品种的隐式实例。Akka-http默许的Json工具库是Spray-Json,重视case
class,而且要提供JsonFormat?(case-class),其中?代表case
class的参数个数,用起来略显复杂。可是因为是Akka-http的配套库,在后日Akka-http的不止提升中有所自然的优势,所以我们依旧用它来拓展下边的演示。

上面是这次商量的示范代码:

设想到在数量转换的进度中只怕会出现非常。需要极度处理方式来刑满释放backpressure:

 
在前头一篇钻探里我们介绍了通过http进行文件的置换。因为文件内容是以一堆bytes来表示的,而http消息的数量部分也是byte类型的,所以我们得以一贯用Source[ByteString,_]来读取文件然后放进HttpEntity中。大家还关乎:若是须要开展数据库数据沟通的话,可以用Source[ROW,_]来表示库表行,但首先必须开展ROW
->
ByteString的更换。在上期议论大家提到过这种转移其实是ROW->Json->ByteString大概反方向的转移,在Akka-http里称之为马尔斯halling和Unmarshalling。Akka-http的马尔斯halling已毕选取了type-class编程情势,须要为各种档次与Json的转移在可视域内提供马尔斯haller[A,B]花色的隐式实例。Akka-http暗中同意的Json工具库是Spray-Json,重视case
class,而且要提供JsonFormat?(case-class),其中?代表case
class的参数个数,用起来略显复杂。但是因为是Akka-http的配套库,在今后Akka-http的频频前进中颇具一定的优势,所以大家仍然用它来开展上面的示范。