@Test public void closeTest() throws IOException { final AtomicBoolean streamOpen = new AtomicBoolean(true); InputStream stream = new InputStream() { @Override public int read() throws IOException { throw new IllegalStateException("Not implemented"); } @Override public void close() { streamOpen.set(false); } }; InputStreamReadableStreamChannel channel = new InputStreamReadableStreamChannel(stream, EXECUTOR_SERVICE); assertTrue("Channel should be open", channel.isOpen()); channel.close(); assertFalse("Channel should be closed", channel.isOpen()); assertFalse("Stream should be closed", streamOpen.get()); channel.close(); }
@Override public void close() throws IOException { if (channelOpen.compareAndSet(true, false)) { inputStream.close(); } }
InputStreamReadableStreamChannel implements ReadableStreamChannel { @Override public void close() throws IOException { if (channelOpen.compareAndSet(true, false)) { inputStream.close(); } } }
InputStreamReadableStreamChannel implements ReadableStreamChannel { @Override public void close() throws IOException { if (channelOpen.compareAndSet(true, false)) { inputStream.close(); } } InputStreamReadableStreamChannel(InputStream inputStream, ExecutorService executorService);  InputStreamReadableStreamChannel(InputStream inputStream, long size, ExecutorService executorService); }
InputStreamReadableStreamChannel implements ReadableStreamChannel { @Override public void close() throws IOException { if (channelOpen.compareAndSet(true, false)) { inputStream.close(); } } InputStreamReadableStreamChannel(InputStream inputStream, ExecutorService executorService);  InputStreamReadableStreamChannel(InputStream inputStream, long size, ExecutorService executorService); @Override long getSize(); @Override Future<Long> readInto(AsyncWritableChannel asyncWritableChannel, Callback<Long> callback); @Override boolean isOpen(); @Override void close(); }
InputStreamReadableStreamChannel implements ReadableStreamChannel { @Override public void close() throws IOException { if (channelOpen.compareAndSet(true, false)) { inputStream.close(); } } InputStreamReadableStreamChannel(InputStream inputStream, ExecutorService executorService);  InputStreamReadableStreamChannel(InputStream inputStream, long size, ExecutorService executorService); @Override long getSize(); @Override Future<Long> readInto(AsyncWritableChannel asyncWritableChannel, Callback<Long> callback); @Override boolean isOpen(); @Override void close();  }