Function TEST_P

src/test/msgr/test_async_networkstack.cc:161–263

};

TEST_P(NetworkWorkerTest, SimpleTest) {
  entity_addr_t bind_addr;
  ASSERT_TRUE(bind_addr.parse(get_addr().c_str()));
  std::atomic_bool accepted(false);
  std::atomic_bool *accepted_p = &accepted;

  exec_events([this, accepted_p, bind_addr](Worker *worker) mutable {
    entity_addr_t cli_addr;
    SocketOptions options;
    ServerSocket bind_socket;
    EventCenter *center = &worker->center;
    ssize_t r = 0;
    if (stack->support_local_listen_table() || worker->id == 0)
      r = worker->listen(bind_addr, 0, options, &bind_socket);
    ASSERT_EQ(0, r);

    ConnectedSocket cli_socket, srv_socket;
    if (worker->id == 0) {
      r = worker->connect(bind_addr, options, &cli_socket);
      ASSERT_EQ(0, r);
    }

    bool is_my_accept = false;
    if (bind_socket) {
      C_poll cb(center);
      center->create_file_event(bind_socket.fd(), EVENT_READABLE, &cb);
      if (cb.poll(500)) {
        *accepted_p = true;
        is_my_accept = true;
      }
      ASSERT_TRUE(*accepted_p);
      center->delete_file_event(bind_socket.fd(), EVENT_READABLE);
    }

    if (is_my_accept) {
      r = bind_socket.accept(&srv_socket, options, &cli_addr, worker);
      ASSERT_EQ(0, r);
      ASSERT_TRUE(srv_socket.fd() > 0);
    }

    if (worker->id == 0) {
      C_poll cb(center);
      center->create_file_event(cli_socket.fd(), EVENT_READABLE, &cb);
      r = cli_socket.is_connected();
      if (r == 0) {
        ASSERT_EQ(true, cb.poll(500));
        r = cli_socket.is_connected();
      }
      ASSERT_EQ(1, r);
      center->delete_file_event(cli_socket.fd(), EVENT_READABLE);
    }

    const char *message = "this is a new message";
    int len = strlen(message);
    bufferlist bl;
    bl.append(message, len);
    if (worker->id == 0) {
      r = cli_socket.send(bl, false);