@Test
/**
 * This is a unit test for DDSDBUS-3537. There is a lag between a network channel disconnect and the
 * state change in AbstractNettyHttpConnection. This can cause a race condition in various requestXXX objects which
 * check the state of the connection using the network channel. As a result, they may attempt to reconnect while
 * AbstractNettyHttpConnection is still in CONNECTED state which causes an error for an incorrect transition
 * CONNECTED -> CONNECTING.
 *
 *  The test simulates the above condition by injecting a handler in the client pipeline which artificially holds up
 *  the channelClosed message. As a result we can inject a request while the netty channel is disconnected but the
 *  AbstractNettyHttpConnection object has not detected this yet.
 */
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());
    //wait for the response
    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");
  }
}