ByteChannelReader { public void blockingStart(final ReadableByteChannel readableByteChannel) { try { while ((readableByteChannel.read(buffer)) != -1) { blocker.mayWait(); buffer.flip(); byteStreamMessageParser.parse(buffer); buffer.clear(); } } catch (final ClosedChannelException e) { } catch (final IOException e) { LOGGER.error("An error occurred trying to read from the socket", e); } finally { if (readableByteChannel != null) { try { if (readableByteChannel.isOpen()) { readableByteChannel.close(); } } catch (IOException e) { } connectionObserver.connectionClosed(); } } } ByteChannelReader(final ByteStreamMessageParser byteStreamMessageParser, final Blocker blocker, final ConnectionObserver connectionObserver); void blockingStart(final ReadableByteChannel readableByteChannel);  }
@Test public void shouldPassBytesFromInputStreamToParser() throws Exception { mockery.checking(new Expectations() { { ignoring(connectionObserver).connectionClosed(); allowing(readableByteChannel).isOpen(); will(returnValue(false)); one(readableByteChannel).read(with(any(ByteBuffer.class))); will(returnValue(1)); one(readableByteChannel).read(with(any(ByteBuffer.class))); will(returnValue(-1)); one(byteStreamMessageParser).parse(with(any(ByteBuffer.class))); } }); inputStreamReader.blockingStart(readableByteChannel); }
@Test(expected = RuntimeException.class) public void shouldCloseInputStreamIfExceptionIsThrownAndInputStreamIsStillOpen() throws Exception { mockery.checking(new Expectations() { { ignoring(connectionObserver).connectionClosed(); allowing(byteStreamMessageParser).parse(with(any(ByteBuffer.class))); one(readableByteChannel).read(with(any(ByteBuffer.class))); will(throwException(new RuntimeException("boom!"))); one(readableByteChannel).isOpen(); will(returnValue(true)); one(readableByteChannel).close(); } }); inputStreamReader.blockingStart(readableByteChannel); }
@Test public void shouldNotifyTransportObserverWhenConnectionIsClosed() throws Exception { mockery.checking(new Expectations() { { ignoring(readableByteChannel).isOpen(); allowing(byteStreamMessageParser).parse(with(any(ByteBuffer.class))); allowing(readableByteChannel).read(with(any(ByteBuffer.class))); will(throwException(new ClosedChannelException())); allowing(readableByteChannel).close(); one(connectionObserver).connectionClosed(); } }); inputStreamReader.blockingStart(readableByteChannel); }
@Test public void shouldNotifyTransportObserverWhenIOExceptionIsThrown() throws Exception { mockery.checking(new Expectations() { { ignoring(readableByteChannel).isOpen(); allowing(byteStreamMessageParser).parse(with(any(ByteBuffer.class))); allowing(readableByteChannel).read(with(any(ByteBuffer.class))); will(throwException(new IOException())); allowing(readableByteChannel).close(); one(connectionObserver).connectionClosed(); } }); inputStreamReader.blockingStart(readableByteChannel); }
@Test public void shouldCloseChannelWhenEndOfStreamIsReached() throws Exception { expectEndOfInputStream(); mockery.checking(new Expectations() { { one(readableByteChannel).isOpen(); will(returnValue(true)); one(readableByteChannel).close(); one(connectionObserver).connectionClosed(); } }); inputStreamReader.blockingStart(readableByteChannel); }