@Test public void shouldWriteFromPublisher() { Buffer firstChunk = Buffer.buffer("chunk 1"); Buffer secondChunk = Buffer.buffer("chunk 2"); TestPublisher<DataBuffer> source = TestPublisher.create(); Mono<Void> result = request.writeWith(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> writeWith(Publisher<? extends DataBuffer> chunks) { Mono<Void> writeCompletion = Mono.create(sink -> { logger.debug("Subscribing to body publisher"); Subscriber<DataBuffer> subscriber = new WriteStreamSubscriber.Builder<HttpClientRequest, DataBuffer>() .writeStream(delegate) .endHook(sink) .nextHandler((stream, value) -> stream.write(bufferConverter.toBuffer(value))) .build(); chunks.subscribe(subscriber); }); Mono<Void> endCompletion = Mono.create(sink -> { logger.debug("Completing request after writing"); delegate.end(); sink.success(); }); return doCommit(() -> writeCompletion.then(endCompletion)); }
VertxClientHttpRequest extends AbstractClientHttpRequest { @Override public Mono<Void> writeWith(Publisher<? extends DataBuffer> chunks) { Mono<Void> writeCompletion = Mono.create(sink -> { logger.debug("Subscribing to body publisher"); Subscriber<DataBuffer> subscriber = new WriteStreamSubscriber.Builder<HttpClientRequest, DataBuffer>() .writeStream(delegate) .endHook(sink) .nextHandler((stream, value) -> stream.write(bufferConverter.toBuffer(value))) .build(); chunks.subscribe(subscriber); }); Mono<Void> endCompletion = Mono.create(sink -> { logger.debug("Completing request after writing"); delegate.end(); sink.success(); }); return doCommit(() -> writeCompletion.then(endCompletion)); } }
VertxClientHttpRequest extends AbstractClientHttpRequest { @Override public Mono<Void> writeWith(Publisher<? extends DataBuffer> chunks) { Mono<Void> writeCompletion = Mono.create(sink -> { logger.debug("Subscribing to body publisher"); Subscriber<DataBuffer> subscriber = new WriteStreamSubscriber.Builder<HttpClientRequest, DataBuffer>() .writeStream(delegate) .endHook(sink) .nextHandler((stream, value) -> stream.write(bufferConverter.toBuffer(value))) .build(); chunks.subscribe(subscriber); }); Mono<Void> endCompletion = Mono.create(sink -> { logger.debug("Completing request after writing"); delegate.end(); sink.success(); }); return doCommit(() -> writeCompletion.then(endCompletion)); } VertxClientHttpRequest(HttpClientRequest delegate, BufferConverter bufferConverter); }
VertxClientHttpRequest extends AbstractClientHttpRequest { @Override public Mono<Void> writeWith(Publisher<? extends DataBuffer> chunks) { Mono<Void> writeCompletion = Mono.create(sink -> { logger.debug("Subscribing to body publisher"); Subscriber<DataBuffer> subscriber = new WriteStreamSubscriber.Builder<HttpClientRequest, DataBuffer>() .writeStream(delegate) .endHook(sink) .nextHandler((stream, value) -> stream.write(bufferConverter.toBuffer(value))) .build(); chunks.subscribe(subscriber); }); Mono<Void> endCompletion = Mono.create(sink -> { logger.debug("Completing request after writing"); delegate.end(); sink.success(); }); return doCommit(() -> writeCompletion.then(endCompletion)); } 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> writeWith(Publisher<? extends DataBuffer> chunks) { Mono<Void> writeCompletion = Mono.create(sink -> { logger.debug("Subscribing to body publisher"); Subscriber<DataBuffer> subscriber = new WriteStreamSubscriber.Builder<HttpClientRequest, DataBuffer>() .writeStream(delegate) .endHook(sink) .nextHandler((stream, value) -> stream.write(bufferConverter.toBuffer(value))) .build(); chunks.subscribe(subscriber); }); Mono<Void> endCompletion = Mono.create(sink -> { logger.debug("Completing request after writing"); delegate.end(); sink.success(); }); return doCommit(() -> writeCompletion.then(endCompletion)); } 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();  }