@Test public void aggregatesRequestAndStreamsResponses() { AtomicReference<String> result = new AtomicReference<>(); WebServiceHandler webServiceHandler = (request, ctx) -> { result.set(request.bodyAs(UTF_8)); return Eventual.of(response(OK) .body("abcdef", UTF_8) .build()); }; LiveHttpResponse response = Mono.from( new HttpAggregator(500, webServiceHandler) .handle(LiveHttpRequest.post("/") .body(ByteStream.from("ABCDEF", UTF_8)) .build(), requestContext())) .block(); assertThat(result.get(), is("ABCDEF")); assertThat(Mono.from(response.aggregate(500)).block().bodyAs(UTF_8), is("abcdef")); }
@Override public Eventual<LiveHttpResponse> handle(LiveHttpRequest request, HttpInterceptor.Context context) { return request.aggregate(bytes) .flatMap(aggregated -> this.delegate.handle(aggregated, context)) .map(HttpResponse::stream); }
HttpAggregator implements HttpHandler { @Override public Eventual<LiveHttpResponse> handle(LiveHttpRequest request, HttpInterceptor.Context context) { return request.aggregate(bytes) .flatMap(aggregated -> this.delegate.handle(aggregated, context)) .map(HttpResponse::stream); } }
HttpAggregator implements HttpHandler { @Override public Eventual<LiveHttpResponse> handle(LiveHttpRequest request, HttpInterceptor.Context context) { return request.aggregate(bytes) .flatMap(aggregated -> this.delegate.handle(aggregated, context)) .map(HttpResponse::stream); } HttpAggregator(int bytes, WebServiceHandler delegate);  HttpAggregator(WebServiceHandler delegate); }
HttpAggregator implements HttpHandler { @Override public Eventual<LiveHttpResponse> handle(LiveHttpRequest request, HttpInterceptor.Context context) { return request.aggregate(bytes) .flatMap(aggregated -> this.delegate.handle(aggregated, context)) .map(HttpResponse::stream); } HttpAggregator(int bytes, WebServiceHandler delegate);  HttpAggregator(WebServiceHandler delegate); @Override Eventual<LiveHttpResponse> handle(LiveHttpRequest request, HttpInterceptor.Context context); }
HttpAggregator implements HttpHandler { @Override public Eventual<LiveHttpResponse> handle(LiveHttpRequest request, HttpInterceptor.Context context) { return request.aggregate(bytes) .flatMap(aggregated -> this.delegate.handle(aggregated, context)) .map(HttpResponse::stream); } HttpAggregator(int bytes, WebServiceHandler delegate);  HttpAggregator(WebServiceHandler delegate); @Override Eventual<LiveHttpResponse> handle(LiveHttpRequest request, HttpInterceptor.Context context);  }