Function TestShutdownWrite

test/brpc_socket_unittest.cpp:1615–1675

}

void TestShutdownWrite() {
    const size_t REP = 100;
    int fds[2];
    ASSERT_EQ(0, socketpair(AF_UNIX, SOCK_STREAM, 0, fds));

    brpc::SocketId id = 8888;
    butil::EndPoint dummy;
    ASSERT_EQ(0, str2endpoint("192.168.1.26:8080", &dummy));
    brpc::SocketOptions options;
    options.fd = fds[1];
    options.remote_side = dummy;
    options.user = new CheckRecycle;
    ASSERT_EQ(0, brpc::Socket::Create(options, &id));
    brpc::SocketUniquePtr s;
    ASSERT_EQ(0, brpc::Socket::Address(id, &s));
    s->_ssl_state = brpc::SSL_OFF;
    ASSERT_EQ(2, brpc::NRefOfVRef(s->_versioned_ref));
    global_sock = s.get();
    ASSERT_TRUE(s.get());
    ASSERT_EQ(fds[1], s->fd());
    ASSERT_EQ(dummy, s->remote_side());
    ASSERT_EQ(id, s->id());
    ASSERT_FALSE(s->IsWriteShutdown());

    pthread_t rth;
    ReaderArg reader_arg = { fds[0], 0 };
    pthread_create(&rth, NULL, reader, &reader_arg);

    bthread_t th[3];
    ShutdownWriterArg args[ARRAY_SIZE(th)];
    for (size_t i = 0; i < ARRAY_SIZE(th); ++i) {
        args[i].times = REP;
        args[i].socket_id = id;
        args[i].total_count = 0;
        args[i].success_count = 0;
        bthread_start_background(&th[i], NULL, ShutdownWriter, &args[i]);
    }

    for (size_t i = 0; i < ARRAY_SIZE(th); ++i) {
        ASSERT_EQ(0, bthread_join(th[i], NULL));
    }
    bthread_usleep(50 * 1000);

    ASSERT_TRUE(s->IsWriteShutdown());
    ASSERT_FALSE(s->Failed());
    ASSERT_EQ(0, s->SetFailed());
    s.release()->Dereference();
    pthread_join(rth, NULL);
    ASSERT_EQ((brpc::Socket*)NULL, global_sock);
    close(fds[0]);

    size_t total_count = 0;
    size_t success_count = 0;
    for (auto & arg : args) {
        total_count += arg.total_count;
        success_count += arg.success_count;
    }
    ASSERT_EQ(REP * ARRAY_SIZE(th), total_count);