TEST_P(MessengerTest, StatefulTest) {
  Message *m;
  FakeDispatcher cli_dispatcher(false), srv_dispatcher(true);
  entity_addr_t bind_addr;
  bind_addr.parse("127.0.0.1");
  Messenger::Policy p = Messenger::Policy::stateful_server(0);
  server_msgr->set_policy(entity_name_t::TYPE_CLIENT, p);
  p = Messenger::Policy::lossless_client(0);
  client_msgr->set_policy(entity_name_t::TYPE_OSD, p);

  server_msgr->bind(bind_addr);
  server_msgr->add_dispatcher_head(&srv_dispatcher);
  server_msgr->start();
  client_msgr->add_dispatcher_head(&cli_dispatcher);
  client_msgr->start();

  // 1. test for server standby
  ConnectionRef conn = client_msgr->get_connection(server_msgr->get_myinst());
  {
    m = new MPing();
    ASSERT_EQ(conn->send_message(m), 0);
    Mutex::Locker l(cli_dispatcher.lock);
    while (!cli_dispatcher.got_new)
      cli_dispatcher.cond.Wait(cli_dispatcher.lock);
    cli_dispatcher.got_new = false;
  }
  ASSERT_TRUE(static_cast<Session*>(conn->get_priv())->get_count() == 1);
  conn->mark_down();
  ASSERT_FALSE(conn->is_connected());
  ConnectionRef server_conn = server_msgr->get_connection(client_msgr->get_myinst());
  // don't lose state
  ASSERT_TRUE(static_cast<Session*>(server_conn->get_priv())->get_count() == 1);

  srv_dispatcher.got_new = false;
  conn = client_msgr->get_connection(server_msgr->get_myinst());
  {
    m = new MPing();
    ASSERT_EQ(conn->send_message(m), 0);
    Mutex::Locker l(cli_dispatcher.lock);
    while (!cli_dispatcher.got_new)
      cli_dispatcher.cond.Wait(cli_dispatcher.lock);
    cli_dispatcher.got_new = false;
  }
  ASSERT_TRUE(static_cast<Session*>(conn->get_priv())->get_count() == 1);
  server_conn = server_msgr->get_connection(client_msgr->get_myinst());
  {
    Mutex::Locker l(srv_dispatcher.lock);
    while (!srv_dispatcher.got_remote_reset)
      srv_dispatcher.cond.Wait(srv_dispatcher.lock);
  }

  // 2. test for client reconnect
  ASSERT_FALSE(cli_dispatcher.got_remote_reset);
  cli_dispatcher.got_connect = false;
  cli_dispatcher.got_new = false;
  cli_dispatcher.got_remote_reset = false;
  server_conn->mark_down();
  ASSERT_FALSE(server_conn->is_connected());
  // ensure client detect server socket closed
  {
    Mutex::Locker l(cli_dispatcher.lock);
    while (!cli_dispatcher.got_remote_reset)
      cli_dispatcher.cond.Wait(cli_dispatcher.lock);
    cli_dispatcher.got_remote_reset = false;
  }
  {
    Mutex::Locker l(cli_dispatcher.lock);
    while (!cli_dispatcher.got_connect)
      cli_dispatcher.cond.Wait(cli_dispatcher.lock);
    cli_dispatcher.got_connect = false;
  }
  CHECK_AND_WAIT_TRUE(conn->is_connected());
  ASSERT_TRUE(conn->is_connected());

  {
    m = new MPing();
    ASSERT_EQ(conn->send_message(m), 0);
    ASSERT_TRUE(conn->is_connected());
    Mutex::Locker l(cli_dispatcher.lock);
    while (!cli_dispatcher.got_new)
      cli_dispatcher.cond.Wait(cli_dispatcher.lock);
    cli_dispatcher.got_new = false;
  }
  // resetcheck happen
  ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv())->get_count());
  server_conn = server_msgr->get_connection(client_msgr->get_myinst());
  ASSERT_EQ(1U, static_cast<Session*>(server_conn->get_priv())->get_count());
  cli_dispatcher.got_remote_reset = false;

  server_msgr->shutdown();
  client_msgr->shutdown();
  server_msgr->wait();
  client_msgr->wait();
}
TEST_P(MessengerTest, StatefulTest) {
  Message *m;
  FakeDispatcher cli_dispatcher(false), srv_dispatcher(true);
  entity_addr_t bind_addr;
  bind_addr.parse("127.0.0.1");
  Messenger::Policy p = Messenger::Policy::stateful_server(0);
  server_msgr->set_policy(entity_name_t::TYPE_CLIENT, p);
  p = Messenger::Policy::lossless_client(0);
  client_msgr->set_policy(entity_name_t::TYPE_OSD, p);

  server_msgr->bind(bind_addr);
  server_msgr->add_dispatcher_head(&srv_dispatcher);
  server_msgr->start();
  client_msgr->add_dispatcher_head(&cli_dispatcher);
  client_msgr->start();

   
  ConnectionRef conn = client_msgr->get_connection(server_msgr->get_myinst());
  {
    m = new MPing();
    ASSERT_EQ(conn->send_message(m), 0);
    Mutex::Locker l(cli_dispatcher.lock);
    while (!cli_dispatcher.got_new)
      cli_dispatcher.cond.Wait(cli_dispatcher.lock);
    cli_dispatcher.got_new = false;
  }
  ASSERT_TRUE(static_cast<Session*>(conn->get_priv())->get_count() == 1);
  conn->mark_down();
  ASSERT_FALSE(conn->is_connected());
  ConnectionRef server_conn = server_msgr->get_connection(client_msgr->get_myinst());
   
  ASSERT_TRUE(static_cast<Session*>(server_conn->get_priv())->get_count() == 1);

  srv_dispatcher.got_new = false;
  conn = client_msgr->get_connection(server_msgr->get_myinst());
  {
    m = new MPing();
    ASSERT_EQ(conn->send_message(m), 0);
    Mutex::Locker l(cli_dispatcher.lock);
    while (!cli_dispatcher.got_new)
      cli_dispatcher.cond.Wait(cli_dispatcher.lock);
    cli_dispatcher.got_new = false;
  }
  ASSERT_TRUE(static_cast<Session*>(conn->get_priv())->get_count() == 1);
  server_conn = server_msgr->get_connection(client_msgr->get_myinst());
  {
    Mutex::Locker l(srv_dispatcher.lock);
    while (!srv_dispatcher.got_remote_reset)
      srv_dispatcher.cond.Wait(srv_dispatcher.lock);
  }

   
  ASSERT_FALSE(cli_dispatcher.got_remote_reset);
  cli_dispatcher.got_connect = false;
  cli_dispatcher.got_new = false;
  cli_dispatcher.got_remote_reset = false;
  server_conn->mark_down();
  ASSERT_FALSE(server_conn->is_connected());
   
  {
    Mutex::Locker l(cli_dispatcher.lock);
    while (!cli_dispatcher.got_remote_reset)
      cli_dispatcher.cond.Wait(cli_dispatcher.lock);
    cli_dispatcher.got_remote_reset = false;
  }
  {
    Mutex::Locker l(cli_dispatcher.lock);
    while (!cli_dispatcher.got_connect)
      cli_dispatcher.cond.Wait(cli_dispatcher.lock);
    cli_dispatcher.got_connect = false;
  }
  CHECK_AND_WAIT_TRUE(conn->is_connected());
  ASSERT_TRUE(conn->is_connected());

  {
    m = new MPing();
    ASSERT_EQ(conn->send_message(m), 0);
    ASSERT_TRUE(conn->is_connected());
    Mutex::Locker l(cli_dispatcher.lock);
    while (!cli_dispatcher.got_new)
      cli_dispatcher.cond.Wait(cli_dispatcher.lock);
    cli_dispatcher.got_new = false;
  }
   
  ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv())->get_count());
  server_conn = server_msgr->get_connection(client_msgr->get_myinst());
  ASSERT_EQ(1U, static_cast<Session*>(server_conn->get_priv())->get_count());
  cli_dispatcher.got_remote_reset = false;

  server_msgr->shutdown();
  client_msgr->shutdown();
  server_msgr->wait();
  client_msgr->wait();
}