ChannelInitializer implements ConnectionObserver { @Override public void connectionEstablished() { outboundMessageSender.initialiseOutboundChannel(transport.getWritableByteChannel()); channelReaderExecutorService.execute(new Runnable() { @Override public void run() { final ReadableByteChannel readableByteChannel = transport.getReadableByteChannel(); try { inputStreamReader.blockingStart(readableByteChannel); } catch (RuntimeException e) { LOGGER.error("Exception thrown while reading from stream, channel: " + readableByteChannel, e); } } }); countDownLatch.countDown(); } ChannelInitializer(final Transport transport, final ByteChannelReader inputStreamReader, final OutboundMessageHandler outboundMessageSender, Executor channelReaderExecutorService); @Override void connectionEstablished(); void awaitConnection(); @Override void connectionClosed();  }
@Test public void shouldStartByteChannelReaderOnConnectionEstablished() throws Exception { deterministicExecutor = new DeterministicExecutor(); final ChannelInitializer channelInitializer = new ChannelInitializer(transport, byteChannelReader, outboundMessageHandler, deterministicExecutor); mockery.checking(new Expectations() { { one(transport).getWritableByteChannel(); will(returnValue(writableByteChannel)); one(outboundMessageHandler).initialiseOutboundChannel(writableByteChannel); } }); channelInitializer.connectionEstablished(); mockery.checking(new Expectations() { { one(transport).getReadableByteChannel(); will(returnValue(readableByteChannel)); one(byteChannelReader).blockingStart(readableByteChannel); } }); deterministicExecutor.runPendingCommands(); }