ChannelConnectedListener implements ChannelFutureListener { @Override public void operationComplete( ChannelFuture future ) { Channel channel = future.channel(); Logger log = new ChannelActivityLogger( channel, logging, getClass() ); if ( future.isSuccess() ) { log.trace( "Channel %s connected, initiating bolt handshake", channel ); ChannelPipeline pipeline = channel.pipeline(); pipeline.addLast( new HandshakeHandler( pipelineBuilder, handshakeCompletedPromise, logging ) ); log.debug( "C: [Bolt Handshake] %s", handshakeString() ); channel.writeAndFlush( handshakeBuf(), channel.voidPromise() ); } else { handshakeCompletedPromise.setFailure( databaseUnavailableError( address, future.cause() ) ); } } ChannelConnectedListener( BoltServerAddress address, ChannelPipelineBuilder pipelineBuilder,
            ChannelPromise handshakeCompletedPromise, Logging logging ); @Override void operationComplete( ChannelFuture future );  }
@Test void shouldFailPromiseWhenChannelConnectionFails() { ChannelPromise handshakeCompletedPromise = channel.newPromise(); ChannelConnectedListener listener = newListener( handshakeCompletedPromise ); ChannelPromise channelConnectedPromise = channel.newPromise(); IOException cause = new IOException( "Unable to connect!" ); channelConnectedPromise.setFailure( cause ); listener.operationComplete( channelConnectedPromise ); ServiceUnavailableException error = assertThrows( ServiceUnavailableException.class, () -> await( handshakeCompletedPromise ) ); assertEquals( cause, error.getCause() ); }
@Test void shouldWriteHandshakeWhenChannelConnected() { ChannelPromise handshakeCompletedPromise = channel.newPromise(); ChannelConnectedListener listener = newListener( handshakeCompletedPromise ); ChannelPromise channelConnectedPromise = channel.newPromise(); channelConnectedPromise.setSuccess(); listener.operationComplete( channelConnectedPromise ); assertNotNull( channel.pipeline().get( HandshakeHandler.class ) ); assertTrue( channel.finish() ); assertEquals( handshakeBuf(), channel.readOutbound() ); }