Akka(36): Http:Client-side-Api,Client-Connections

  def runService(request: HttpRequest, rentity: RequestEntity) = {
   val futResp = for {
     entity <- Future.successful(rentity)
     resp <- Http(sys).singleRequest(
       request.copy(entity = rentity)
     )
   } yield resp

   futResp
    .andThen {
      case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) =>
        entity.dataBytes.map(_.utf8String).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}")

    }
  }

在以上这些例子里实际同样会碰着Connection-Level-Api所遇的的题目,这是因为获取的线程池内的线程依旧有数的,只好解决因为request速率超出response速率所导致的request积压。近日最管用的主意如故经过接纳一个queue来暂存request后再逐个处理:

 

  /**
   * Triggers an orderly shutdown of all host connections pools currently maintained by the [[akka.actor.ActorSystem]].
   * The returned future is completed when all pools that were live at the time of this method call
   * have completed their shutdown process.
   *
   * If existing pool client flows are re-used or new ones materialized concurrently with or after this
   * method call the respective connection pools will be restarted and not contribute to the returned future.
   */
  def shutdownAllConnectionPools(): Future[Unit] = {
    val shutdownCompletedPromise = Promise[Done]()
    poolMasterActorRef ! ShutdownAll(shutdownCompletedPromise)
    shutdownCompletedPromise.future.map(_ ⇒ ())(system.dispatcher)
  }

从展现的结果可以得出runService函数中的entity.dataBytes.map(_.utf8String)已经把ByteString转换成了String,也就是说服务器端发送的Entity里的数量是ByteString。

 

   futResp
    .andThen {
      case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) =>
        val futEnt = entity.dataBytes.map(_.utf8String).runForeach(println)
             Await.result(futEnt, Duration.Inf) // throws if binding fails
             println("End of stream!!!")
      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}")

    }

 

object HttpEntity {
  implicit def apply(string: String): HttpEntity.Strict = apply(ContentTypes.`text/plain(UTF-8)`, string)
  implicit def apply(bytes: Array[Byte]): HttpEntity.Strict = apply(ContentTypes.`application/octet-stream`, bytes)
  implicit def apply(data: ByteString): HttpEntity.Strict = apply(ContentTypes.`application/octet-stream`, data)
  def apply(contentType: ContentType.NonBinary, string: String): HttpEntity.Strict =
    if (string.isEmpty) empty(contentType) else apply(contentType, ByteString(string.getBytes(contentType.charset.nioCharset)))
  def apply(contentType: ContentType, bytes: Array[Byte]): HttpEntity.Strict =
    if (bytes.length == 0) empty(contentType) else apply(contentType, ByteString(bytes))
  def apply(contentType: ContentType, data: ByteString): HttpEntity.Strict =
    if (data.isEmpty) empty(contentType) else HttpEntity.Strict(contentType, data)

  def apply(contentType: ContentType, contentLength: Long, data: Source[ByteString, Any]): UniversalEntity =
    if (contentLength == 0) empty(contentType) else HttpEntity.Default(contentType, contentLength, data)
  def apply(contentType: ContentType, data: Source[ByteString, Any]): HttpEntity.Chunked =
    HttpEntity.Chunked.fromData(contentType, data)
...

Akka-http客户端api还有一种实用的Host-Level-Client-Side-Api格局。这套api能自动针对各种端点维护一个连接池(connection-pool),用户只需对连接池举办安排。系统遵照连接池配置活动珍贵池内线程的生、死、动、停。akka-http.host-connection-pool配置中max-connections,max-open-requests,pipelining-limit等控制着connection、在途request的多寡,需要特别注意。针对某个端点的连接池是由此Http().cachedHostConnectionPool(endPoint)获取的。同样,获取的也是一个client-flow实例。因为系统活动珍重着线程池,所以client-flow实例可以自由引用,无论调用次数与调用时间间隔。cachedHostConnectionPool()函数定义如下:

 

 //close connection by cancelling response entity
  resp.entity.dataBytes.runWith(Sink.cancelled)
 //close connection by receiving response with close header
  Http().bindAndHandleSync(
    { req ⇒ HttpResponse(headers = headers.Connection("close") :: Nil) },
    "akka.io",
    80)(mat)

咱俩知晓Akka-http是基于Akka-Stream的,具备Reactive-Stream功效特色。下面大家就示范一下咋样进展stream的上传下载。首先定制一个Source:

 

再示范一下在客户端用Source上传数据。服务端代码:

俺们见到:这些函数实现了对Server端地址host+port的设定,重临的结果类型是Flow[HttpRequest,HttpResponse,Future[OutgoingConnection]]。这些Flow代表将输入的HttpRequest转换成输出的HttpResponse。这多少个转换过程包括了与Server之间的Request,Response信息沟通。下边我们试着用这一个Flow来向Server端发送request,并拿到response:

 val numbers = Source.fromIterator(() =>
    Iterator.continually(Random.nextInt()))
    .map(n => ByteString(s"$n\n"))
  //make conform to withoutSizeLimit constrain
  val source = limitableByteSource(numbers)

  val bytes = HttpEntity(
    ContentTypes.`application/octet-stream`,
    // transform each number to a chunk of bytes
    source.take(10000)
  )
  val postRandom = HttpRequest(HttpMethods.POST,uri = s"http://localhost:8011/random")
  runService(postRandom,bytes)
    .andThen{case _ => sys.terminate()}

 
 Akka-http的客户端Api应该是以HttpRequest操作为主轴的网上消息交流情势编程工具。我们了解:Akka-http是搭建在Akka-stream之上的。所以,Akka-http在客户端构建与服务器的连续通道也得以用Akka-stream的Flow来表示。那些Flow可以由此调用Http.outgoingConnection来赢得:

  val reqRandom = HttpRequest(uri = s"http://localhost:8011/random")
    runService(reqRandom,HttpEntity.Empty)
     .andThen{case _ => sys.terminate()}

函数重返结果类型:Flow[(HttpRequest,T),(Try[HttpResponse],T),HostConnectionPool]。因为线程池内的线程是异步构建request和接收response的,而回到response的各类未必遵照发送request的逐条,所以需要一个tuple2的T类型标示request与再次回到的response举行匹配。线程池会依照idle-timeout自动终止,也可以手动通过HostConnectionPool.shutDown()实现:

 

 

  val numbers = Source.fromIterator(() =>
    Iterator.continually(Random.nextInt()))
    .map(n => ByteString(s"$n\n"))
  //make conform to withoutSizeLimit constrain
  val source = limitableByteSource(numbers)
    val QueueSize = 10
    // This idea came initially from this blog post:
    // http://kazuhiro.github.io/scala/akka/akka-http/akka-streams/2016/01/31/connection-pooling-with-akka-http-and-source-queue.html
    val poolClientFlow = Http().cachedHostConnectionPool[Promise[HttpResponse]]("akka.io")
    val queue =
      Source.queue[(HttpRequest, Promise[HttpResponse])](QueueSize, OverflowStrategy.dropNew)
        .via(poolClientFlow)
        .toMat(Sink.foreach({
          case ((Success(resp), p)) => p.success(resp)
          case ((Failure(e), p))    => p.failure(e)
        }))(Keep.left)
        .run()

    def queueRequest(request: HttpRequest): Future[HttpResponse] = {
      val responsePromise = Promise[HttpResponse]()
      queue.offer(request -> responsePromise).flatMap {
        case QueueOfferResult.Enqueued    => responsePromise.future
        case QueueOfferResult.Dropped     => Future.failed(new RuntimeException("Queue overflowed. Try again later."))
        case QueueOfferResult.Failure(ex) => Future.failed(ex)
        case QueueOfferResult.QueueClosed => Future.failed(new RuntimeException("Queue was closed (pool shut down) while running the request. Try again later."))
      }
    }

    val responseFuture: Future[HttpResponse] = queueRequest(HttpRequest(uri = "/"))
    responseFuture.andThen {
      case Success(resp) => println(s"got response: ${resp.status.intValue()}")
      case Failure(err) => println(s"request failed: ${err.getMessage}")
    }.andThen {case _ => sys.terminate()}
  val reqText = HttpRequest(uri = s"http://localhost:8011/text")
  runService(reqText,HttpEntity.Empty)
    .andThen{case _ => sys.terminate()}

也足以通过Http().shutdownAllConnectionPools()一遍性终止ActorSystem内所无线程池:

 

下面是本次Akka-http-client-side-connection琢磨的示范源代码:

服务端也是用HttpEntity来装载这些Source然后经过HttpRequest传给客户端的:

  val connFlow: Flow[HttpRequest,HttpResponse,Future[Http.OutgoingConnection]] =
    Http().outgoingConnection("akka.io")

  def sendHttpRequest(req: HttpRequest) = {
    Source.single(req)
      .via(connFlow)
      .runWith(Sink.head)
  }

  sendHttpRequest(HttpRequest(uri="/"))
      .andThen{
        case Success(resp) => println(s"got response: ${resp.status.intValue()}")
        case Failure(err) => println(s"request failed: ${err.getMessage}")
      }
     .andThen {case _ => sys.terminate()}

 

  /**
   * Creates a [[akka.stream.scaladsl.Flow]] representing a prospective HTTP client connection to the given endpoint.
   * Every materialization of the produced flow will attempt to establish a new outgoing connection.
   *
   * To configure additional settings for requests made using this method,
   * use the `akka.http.client` config section or pass in a [[akka.http.scaladsl.settings.ClientConnectionSettings]] explicitly.
   */
  def outgoingConnection(host: String, port: Int = 80,
                         localAddress: Option[InetSocketAddress] = None,
                         settings:     ClientConnectionSettings  = ClientConnectionSettings(system),
                         log:          LoggingAdapter            = system.log): Flow[HttpRequest, HttpResponse, Future[OutgoingConnection]] =
    _outgoingConnection(host, port, settings.withLocalAddressOverride(localAddress), ConnectionContext.noEncryption(), ClientTransport.TCP, log)

很显然,HttpEntity可以分两大类,一种是Strict类型的,它的data是ByteString。另一种是UniversalEntity类型,它的多少dataBytes是Source[ByteString,Any]。无论怎么样最终在线上的或者ByteString。HttpEntity的ContentType注明了传输数据格式,有:

 

俺们可以看到放进entity里的数额是ByteString。

  val pooledFlow: Flow[(HttpRequest,Int),(Try[HttpResponse],Int),Http.HostConnectionPool] =
      Http().cachedHostConnectionPool[Int](host="akka.io",port=80)

  def sendPoolRequest(req: HttpRequest, marker: Int) = {
    Source.single(req -> marker)
      .via(pooledFlow)
      .runWith(Sink.head)
  }

  sendPoolRequest(HttpRequest(uri="/"), 1)
    .andThen{
      case Success((tryResp, mk)) =>
        tryResp match {
          case Success(resp) => println(s"got response: ${resp.status.intValue()}")
          case Failure(err) => println(s"request failed: ${err.getMessage}")
        }
      case Failure(err) => println(s"request failed: ${err.getMessage}")
    }
    .andThen {case _ => sys.terminate()}

大家再试着发送一些数量給服务端,然后让服务端把结果通过response
entity再次回到来:

 

大家只需要对这些函数传入RequestEntity就可以了然再次来到Response里Entity的众多细节了。首先我们渴求服务端发送一个纯字符串Hello
World。服务端代码如下:

import akka.actor._
import akka.http.javadsl.{HostConnectionPool, OutgoingConnection}
import akka.stream._
import akka.stream.scaladsl._
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._

import scala.concurrent._
import scala.util._

object ClientApiDemo extends App {
  implicit val sys = ActorSystem("ClientSys")
  implicit val mat = ActorMaterializer()
  implicit val ec = sys.dispatcher
/*
  val connFlow: Flow[HttpRequest,HttpResponse,Future[Http.OutgoingConnection]] =
    Http().outgoingConnection("www.sina.com")

  def sendHttpRequest(req: HttpRequest) = {
    Source.single(req)
      .via(connFlow)
      .runWith(Sink.head)
  }

  sendHttpRequest(HttpRequest(uri="/"))
      .andThen{
        case Success(resp) =>
          //close connection by cancelling response entity
          resp.entity.dataBytes.runWith(Sink.cancelled)
          println(s"got response: ${resp.status.intValue()}")
        case Failure(err) => println(s"request failed: ${err.getMessage}")
      }
  //   .andThen {case _ => sys.terminate()}


 //close connection by receiving response with close header
  Http().bindAndHandleSync(
    { req ⇒ HttpResponse(headers = headers.Connection("close") :: Nil) },
    "akka.io",
    80)(mat)

  val pooledFlow: Flow[(HttpRequest,Int),(Try[HttpResponse],Int),Http.HostConnectionPool] =
      Http().cachedHostConnectionPool[Int](host="akka.io",port=80)

  def sendPoolRequest(req: HttpRequest, marker: Int) = {
    Source.single(req -> marker)
      .via(pooledFlow)
      .runWith(Sink.head)
  }

  sendPoolRequest(HttpRequest(uri="/"), 1)
    .andThen{
      case Success((tryResp, mk)) =>
        tryResp match {
          case Success(resp) => println(s"got response: ${resp.status.intValue()}")
          case Failure(err) => println(s"request failed: ${err.getMessage}")
        }
      case Failure(err) => println(s"request failed: ${err.getMessage}")
    }
    .andThen {case _ => sys.terminate()}
*/

    val QueueSize = 10
    // This idea came initially from this blog post:
    // http://kazuhiro.github.io/scala/akka/akka-http/akka-streams/2016/01/31/connection-pooling-with-akka-http-and-source-queue.html
    val poolClientFlow = Http().cachedHostConnectionPool[Promise[HttpResponse]]("akka.io")
    val queue =
      Source.queue[(HttpRequest, Promise[HttpResponse])](QueueSize, OverflowStrategy.dropNew)
        .via(poolClientFlow)
        .toMat(Sink.foreach({
          case ((Success(resp), p)) => p.success(resp)
          case ((Failure(e), p))    => p.failure(e)
        }))(Keep.left)
        .run()

    def queueRequest(request: HttpRequest): Future[HttpResponse] = {
      val responsePromise = Promise[HttpResponse]()
      queue.offer(request -> responsePromise).flatMap {
        case QueueOfferResult.Enqueued    => responsePromise.future
        case QueueOfferResult.Dropped     => Future.failed(new RuntimeException("Queue overflowed. Try again later."))
        case QueueOfferResult.Failure(ex) => Future.failed(ex)
        case QueueOfferResult.QueueClosed => Future.failed(new RuntimeException("Queue was closed (pool shut down) while running the request. Try again later."))
      }
    }

    val responseFuture: Future[HttpResponse] = queueRequest(HttpRequest(uri = "/"))
    responseFuture.andThen {
      case Success(resp) => println(s"got response: ${resp.status.intValue()}")
      case Failure(err) => println(s"request failed: ${err.getMessage}")
    }.andThen {case _ => sys.terminate()}

}

服务端:

俺们用cachedHostConnectionPool获取一个client-flow实例:

 

下边的这种形式就是所谓Connection-Level-Client-Side-Api。那种格局可以让用户有更大程度的自由度控制connection的构建、使用及在connection上发送request的点子。一般来讲,当重返response的entity被全然消耗后系统会活动close
connection,这套api还提供了一些手动方法可以在有需要的动静出手动举办connection
close,如下:

 

  /**
   * Same as [[#cachedHostConnectionPool]] but for encrypted (HTTPS) connections.
   *
   * If an explicit [[ConnectionContext]] is given then it rather than the configured default [[ConnectionContext]] will be used
   * for encryption on the connections.
   *
   * To configure additional settings for the pool (and requests made using it),
   * use the `akka.http.host-connection-pool` config section or pass in a [[ConnectionPoolSettings]] explicitly.
   */
  def cachedHostConnectionPoolHttps[T](host: String, port: Int = 443,
                                       connectionContext: HttpsConnectionContext = defaultClientHttpsContext,
                                       settings:          ConnectionPoolSettings = defaultConnectionPoolSettings,
                                       log:               LoggingAdapter         = system.log)(implicit fm: Materializer): Flow[(HttpRequest, T), (Try[HttpResponse], T), HostConnectionPool] = {
    val cps = ConnectionPoolSetup(settings, connectionContext, log)
    val setup = HostConnectionPoolSetup(host, port, cps)
    cachedHostConnectionPool(setup)
  }
  path("random") {
      get {
        complete(
          HttpEntity(
            ContentTypes.`application/octet-stream`,
            // transform each number to a chunk of bytes
            source.take(10000)
          )
        )
      } ~
  /**
   * Represents a connection pool to a specific target host and pool configuration.
   */
  final case class HostConnectionPool private[http] (setup: HostConnectionPoolSetup)(
    private[http] val gateway: PoolGateway) { // enable test access

    /**
     * Asynchronously triggers the shutdown of the host connection pool.
     *
     * The produced [[scala.concurrent.Future]] is fulfilled when the shutdown has been completed.
     */
    def shutdown()(implicit ec: ExecutionContextExecutor): Future[Done] = gateway.shutdown()

    private[http] def toJava = new akka.http.javadsl.HostConnectionPool {
      override def setup = HostConnectionPool.this.setup
      override def shutdown(executor: ExecutionContextExecutor): CompletionStage[Done] = HostConnectionPool.this.shutdown()(executor).toJava
    }
  }
import akka.actor._
import akka.stream._
import akka.stream.scaladsl._
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpEntity.limitableByteSource
import akka.http.scaladsl.model._

import scala.concurrent.duration._
import akka.util.ByteString

import scala.concurrent._
import scala.util._

object ClientEntity extends App {

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

  def runService(request: HttpRequest, rentity: RequestEntity) = {
   val futResp = for {
     entity <- Future.successful(rentity)
     resp <- Http(sys).singleRequest(
       request.copy(entity = rentity)
     )
   } yield resp

   futResp
    .andThen {
      case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) =>
        val futEnt = entity.dataBytes.map(_.utf8String).runForeach(println)
             Await.result(futEnt, Duration.Inf) // throws if binding fails
             println("End of stream!!!")
      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}")

    }
  }

  val reqText = HttpRequest(uri = s"http://localhost:8011/text")
//  runService(reqText,HttpEntity.Empty)
//    .andThen{case _ => sys.terminate()}

  val postText = HttpRequest(HttpMethods.POST,uri = s"http://localhost:8011/text")
  val uploadText = HttpEntity(
    ContentTypes.`text/plain(UTF-8)`,
    // transform each number to a chunk of bytes
    ByteString("hello world again")
  )
//  runService(postText,uploadText)
//    .andThen{case _ => sys.terminate()}

  val reqRandom = HttpRequest(uri = s"http://localhost:8011/random")
 //   runService(reqRandom,HttpEntity.Empty)
 //    .andThen{case _ => sys.terminate()}

  val numbers = Source.fromIterator(() =>
    Iterator.continually(Random.nextInt()))
    .map(n => ByteString(s"$n\n"))
  //make conform to withoutSizeLimit constrain
  val source = limitableByteSource(numbers)

  val bytes = HttpEntity(
    ContentTypes.`application/octet-stream`,
    // transform each number to a chunk of bytes
    source.take(10000)
  )
  val postRandom = HttpRequest(HttpMethods.POST,uri = s"http://localhost:8011/random")
  runService(postRandom,bytes)
    .andThen{case _ => sys.terminate()}


}

 

用上边的艺术调用:

 

 

Flow[(HttpRequest,T),(Try[HttpResponse],T),HostConnectionPool]后就可以举办输入HttpRequest到HttpResponse的变换处理。如上边的事例:

 

 

大家在客户端或者用runService(Service)来分析传过来的entity。由于接到一个巨型的Source,所以需要修改一下收受格局代码:

客户端上传数据范例:

       post {
          withoutSizeLimit {
            extractDataBytes { bytes =>
              val data = bytes.runFold(ByteString())(_ ++ _)
              onComplete(data) { t =>
                complete(t)
              }
            }
          }
        }

 

 

 
我们说过Akka-http是一个好的体系融为一体工具,集成是经过数据互换情势实现的。Http是个在网上传输和接到的科班商事。所以,在接纳Akka-http以前,可能我们仍然需要把Http形式的网上数据互换细节了然清楚。数据交流双方是透过Http音信类型Request和Response来兑现的。在Akka-http中对应的是HttpRequest和HttpResponse。这六个门类都存有HttpEntity类型来装载需要交流的多少。首先,无论怎么着数据在线上的表现形式肯定是一串bytes。所以,数据交流四头Request,Response中的Entity也必须是以bytes来抒发的。在Akka-http里大家把需要传输的数量转换成ByteString,通过网络发送給接收端、接收端再把接受音信Entity中的ByteString转换成目标项目标数码。这五个转移过程就是Akka-http的马尔斯halling和Unmarshalling过程了。我们先从HttpEntity的构建函数来打听它的定义:

  val postText = HttpRequest(HttpMethods.POST,uri = s"http://localhost:8011/text")
  val uploadText = HttpEntity(
    ContentTypes.`text/plain(UTF-8)`,
    // transform each number to a chunk of bytes
    ByteString("hello world again")
  )
  runService(postText,uploadText)
    .andThen{case _ => sys.terminate()}

从下边商讨大家询问了在马尔斯hal,Unmarshal下层只是ByteString的操作和转移。下面是这次琢磨示范源代码:

object ContentTypes {
  val `application/json` = ContentType(MediaTypes.`application/json`)
  val `application/octet-stream` = ContentType(MediaTypes.`application/octet-stream`)
  val `text/plain(UTF-8)` = MediaTypes.`text/plain` withCharset HttpCharsets.`UTF-8`
  val `text/html(UTF-8)` = MediaTypes.`text/html` withCharset HttpCharsets.`UTF-8`
  val `text/xml(UTF-8)` = MediaTypes.`text/xml` withCharset HttpCharsets.`UTF-8`
  val `text/csv(UTF-8)` = MediaTypes.`text/csv` withCharset HttpCharsets.`UTF-8`

  // used for explicitly suppressing the rendering of Content-Type headers on requests and responses
  val NoContentType = ContentType(MediaTypes.NoMediaType)
}

瞩目:ContentType只是一种备注,不影响线上多少表达形式,线上的数量永远是ByteString。可是,其中的application/octet-stream类型代表数量必须是Source[ByteString,Any]。我们上面就透过客户端的事例来了然HttpEntity。下边是一个客户端测试函数:

虽然complete(“Hello
World!”)有些迷糊,可是相应complete做了些字符串到ByteString的转移。我们可以从上边这些run瑟维斯(Service)(Service)函数拿到验证。下边是这么些例子的调用:

 

 

    } ~ path("text") {
      get {
        complete("Hello World!")
      } ~
        post {
          withoutSizeLimit {
            extractDataBytes { bytes =>
              val data = bytes.runFold(ByteString())(_ ++ _)
              onComplete(data) { t =>
                complete(t)
              }
            }
          }
        }

我们见到服务端对request
entity的操作是以ByteString举办的。客户端上传一串字符的request如下:

 

import akka.actor._
import akka.stream._
import akka.stream.scaladsl._
import akka.http.scaladsl.Http
import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.model._
import akka.util.ByteString
import akka.http.scaladsl.model.HttpEntity._

import scala.util.Random

object ServerEntity extends App {

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


  val numbers = Source.fromIterator(() =>
    Iterator.continually(Random.nextInt()))
    .map(n => ByteString(s"$n\n"))
  //make conform to withoutSizeLimit constrain
  val source = limitableByteSource(numbers)



  val route =
    path("random") {
      get {
        withoutSizeLimit {
          complete(
            HttpEntity(
              ContentTypes.`application/octet-stream`,
              // transform each number to a chunk of bytes
              source.take(1000))
          )
        }
      } ~
        post {
          withoutSizeLimit {
            extractDataBytes { bytes =>
              val data = bytes.runFold(ByteString())(_ ++ _)
              onComplete(data) { t =>
                complete(t)
              }
            }
          }
        }
    } ~ path("text") {
      get {
        complete("Hello World!")
      } ~
        post {
          withoutSizeLimit {
            extractDataBytes { bytes =>
              val data = bytes.runFold(ByteString())(_ ++ _)
              onComplete(data) { t =>
                complete(t)
              }
            }
          }
        }
    }


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

}

 

客户端:

 } ~ path("text") {
      get {
        complete("Hello World!")
      } ~