@Test public void shouldWriteFromPublisher() { Buffer firstChunk = Buffer.buffer("chunk 1"); Buffer secondChunk = Buffer.buffer("chunk 2"); TestPublisher<DataBuffer> source = TestPublisher.create(); Mono<Void> result = response.writeWithInternal(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> writeWithInternal(Publisher<? extends DataBuffer> chunks) { return Mono.create(sink -> { logger.debug("Subscribing to body publisher"); Subscriber<DataBuffer> subscriber = new WriteStreamSubscriber.Builder<HttpServerResponse, DataBuffer>() .writeStream(delegate) .endHook(sink) .nextHandler((stream, value) -> stream.write(bufferConverter.toBuffer(value))) .build(); chunks.subscribe(subscriber); }); }
VertxServerHttpResponse extends AbstractServerHttpResponse implements ZeroCopyHttpOutputMessage { @Override protected Mono<Void> writeWithInternal(Publisher<? extends DataBuffer> chunks) { return Mono.create(sink -> { logger.debug("Subscribing to body publisher"); Subscriber<DataBuffer> subscriber = new WriteStreamSubscriber.Builder<HttpServerResponse, DataBuffer>() .writeStream(delegate) .endHook(sink) .nextHandler((stream, value) -> stream.write(bufferConverter.toBuffer(value))) .build(); chunks.subscribe(subscriber); }); } }
VertxServerHttpResponse extends AbstractServerHttpResponse implements ZeroCopyHttpOutputMessage { @Override protected Mono<Void> writeWithInternal(Publisher<? extends DataBuffer> chunks) { return Mono.create(sink -> { logger.debug("Subscribing to body publisher"); Subscriber<DataBuffer> subscriber = new WriteStreamSubscriber.Builder<HttpServerResponse, DataBuffer>() .writeStream(delegate) .endHook(sink) .nextHandler((stream, value) -> stream.write(bufferConverter.toBuffer(value))) .build(); chunks.subscribe(subscriber); }); } VertxServerHttpResponse(RoutingContext context, BufferConverter bufferConverter); }
VertxServerHttpResponse extends AbstractServerHttpResponse implements ZeroCopyHttpOutputMessage { @Override protected Mono<Void> writeWithInternal(Publisher<? extends DataBuffer> chunks) { return Mono.create(sink -> { logger.debug("Subscribing to body publisher"); Subscriber<DataBuffer> subscriber = new WriteStreamSubscriber.Builder<HttpServerResponse, DataBuffer>() .writeStream(delegate) .endHook(sink) .nextHandler((stream, value) -> stream.write(bufferConverter.toBuffer(value))) .build(); chunks.subscribe(subscriber); }); } 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> writeWithInternal(Publisher<? extends DataBuffer> chunks) { return Mono.create(sink -> { logger.debug("Subscribing to body publisher"); Subscriber<DataBuffer> subscriber = new WriteStreamSubscriber.Builder<HttpServerResponse, DataBuffer>() .writeStream(delegate) .endHook(sink) .nextHandler((stream, value) -> stream.write(bufferConverter.toBuffer(value))) .build(); chunks.subscribe(subscriber); }); } VertxServerHttpResponse(RoutingContext context, BufferConverter bufferConverter); @SuppressWarnings("unchecked") @Override T getNativeResponse(); @Override Mono<Void> writeWith(Path file, long position, long count);  }