@Test public void closeMoreTheOnce() throws IOException { InputStream is = new DataChunkInputStream(Single.just(DataChunk.create("test".getBytes()))); is.close(); is.close(); }
@Override public void close() { Optional.ofNullable(current).ifPresent(it -> current.whenComplete(DataChunkInputStream::releaseChunk)); current = null; bufferIndex = 0; }
DataChunkInputStream extends InputStream { @Override public void close() { Optional.ofNullable(current).ifPresent(it -> current.whenComplete(DataChunkInputStream::releaseChunk)); current = null; bufferIndex = 0; } }
DataChunkInputStream extends InputStream { @Override public void close() { Optional.ofNullable(current).ifPresent(it -> current.whenComplete(DataChunkInputStream::releaseChunk)); current = null; bufferIndex = 0; } DataChunkInputStream(Flow.Publisher<DataChunk> originalPublisher);  DataChunkInputStream(Flow.Publisher<DataChunk> originalPublisher, boolean validate); }
DataChunkInputStream extends InputStream { @Override public void close() { Optional.ofNullable(current).ifPresent(it -> current.whenComplete(DataChunkInputStream::releaseChunk)); current = null; bufferIndex = 0; } DataChunkInputStream(Flow.Publisher<DataChunk> originalPublisher);  DataChunkInputStream(Flow.Publisher<DataChunk> originalPublisher, boolean validate); @Override void close(); @Override int read(); @Override int read(byte[] buf, int off, int len); }
DataChunkInputStream extends InputStream { @Override public void close() { Optional.ofNullable(current).ifPresent(it -> current.whenComplete(DataChunkInputStream::releaseChunk)); current = null; bufferIndex = 0; } DataChunkInputStream(Flow.Publisher<DataChunk> originalPublisher);  DataChunkInputStream(Flow.Publisher<DataChunk> originalPublisher, boolean validate); @Override void close(); @Override int read(); @Override int read(byte[] buf, int off, int len);  }