リクエスト/レスポンスエンティティのストリーミングにおける性質

import java.io.File
import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Framing
import akka.stream.scaladsl.FileIO
import akka.http.scaladsl.model._

implicit val system = ActorSystem()
implicit val dispatcher = system.dispatcher
implicit val materializer = ActorMaterializer()

val response: HttpResponse = ???

response.entity.dataBytes
  .via(Framing.delimiter(ByteString("\n"), maximumFrameLength = 256))
  .map(transformEachLine)
  .runWith(FileIO.toPath(new File("/tmp/example.out").toPath))

def transformEachLine(line: ByteString): ByteString = ???

import java.io.File
import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.http.scaladsl.model._
import scala.concurrent.duration._

implicit val system = ActorSystem()
implicit val dispatcher = system.dispatcher
implicit val materializer = ActorMaterializer()

case class ExamplePerson(name: String)
def parse(line: ByteString): ExamplePerson = ???

val response: HttpResponse = ???

// toStrict to enforce all data be loaded into memory from the connection
val strictEntity: Future[HttpEntity.Strict] = response.entity.toStrict(3.seconds)

// while API remains the same to consume dataBytes, now they're in memory already:
val transformedData: Future[ExamplePerson] =
  strictEntity flatMap { e =>
    e.dataBytes
      .runFold(ByteString.empty) { case (acc, b) => acc ++ b }
      .map(parse)
  }

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.http.scaladsl.model._

implicit val system = ActorSystem()
implicit val dispatcher = system.dispatcher
implicit val materializer = ActorMaterializer()

val response1: HttpResponse = ??? // obtained from an HTTP call (see examples below)

val discarded: DiscardedEntity = response1.discardEntityBytes()
discarded.future.onComplete { case done => println("Entity discarded completely!") }

val response1: HttpResponse = ??? // obtained from an HTTP call (see examples below)

val discardingComplete: Future[Done] = response1.entity.dataBytes.runWith(Sink.ignore)
discardingComplete.onComplete { case done => println("Entity discarded completely!") }

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

implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()
// needed for the future flatMap/onComplete in the end
implicit val executionContext = system.dispatcher

final case class Bid(userId: String, bid: Int)

// these are from spray-json
implicit val bidFormat = jsonFormat2(Bid)

val route =
  path("bid") {
    put {
      entity(as[Bid]) { bid =>
        // incoming entity is fully consumed and converted into a Bid
        complete("The bid was: " + bid)
      }
    }
  }

import akka.actor.ActorSystem
import akka.stream.scaladsl.FileIO
import akka.http.scaladsl.server.Directives._
import akka.stream.ActorMaterializer
import java.io.File

implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()
// needed for the future flatMap/onComplete in the end
implicit val executionContext = system.dispatcher

val route =
  (put & path("lines")) {
    withoutSizeLimit {
      extractDataBytes { bytes =>
        val finishedWriting = bytes.runWith(FileIO.toPath(new File("/tmp/example.out").toPath))
        
        // we only want to respond once the incoming data has been handled:
        onComplete(finishedWriting) { ioResult =>
          complete("Finished writing data: " + ioResult)
        }
      }
    }
  }

import akka.actor.ActorSystem
import akka.stream.scaladsl.FileIO
import akka.http.scaladsl.server.Directives._
import akka.stream.ActorMaterializer
import akka.http.scaladsl.model.HttpRequest

implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()
// needed for the future flatMap/onComplete in the end
implicit val executionContext = system.dispatcher

val route =
  (put & path("lines")) {
    withoutSizeLimit {
      extractRequest { r: HttpRequest =>
        val finishedWriting = r.discardEntityBytes().future
        
        // we only want to respond once the incoming data has been handled:
        onComplete(finishedWriting) { done =>
          complete("Drained all data from connection... (" + done + ")")
        }
      }
    }
  }

import akka.actor.ActorSystem
import akka.stream.scaladsl.Sink
import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.model.headers.Connection
import akka.stream.ActorMaterializer

implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()
// needed for the future flatMap/onComplete in the end
implicit val executionContext = system.dispatcher

val route =
  (put & path("lines")) {
    withoutSizeLimit {
      extractDataBytes { data => 
        // Closing connections, method 1 (eager):
        // we deem this request as illegal, and close the connection right away:
        data.runWith(Sink.cancelled) // "brutally" closes the connection
        
        // Closing connections, method 2 (graceful):
        // consider draining connection and replying with `Connection: Close` header
        // if you want the client to close after this request/reply cycle instead:
        respondWithHeader(Connection("close"))
        complete(StatusCodes.Forbidden -> "Not allowed!")
      }
    }
  }