Method Create

src/brpc/stream.cpp:78–127

}

int Stream::Create(const StreamOptions &options, 
                   const StreamSettings *remote_settings,
                   StreamId *id, bool parse_rpc_response) {
    Stream* s = new Stream();
    s->_host_socket = NULL;
    s->_fake_socket_weak_ref = NULL;
    s->_connected = false;
    s->_options = options;
    s->_closed = false;
    s->_error_code = 0;
    s->_cur_buf_size = options.max_buf_size > 0 ? options.max_buf_size : 0;
    if (options.max_buf_size > 0 && options.min_buf_size > options.max_buf_size) {
        // set 0 if min_buf_size is invalid.
        s->_options.min_buf_size = 0;
        LOG(WARNING) << "options.min_buf_size is larger than options.max_buf_size, it will be set to 0.";
    }
    if (FLAGS_socket_max_streams_unconsumed_bytes > 0 && s->_options.min_buf_size > 0) {
        s->_cur_buf_size = s->_options.min_buf_size;
    }

    if (remote_settings != NULL) {
        s->_remote_settings.MergeFrom(*remote_settings);
    }
    s->_parse_rpc_response = parse_rpc_response;
    if (bthread_id_list_init(&s->_writable_wait_list, 8, 8/*FIXME*/)) {
        delete s;
        return -1;
    }
    bthread::ExecutionQueueOptions q_opt;
    q_opt.bthread_attr 
        = FLAGS_usercode_in_pthread ? BTHREAD_ATTR_PTHREAD : BTHREAD_ATTR_NORMAL;
    if (bthread::execution_queue_start(&s->_consumer_queue, &q_opt, Consume, s) != 0) {
        LOG(FATAL) << "Fail to create ExecutionQueue";
        delete s;
        return -1;
    }
    SocketOptions sock_opt;
    sock_opt.conn = s;
    SocketId fake_sock_id;
    if (Socket::Create(sock_opt, &fake_sock_id) != 0) {
        s->BeforeRecycle(NULL);
        return -1;
    }
    SocketUniquePtr ptr;
    CHECK_EQ(0, Socket::Address(fake_sock_id, &ptr));
    s->_fake_socket_weak_ref = ptr.get();
    s->_id = fake_sock_id;
    *id = s->id();
    return 0;
}

void Stream::BeforeRecycle(Socket *) {
    // No one holds reference now, so we don't need lock here