@Test public void shouldWriteFromPublisherAndFlush() { Buffer firstChunk = Buffer.buffer("chunk 1"); Buffer secondChunk = Buffer.buffer("chunk 2"); TestPublisher<DataBuffer> source = TestPublisher.create(); Mono<Void> result = request.writeAndFlushWith(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(mockHttpClientRequest).write(firstChunk); verify(mockHttpClientRequest).write(secondChunk); verify(mockHttpClientRequest).end(); }
@Override public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> chunks) { return writeWith(Flux.from(chunks).flatMap(identity())); }
VertxClientHttpRequest extends AbstractClientHttpRequest { @Override public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> chunks) { return writeWith(Flux.from(chunks).flatMap(identity())); } }
VertxClientHttpRequest extends AbstractClientHttpRequest { @Override public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> chunks) { return writeWith(Flux.from(chunks).flatMap(identity())); } VertxClientHttpRequest(HttpClientRequest delegate, BufferConverter bufferConverter); }
VertxClientHttpRequest extends AbstractClientHttpRequest { @Override public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> chunks) { return writeWith(Flux.from(chunks).flatMap(identity())); } VertxClientHttpRequest(HttpClientRequest delegate, BufferConverter bufferConverter); @Override HttpMethod getMethod(); @Override URI getURI(); @Override DataBufferFactory bufferFactory(); @Override Mono<Void> writeWith(Publisher<? extends DataBuffer> chunks); @Override Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> chunks); @Override Mono<Void> setComplete(); }
VertxClientHttpRequest extends AbstractClientHttpRequest { @Override public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> chunks) { return writeWith(Flux.from(chunks).flatMap(identity())); } VertxClientHttpRequest(HttpClientRequest delegate, BufferConverter bufferConverter); @Override HttpMethod getMethod(); @Override URI getURI(); @Override DataBufferFactory bufferFactory(); @Override Mono<Void> writeWith(Publisher<? extends DataBuffer> chunks); @Override Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> chunks); @Override Mono<Void> setComplete();  }