@Test public void badInputStreamTest() throws Exception { final String errMsg = "@@ExpectedExceptionMessage@@"; InputStream stream = new InputStream() { @Override public int read() throws IOException { throw new IllegalStateException(errMsg); } }; InputStreamReadableStreamChannel channel = new InputStreamReadableStreamChannel(stream, EXECUTOR_SERVICE); ReadIntoCallback callback = new ReadIntoCallback(); try { channel.readInto(new BadAsyncWritableChannel(new IOException(errMsg)), callback).get(); fail("Should have failed because the InputStream would have thrown exception"); } catch (ExecutionException e) { Exception exception = (Exception) Utils.getRootCause(e); assertEquals("Exception message does not match expected (future)", errMsg, exception.getMessage()); callback.awaitCallback(); assertEquals("Exception message does not match expected (callback)", errMsg, callback.exception.getMessage()); } }
@Override public Future<Long> readInto(AsyncWritableChannel asyncWritableChannel, Callback<Long> callback) { Future<Long> future; if (!channelOpen.get()) { ClosedChannelException closedChannelException = new ClosedChannelException(); FutureResult<Long> futureResult = new FutureResult<Long>(); futureResult.done(0L, closedChannelException); future = futureResult; if (callback != null) { callback.onCompletion(0L, closedChannelException); } } else if (!readIntoCalled.compareAndSet(false, true)) { throw new IllegalStateException("ReadableStreamChannel cannot be read more than once"); } else { callbackWrapper = new ReadIntoCallbackWrapper(callback); future = callbackWrapper.futureResult; if (size == 0) { callbackWrapper.invokeCallback(null); } else { writeToChannel(asyncWritableChannel); } } return future; }
InputStreamReadableStreamChannel implements ReadableStreamChannel { @Override public Future<Long> readInto(AsyncWritableChannel asyncWritableChannel, Callback<Long> callback) { Future<Long> future; if (!channelOpen.get()) { ClosedChannelException closedChannelException = new ClosedChannelException(); FutureResult<Long> futureResult = new FutureResult<Long>(); futureResult.done(0L, closedChannelException); future = futureResult; if (callback != null) { callback.onCompletion(0L, closedChannelException); } } else if (!readIntoCalled.compareAndSet(false, true)) { throw new IllegalStateException("ReadableStreamChannel cannot be read more than once"); } else { callbackWrapper = new ReadIntoCallbackWrapper(callback); future = callbackWrapper.futureResult; if (size == 0) { callbackWrapper.invokeCallback(null); } else { writeToChannel(asyncWritableChannel); } } return future; } }
InputStreamReadableStreamChannel implements ReadableStreamChannel { @Override public Future<Long> readInto(AsyncWritableChannel asyncWritableChannel, Callback<Long> callback) { Future<Long> future; if (!channelOpen.get()) { ClosedChannelException closedChannelException = new ClosedChannelException(); FutureResult<Long> futureResult = new FutureResult<Long>(); futureResult.done(0L, closedChannelException); future = futureResult; if (callback != null) { callback.onCompletion(0L, closedChannelException); } } else if (!readIntoCalled.compareAndSet(false, true)) { throw new IllegalStateException("ReadableStreamChannel cannot be read more than once"); } else { callbackWrapper = new ReadIntoCallbackWrapper(callback); future = callbackWrapper.futureResult; if (size == 0) { callbackWrapper.invokeCallback(null); } else { writeToChannel(asyncWritableChannel); } } return future; } InputStreamReadableStreamChannel(InputStream inputStream, ExecutorService executorService);  InputStreamReadableStreamChannel(InputStream inputStream, long size, ExecutorService executorService); }
InputStreamReadableStreamChannel implements ReadableStreamChannel { @Override public Future<Long> readInto(AsyncWritableChannel asyncWritableChannel, Callback<Long> callback) { Future<Long> future; if (!channelOpen.get()) { ClosedChannelException closedChannelException = new ClosedChannelException(); FutureResult<Long> futureResult = new FutureResult<Long>(); futureResult.done(0L, closedChannelException); future = futureResult; if (callback != null) { callback.onCompletion(0L, closedChannelException); } } else if (!readIntoCalled.compareAndSet(false, true)) { throw new IllegalStateException("ReadableStreamChannel cannot be read more than once"); } else { callbackWrapper = new ReadIntoCallbackWrapper(callback); future = callbackWrapper.futureResult; if (size == 0) { callbackWrapper.invokeCallback(null); } else { writeToChannel(asyncWritableChannel); } } return future; } 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 Future<Long> readInto(AsyncWritableChannel asyncWritableChannel, Callback<Long> callback) { Future<Long> future; if (!channelOpen.get()) { ClosedChannelException closedChannelException = new ClosedChannelException(); FutureResult<Long> futureResult = new FutureResult<Long>(); futureResult.done(0L, closedChannelException); future = futureResult; if (callback != null) { callback.onCompletion(0L, closedChannelException); } } else if (!readIntoCalled.compareAndSet(false, true)) { throw new IllegalStateException("ReadableStreamChannel cannot be read more than once"); } else { callbackWrapper = new ReadIntoCallbackWrapper(callback); future = callbackWrapper.futureResult; if (size == 0) { callbackWrapper.invokeCallback(null); } else { writeToChannel(asyncWritableChannel); } } return future; } 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();  }