@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));
    }
}