Function SendTwoMessagesOnServerExtraStream

test/brpc_streaming_rpc_unittest.cpp:134–188

};

static void* SendTwoMessagesOnServerExtraStream(void* arg) {
    auto* state = static_cast<BatchStreamFeedbackRaceState*>(arg);
    const brpc::StreamId sid = state->server_extra_stream_id;

    // Wait until server-side stream is connected.
    const int64_t connect_deadline_us = butil::gettimeofday_us() + 2 * 1000 * 1000L;
    bool connected = false;
    while (butil::gettimeofday_us() < connect_deadline_us) {
        brpc::SocketUniquePtr ptr;
        if (brpc::Socket::Address(sid, &ptr) == 0) {
            brpc::Stream* s = static_cast<brpc::Stream*>(ptr->conn());
            if (s->_host_socket != NULL && s->_connected) {
                connected = true;
                break;
            }
        }
        usleep(1000);
    }

    if (!connected) {
        state->server_first_write_rc.store(ETIMEDOUT, std::memory_order_relaxed);
        state->server_second_write_rc.store(ETIMEDOUT, std::memory_order_relaxed);
        state->server_write_done.store(true, std::memory_order_release);
        return NULL;
    }

    // 1) Send a payload exactly equal to max_buf_size(64).
    {
        std::string payload(64, 'a');
        butil::IOBuf out;
        out.append(payload);
        state->server_first_write_rc.store(brpc::StreamWrite(sid, out), std::memory_order_relaxed);
    }

    // 2) Then send another byte. This write should become writable only after
    // client sends FEEDBACK with consumed_size >= 64.
    const int64_t write_deadline_us = butil::gettimeofday_us() + 2 * 1000 * 1000L;
    int rc = -1;
    while (butil::gettimeofday_us() < write_deadline_us) {
        butil::IOBuf out;
        out.append("b", 1);
        rc = brpc::StreamWrite(sid, out);
        if (rc == 0) {
            break;
        }
        if (rc != EAGAIN) {
            break;
        }
        const timespec duetime = butil::milliseconds_from_now(100);
        (void)brpc::StreamWait(sid, &duetime);
    }
    state->server_second_write_rc.store(rc, std::memory_order_relaxed);
    state->server_write_done.store(true, std::memory_order_release);
    return NULL;
}

class MyServiceWithBatchStream : public test::EchoService {
public: