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

import java.io.File;
import akka.actor.ActorSystem;

import java.util.concurrent.TimeUnit;
import java.util.function.Function; 
import akka.stream.ActorMaterializer;
import akka.stream.javadsl.Framing;
import akka.http.javadsl.model.*;
import scala.concurrent.duration.FiniteDuration;
import scala.util.Try;

    final ActorSystem system = ActorSystem.create();
    final ExecutionContextExecutor dispatcher = system.dispatcher();
    final ActorMaterializer materializer = ActorMaterializer.create(system);

    final HttpResponse response = responseFromSomewhere();

    final Function<ByteString, ByteString> transformEachLine = line -> line /* some transformation here */;

    final int maximumFrameLength = 256;

    response.entity().getDataBytes()
      .via(Framing.delimiter(ByteString.fromString("\n"), maximumFrameLength, FramingTruncation.ALLOW))
      .map(transformEachLine::apply)
      .runWith(FileIO.toPath(new File("/tmp/example.out").toPath()), materializer);

final class ExamplePerson {
  final String name;
  public ExamplePerson(String name) { this.name = name; }
}

public ExamplePerson parse(ByteString line) { 
  return new ExamplePerson(line.utf8String()); 
}
  final ActorSystem system = ActorSystem.create();
  final ExecutionContextExecutor dispatcher = system.dispatcher();
  final ActorMaterializer materializer = ActorMaterializer.create(system);
  
  final HttpResponse response = responseFromSomewhere();
  
  // toStrict to enforce all data be loaded into memory from the connection
  final CompletionStage<HttpEntity.Strict> strictEntity = response.entity()
      .toStrict(FiniteDuration.create(3, TimeUnit.SECONDS).toMillis(), materializer);

  // while API remains the same to consume dataBytes, now they're in memory already:

  final CompletionStage<ExamplePerson> person = 
    strictEntity
      .thenCompose(strict ->
        strict.getDataBytes()
          .runFold(ByteString.empty(), (acc, b) -> acc.concat(b), materializer)
          .thenApply(this::parse)
      );

final ActorSystem system = ActorSystem.create();
final ExecutionContextExecutor dispatcher = system.dispatcher();
final ActorMaterializer materializer = ActorMaterializer.create(system);

final HttpResponse response = responseFromSomewhere();

final HttpMessage.DiscardedEntity discarded = response.discardEntityBytes(materializer);

discarded.completionStage().whenComplete((done, ex) -> {
  System.out.println("Entity discarded completely!");
});

final ActorSystem system = ActorSystem.create();
final ExecutionContextExecutor dispatcher = system.dispatcher();
final ActorMaterializer materializer = ActorMaterializer.create(system);

final HttpResponse response = responseFromSomewhere();

final CompletionStage<Done> discardingComplete = response.entity().getDataBytes().runWith(Sink.ignore(), materializer);

discardingComplete.whenComplete((done, ex) -> {
  System.out.println("Entity discarded completely!");
});

class Bid {
  final String userId;
  final int bid;

  Bid(String userId, int bid) {
    this.userId = userId;
    this.bid = bid; 
  }
}
  final ActorSystem system = ActorSystem.create();
  final ExecutionContextExecutor dispatcher = system.dispatcher();
  final ActorMaterializer materializer = ActorMaterializer.create(system);
  
  final Unmarshaller<HttpEntity, Bid> asBid = Jackson.unmarshaller(Bid.class);

  final Route s = path("bid", () ->
    put(() ->
      entity(asBid, bid ->
        // incoming entity is fully consumed and converted into a Bid
        complete("The bid was: " + bid)
      )
    )
  );

final ActorSystem system = ActorSystem.create();
final ExecutionContextExecutor dispatcher = system.dispatcher();
final ActorMaterializer materializer = ActorMaterializer.create(system);

final Route s =
  put(() ->
    path("lines", () ->
      withoutSizeLimit(() ->
        extractDataBytes(bytes -> {
          final CompletionStage<IOResult> res = bytes.runWith(FileIO.toPath(new File("/tmp/example.out").toPath()), materializer);

          return onComplete(() -> res, ioResult ->
            // we only want to respond once the incoming data has been handled:
            complete("Finished writing data :" + ioResult));
        })
      )
    )
  );

final ActorSystem system = ActorSystem.create();
final ExecutionContextExecutor dispatcher = system.dispatcher();
final ActorMaterializer materializer = ActorMaterializer.create(system);

final Route s =
  put(() ->
    path("lines", () ->
      withoutSizeLimit(() ->
        extractRequest(r -> {
          final CompletionStage<Done> res = r.discardEntityBytes(materializer).completionStage();

          return onComplete(() -> res, done ->
            // we only want to respond once the incoming data has been handled:
            complete("Finished writing data :" + done));
        })
      )
    )
  );

final ActorSystem system = ActorSystem.create();
final ExecutionContextExecutor dispatcher = system.dispatcher();
final ActorMaterializer materializer = ActorMaterializer.create(system);

final Route s =
  put(() ->
    path("lines", () ->
      withoutSizeLimit(() ->
        extractDataBytes(bytes -> {
          // Closing connections, method 1 (eager):
          // we deem this request as illegal, and close the connection right away:
          bytes.runWith(Sink.cancelled(), materializer);  // "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:
          return respondWithHeader(Connection.create("close"), () -> 
            complete(StatusCodes.FORBIDDEN, "Not allowed!")
          );
        })
      )
    )
  );