ArmeriaServerHttpRequest extends AbstractServerHttpRequest { @Override public Flux<DataBuffer> getBody() { return body; } ArmeriaServerHttpRequest(ServiceRequestContext ctx,
                             HttpRequest req,
                             DataBufferFactoryWrapper<?> factoryWrapper); @SuppressWarnings("unchecked") @Override T getNativeRequest(); @Override String getMethodValue(); @Override Flux<DataBuffer> getBody(); @Override InetSocketAddress getRemoteAddress(); @Override String toString();  }
@Test void cancel() { final HttpRequest httpRequest = HttpRequest.of(RequestHeaders.builder(HttpMethod.POST, "/") .scheme("http") .authority("127.0.0.1") .build(), Flux.just("a", "b", "c", "d", "e").map(HttpData::ofUtf8)); final ServiceRequestContext ctx = newRequestContext(httpRequest); final ArmeriaServerHttpRequest req = request(ctx); assertThat(httpRequest.whenComplete().isDone()).isFalse(); final Flux<String> body = req.getBody().map(TestUtil::bufferToString); StepVerifier.create(body, 1) .expectNext("a").thenRequest(1) .expectNext("b").thenRequest(1) .thenCancel() .verify(); final CompletableFuture<Void> f = httpRequest.whenComplete(); assertThat(f.isDone()).isTrue(); assertThat(f.isCompletedExceptionally()).isTrue(); assertThatThrownBy(f::get).isInstanceOf(ExecutionException.class) .hasCauseInstanceOf(CancelledSubscriptionException.class); }