Merge pull request #2571 from chenBright/opt_stream_overcrowded

Ignore eovercrowded for streaming control frame

Author: Huixxi

src/brpc/policy/streaming_rpc_protocol.cpp

@@ -134,7 +134,9 @@ void SendStreamRst(Socket *sock, int64_t remote_stream_id) {
     fm.set_frame_type(FRAME_TYPE_RST);
     butil::IOBuf out;
     PackStreamMessage(&out, fm, NULL);
-    sock->Write(&out);
+    Socket::WriteOptions wopt;
+    wopt.ignore_eovercrowded = true;
+    sock->Write(&out, &wopt);
 }
 
 void SendStreamClose(Socket *sock, int64_t remote_stream_id,
@@ -146,19 +148,24 @@ void SendStreamClose(Socket *sock, int64_t remote_stream_id,
     fm.set_frame_type(FRAME_TYPE_CLOSE);
     butil::IOBuf out;
     PackStreamMessage(&out, fm, NULL);
-    sock->Write(&out);
+    Socket::WriteOptions wopt;
+    wopt.ignore_eovercrowded = true;
+    sock->Write(&out, &wopt);
 }
 
 int SendStreamData(Socket* sock, const butil::IOBuf* data,
                    int64_t remote_stream_id, int64_t source_stream_id) {
+    CHECK(sock != NULL);
     StreamFrameMeta fm;
     fm.set_stream_id(remote_stream_id);
     fm.set_source_stream_id(source_stream_id);
     fm.set_frame_type(FRAME_TYPE_DATA);
     fm.set_has_continuation(false);
     butil::IOBuf out;
     PackStreamMessage(&out, fm, data);
-    return sock->Write(&out);
+    Socket::WriteOptions wopt;
+    wopt.ignore_eovercrowded = true;
+    return sock->Write(&out, &wopt);
 }
 
 }  // namespace policy

src/brpc/stream.cpp

@@ -517,7 +517,7 @@ class MessageBatcher {
         _total_length += buf->length();
 
     }
-    size_t total_length() { return _total_length; }
+    size_t total_length() const { return _total_length; }
 private:
     butil::IOBuf** _storage;
     size_t _cap;