ReactiveDataProducer implements AsyncDataProducer, Subscriber<ByteBuffer> { @Override public void produce(final DataStreamChannel channel) throws IOException { if (requestChannel.get() == null) { requestChannel.set(channel); publisher.subscribe(this); } final Throwable t = exception.get(); final Subscription s = subscription.get(); int buffersToReplenish = 0; try { synchronized (buffers) { if (t != null) { throw new HttpStreamResetException(t.getMessage(), t); } else if (this.complete.get() && buffers.isEmpty()) { channel.endStream(); } else { while (buffers.size() > 0) { final ByteBuffer nextBuffer = buffers.remove(); channel.write(nextBuffer); if (nextBuffer.remaining() > 0) { buffers.push(nextBuffer); break; } else if (s != null) { buffersToReplenish++; } } } } } finally { if (s != null && buffersToReplenish > 0) { s.request(buffersToReplenish); } } } ReactiveDataProducer(final Publisher<ByteBuffer> publisher); @Override void onSubscribe(final Subscription subscription); @Override void onNext(final ByteBuffer byteBuffer); @Override void onError(final Throwable throwable); @Override void onComplete(); @Override int available(); @Override void produce(final DataStreamChannel channel); @Override void releaseResources();  }
@Test public void testStreamThatEndsNormally() throws Exception { final Flowable<ByteBuffer> publisher = Flowable.just( ByteBuffer.wrap(new byte[]{ '1', '2', '3' }), ByteBuffer.wrap(new byte[]{ '4', '5', '6' })); final ReactiveDataProducer producer = new ReactiveDataProducer(publisher); final WritableByteChannelMock byteChannel = new WritableByteChannelMock(1024); final DataStreamChannel streamChannel = new BasicDataStreamChannel(byteChannel); producer.produce(streamChannel); Assert.assertTrue(byteChannel.isOpen()); Assert.assertEquals("123456", byteChannel.dump(StandardCharsets.US_ASCII)); producer.produce(streamChannel); Assert.assertFalse(byteChannel.isOpen()); Assert.assertEquals("", byteChannel.dump(StandardCharsets.US_ASCII)); }
@Test public void testStreamThatEndsWithError() throws Exception { final Flowable<ByteBuffer> publisher = Flowable.concatArray( Flowable.just( ByteBuffer.wrap(new byte[]{ '1' }), ByteBuffer.wrap(new byte[]{ '2' }), ByteBuffer.wrap(new byte[]{ '3' }), ByteBuffer.wrap(new byte[]{ '4' }), ByteBuffer.wrap(new byte[]{ '5' }), ByteBuffer.wrap(new byte[]{ '6' })), Flowable.<ByteBuffer>error(new RuntimeException()) ); final ReactiveDataProducer producer = new ReactiveDataProducer(publisher); final WritableByteChannelMock byteChannel = new WritableByteChannelMock(1024); final DataStreamChannel streamChannel = new BasicDataStreamChannel(byteChannel); producer.produce(streamChannel); Assert.assertEquals("12345", byteChannel.dump(StandardCharsets.US_ASCII)); try { producer.produce(streamChannel); Assert.fail("Expected ProtocolException"); } catch (final HttpStreamResetException ex) { Assert.assertTrue("Expected published exception to be rethrown", ex.getCause() instanceof RuntimeException); Assert.assertEquals("", byteChannel.dump(StandardCharsets.US_ASCII)); } }