概述
在一些应用场景中，client 或 server 需要向对面发送大量数据，这些数据非常大或者持续地在产生以至于无法放在一个 RPC 的附件中。比如一个分布式系统的不同节点间传递 replica 或 snapshot。client/server 之间虽然可以通过多次 RPC 把数据切分后传输过去，但存在如下问题：
- 如果这些 RPC 是并行的，无法保证接收端有序地收到数据，拼接数据的逻辑相当复杂。
- 如果这些 RPC 是串行的，每次传递都得等待一次网络 RTT + 处理数据的延时，特别是后者的延时可能是难以预估的。
为了让大块数据以流水线的方式在 client/server 之间传递，我们提供了 Streaming RPC 这种交互模型。Streaming RPC 让用户能够在 client/service 之间建立用户态连接，称为 Stream，同一个 TCP 连接之上能同时存在多个 Stream。Stream 的传输数据以消息为基本单位，输入端可以源源不断地往 Stream 中写入消息，接收端会按输入端写入顺序收到消息。
Streaming RPC 保证：
- 有消息边界。
- 接收消息的顺序和发送消息的顺序严格一致。
- 全双工。
- 支持流控。
- 提供超时提醒。
目前的实现还没有自动切割过大的消息，同一个 TCP 连接上的多个 Stream 之间可能有 Head-of-line blocking 问题，请尽量避免过大的单个消息，实现自动切割后我们会告知并更新文档。
例子见 example/streaming_echo_c++。
建立 Stream
目前 Stream 都由 Client 端建立。Client 先在本地创建一个 Stream，再通过一次 RPC（必须使用 baidu_std 协议）与指定的 Service 建立一个 Stream，如果 Service 在收到请求之后选择接受这个 Stream，那在 response 返回 Client 后 Stream 就会建立成功。过程中的任何错误都把 RPC 标记为失败，同时也意味着 Stream 创建失败。用 linux 下建立连接的过程打比方，Client 先创建 socket（创建 Stream），再调用 connect 尝试与远端建立连接（通过 RPC 建立 Stream），远端 accept 后连接就建立了（service 接受后创建成功）。
注意：如果 Client 尝试向不支持 Streaming RPC 的老 Server 建立 Stream，将总是失败。
程序中我们用 StreamId 代表一个 Stream，对 Stream 的读写、关闭操作都将作用在这个 Id 上。
struct StreamOptions
// The max size of unconsumed data allowed at remote side.
// If |max_buf_size| <= 0, there's no limit of buf size
// default: 2097152 (2M)
int max_buf_size;
// Notify user when there's no data for at least |idle_timeout_ms|
// milliseconds since the last time that on_received_messages or on_idle_timeout
// finished.
// default: -1
long idle_timeout_ms;
// Maximum messages in batch passed to handler->on_received_messages
// default: 128
size_t messages_in_batch;
// Handle input message, if handler is NULL, the remote side is not allowd to
// write any message, who will get EBADF on writting
// default: NULL
StreamInputHandler* handler;
};
// [Called at the client side]
// Create a Stream at client-side along with the |cntl|, which will be connected
// when receiving the response with a Stream from server-side. If |options| is
// NULL, the Stream will be created with default options
// Return 0 on success, -1 otherwise
int StreamCreate(StreamId* request_stream, Controller &cntl, const StreamOptions* options);接受 Stream
如果 client 在 RPC 上附带了一个 Stream，service 在收到 RPC 后可以通过调用 StreamAccept 接受。接受后 Server 端对应产生的 Stream 存放在 response_stream 中，Server 可通过这个 Stream 向 Client 发送数据。
// [Called at the server side]
// Accept the Stream. If client didn't create a Stream with the request
// (cntl.has_remote_stream() returns false), this method would fail.
// Return 0 on success, -1 otherwise.
int StreamAccept(StreamId* response_stream, Controller &cntl, const StreamOptions* options);读取 Stream
在建立或者接受一个 Stream 的时候，用户可以继承 StreamInputHandler 并把这个 handler 填入 StreamOptions 中。通过这个 handler，用户可以处理对端的写入数据、连接关闭以及 idle timeout。
class StreamInputHandler {
public:
// 当接收到消息后被调用
virtual int on_received_messages(StreamId id, butil::IOBuf *const messages[], size_t size) = 0;
// 当Stream上长时间没有数据交互后被调用
virtual void on_idle_timeout(StreamId id) = 0;
// 当Stream被关闭时被调用
virtual void on_closed(StreamId id) = 0;
};第一次收到请求的时间
在 client 端，如果建立过程是一次同步 RPC，那在等待的线程被唤醒之后，on_received_message 就可能会被调用到。如果是异步 RPC 请求，那等到这次请求的 done->Run() 执行完毕之后，on_received_message 就可能会被调用。
在 server 端，当框架传入的 done->Run() 被调用完之后，on_received_message 就可能会被调用。
写入 Stream
// Write |message| into |stream_id|. The remote-side handler will received the
// message by the written order
// Returns 0 on success, errno otherwise
// Errno:
// - EAGAIN: |stream_id| is created with positive |max_buf_size| and buf size
// which the remote side hasn't consumed yet excceeds the number.
// - EINVAL: |stream_id| is invalied or has been closed
int StreamWrite(StreamId stream_id, const butil::IOBuf &message);流控
当存在较多已发送但未接收的数据时，发送端的 Write 操作会立即失败（返回 EAGAIN），这时候可以通过同步或异步的方式等待对端消费掉数据。
// Wait util the pending buffer size is less than |max_buf_size| or error occurs
// Returns 0 on success, errno otherwise
// Errno:
// - ETIMEDOUT: when |due_time| is not NULL and time expired this
// - EINVAL: the Stream was close during waiting
int StreamWait(StreamId stream_id, const timespec* due_time);
// Async wait
void StreamWait(StreamId stream_id, const timespec *due_time,
void (*on_writable)(StreamId stream_id, void* arg, int error_code),
void *arg);关闭 Stream
// Close |stream_id|, after this function is called:
// - All the following |StreamWrite| would fail
// - |StreamWait| wakes up immediately.
// - Both sides |on_closed| would be notifed after all the pending buffers have
// been received
// This function could be called multiple times without side-effects
int StreamClose(StreamId stream_id);
