@Test public void shouldWriteFromPublisherAndFlush() { Buffer firstChunk = Buffer.buffer("chunk 1"); Buffer secondChunk = Buffer.buffer("chunk 2"); TestPublisher<DataBuffer> source = TestPublisher.create(); Mono<Void> result = response.writeAndFlushWithInternal(Flux.just(source)); StepVerifier.create(result) .expectSubscription() .then(() -> source.assertMinRequested(1)) .then(() -> source.next(bufferConverter.toDataBuffer(firstChunk))) .then(() -> source.assertMinRequested(1)) .then(() -> source.next(bufferConverter.toDataBuffer(secondChunk))) .then(() -> source.assertMinRequested(1)) .then(source::complete) .verifyComplete(); verify(mockHttpServerResponse).write(firstChunk); verify(mockHttpServerResponse).write(secondChunk); }
@Override protected Mono<Void> writeAndFlushWithInternal(Publisher<? extends Publisher<? extends DataBuffer>> chunks) { return writeWithInternal(Flux.from(chunks).flatMap(identity())); }
VertxServerHttpResponse extends AbstractServerHttpResponse implements ZeroCopyHttpOutputMessage { @Override protected Mono<Void> writeAndFlushWithInternal(Publisher<? extends Publisher<? extends DataBuffer>> chunks) { return writeWithInternal(Flux.from(chunks).flatMap(identity())); } }
VertxServerHttpResponse extends AbstractServerHttpResponse implements ZeroCopyHttpOutputMessage { @Override protected Mono<Void> writeAndFlushWithInternal(Publisher<? extends Publisher<? extends DataBuffer>> chunks) { return writeWithInternal(Flux.from(chunks).flatMap(identity())); } VertxServerHttpResponse(RoutingContext context, BufferConverter bufferConverter); }
VertxServerHttpResponse extends AbstractServerHttpResponse implements ZeroCopyHttpOutputMessage { @Override protected Mono<Void> writeAndFlushWithInternal(Publisher<? extends Publisher<? extends DataBuffer>> chunks) { return writeWithInternal(Flux.from(chunks).flatMap(identity())); } VertxServerHttpResponse(RoutingContext context, BufferConverter bufferConverter); @SuppressWarnings("unchecked") @Override T getNativeResponse(); @Override Mono<Void> writeWith(Path file, long position, long count); }
VertxServerHttpResponse extends AbstractServerHttpResponse implements ZeroCopyHttpOutputMessage { @Override protected Mono<Void> writeAndFlushWithInternal(Publisher<? extends Publisher<? extends DataBuffer>> chunks) { return writeWithInternal(Flux.from(chunks).flatMap(identity())); } VertxServerHttpResponse(RoutingContext context, BufferConverter bufferConverter); @SuppressWarnings("unchecked") @Override T getNativeResponse(); @Override Mono<Void> writeWith(Path file, long position, long count);  }