NettyHttpDatabusBootstrapConnection extends AbstractNettyHttpConnection implements DatabusBootstrapConnection { @Override public void requestStream(String sourcesIdList, DbusKeyFilter filter, int freeBufferSpace, Checkpoint cp, DatabusBootstrapConnectionStateMessage stateReuse) { _checkpoint = cp; _callbackStateReuse = stateReuse; _sourcesIdList = sourcesIdList; _freeBufferSpace = freeBufferSpace; _filter = filter; _handler = null; if (!hasConnection()) { connect(State.STREAM_REQUEST_CONNECT); } else { onStreamConnectSuccess(); } } NettyHttpDatabusBootstrapConnection(ServerInfo server,
                                         ActorMessageQueue callback,
                                         ClientBootstrap bootstrap,
                                         ContainerStatisticsCollector containerStatsCollector,
                                         RemoteExceptionHandler remoteExceptionHandler,
                                         Timer timeoutTimer,
                                         long writeTimeoutMs,
                                         long readTimeoutMs,
                                         int protocolVersion,
                                         ChannelGroup channelGroup);  NettyHttpDatabusBootstrapConnection(ServerInfo relay,
                                             ActorMessageQueue callback,
                                             ChannelFactory channelFactory,
                                             ContainerStatisticsCollector containerStatsCollector,
                                             RemoteExceptionHandler remoteExceptionHandler,
                                             Timer timeoutTimer,
                                             long writeTimeoutMs,
                                             long readTimeoutMs,
                                             int protocolVersion,
                                             ChannelGroup channelGroup); @Override void requestTargetScn(Checkpoint checkpoint, DatabusBootstrapConnectionStateMessage stateReuse); @Override void requestStartScn(Checkpoint checkpoint, DatabusBootstrapConnectionStateMessage stateReuse, String sourceNamesList); @Override void requestStream(String sourcesIdList, DbusKeyFilter filter,
		  					int freeBufferSpace, Checkpoint cp,
                            DatabusBootstrapConnectionStateMessage stateReuse); static final String MODULE; static final Logger LOG; }
@Test public void testServerBootstrapDisconnect() throws Exception { final Logger log = Logger.getLogger("TestNettyHttpDatabusBootstrapConnection.testServerBootstrapDisconnect"); log.info("starting"); log.info("setup the client"); TestingConnectionCallback callback = TestingConnectionCallback.createAndStart("testServerSourcesDisconnect"); DummyRemoteExceptionHandler remoteExceptionHandler = new DummyRemoteExceptionHandler(); final NettyHttpDatabusBootstrapConnection conn = (NettyHttpDatabusBootstrapConnection) CONN_FACTORY.createConnection(BOOTSTRAP_SERVER_INFO, callback, remoteExceptionHandler); try { log.info("initial setup"); final List<String> sourceNamesList = Arrays.asList(SOURCE1_NAME); final Checkpoint cp = Checkpoint.createOnlineConsumptionCheckpoint(0); BootstrapCheckpointHandler cpHandler = new BootstrapCheckpointHandler(sourceNamesList); cpHandler.createInitialBootstrapCheckpoint(cp, 0L); final DummyDatabusBootstrapConnectionStateMessage bstCallback = new DummyDatabusBootstrapConnectionStateMessage(log); log.info("process a normal startSCN which should establish the connection"); sendStartScnHappyPath(conn, cp, bstCallback, SOURCE1_NAME, 100L, log); Assert.assertTrue(conn.isConnected()); TestUtil.assertWithBackoff(new ConditionCheck() { @Override public boolean check() { return null != bstCallback.getCheckpoint(); } }, "wait for /startSCN response", 100, log); log.info("verify /startSCN response"); final Checkpoint startScnCp = bstCallback.getCheckpoint(); Assert.assertNotNull(startScnCp); Assert.assertEquals(100L, startScnCp.getBootstrapStartScn().longValue()); log.info("instrument the client pipeline so that we can intercept and delay the channelClosed message"); final Semaphore passMessage = new Semaphore(1); final CountDownLatch closeSent = new CountDownLatch(1); passMessage.acquire(); conn._channel.getPipeline().addBefore("handler", "closeChannelDelay", new SimpleChannelHandler(){ @Override public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { closeSent.countDown(); passMessage.acquire(); try { super.channelClosed(ctx, e); } finally { passMessage.release(); } } }); final Channel serverChannel = getServerChannelForClientConn(conn); Thread asyncChannelClose = new Thread(new Runnable() { @Override public void run() { log.info("closing server channel"); serverChannel.close(); log.info("server channel: closed"); closeSent.countDown(); } }, "asyncChannelCloseThread"); asyncChannelClose.setDaemon(true); Thread asyncBootstrapReq = new Thread(new Runnable() { @Override public void run() { conn.requestStream("1", null, 10000, startScnCp, bstCallback); } }, "asyncBootstrapReqThread"); asyncBootstrapReq.setDaemon(true); log.info("simultaneously closing connection and sending /bootstrap request"); bstCallback.reset(); asyncChannelClose.start(); Assert.assertTrue(closeSent.await(1000, TimeUnit.MILLISECONDS)); TestUtil.assertWithBackoff(new ConditionCheck() { @Override public boolean check() { return !conn._channel.isConnected(); } }, "waiting for disconnect on the client side", 1000, log); Assert.assertEquals(AbstractNettyHttpConnection.State.CONNECTED, conn.getNetworkState()); log.info("asynchronously sending /bootstrap"); asyncBootstrapReq.start(); log.info("letting channelClose get through"); TestUtil.assertWithBackoff(new ConditionCheck() { @Override public boolean check() { return bstCallback.isStreamRequestError(); } }, "wait for streamRequestError callback", 1000, log); passMessage.release(); log.info("finished"); } finally { conn.close(); callback.shutdown(); log.info("cleaned"); } }