Add cntl param back to DestroyStreamUserData

src/brpc/controller.cpp

@@ -697,7 +697,7 @@ inline bool does_error_affect_main_socket(int error_code) {
 void Controller::Call::OnComplete(
         Controller* c, int error_code/*note*/, bool responded, bool end_of_rpc) {
     if (stream_user_data) {
-        stream_user_data->DestroyStreamUserData(sending_sock, error_code, end_of_rpc);
+        stream_user_data->DestroyStreamUserData(sending_sock, c, error_code, end_of_rpc);
         stream_user_data = NULL;
     }
 

src/brpc/policy/http2_rpc_protocol.cpp

@@ -1448,10 +1448,12 @@ private:
 };
 
 void H2UnsentRequest::DestroyStreamUserData(SocketUniquePtr& sending_sock,
+                                            Controller* cntl,
                                             int error_code,
                                             bool /*end_of_rpc*/) {
     RemoveRefOnQuit deref_self(this);
     if (sending_sock != NULL && error_code != 0) {
+        CHECK_EQ(cntl, _cntl);
         std::unique_lock<butil::Mutex> mu(_mutex);
         _cntl = NULL;
         if (_stream_id != 0) {

src/brpc/policy/http2_rpc_protocol.h

@@ -153,6 +153,7 @@ public:
 
     // @StreamUserData
     void DestroyStreamUserData(SocketUniquePtr& sending_sock,
+                               Controller* cntl,
                                int error_code,
                                bool end_of_rpc) override;
 

src/brpc/rtmp.cpp

@@ -1747,6 +1747,7 @@ void RtmpClientStream::OnFailedToCreateStream() {
 }
 
 void RtmpClientStream::DestroyStreamUserData(SocketUniquePtr& sending_sock,
+                                             Controller* cntl,
                                              int /*error_code*/,
                                              bool end_of_rpc) {
     if (!end_of_rpc) {

src/brpc/rtmp.h

@@ -827,6 +827,7 @@ friend class RtmpRetryingClientStream;
 
     // @StreamUserData
     void DestroyStreamUserData(SocketUniquePtr& sending_sock,
+                               Controller* cntl,
                                int error_code,
                                bool end_of_rpc) override;
 

src/brpc/stream_creator.h

@@ -66,9 +66,11 @@ public:
     // Params:
     //   sending_sock: The socket chosen by OnCreatingStream(), if an error
     //     happens during choosing, the enclosed socket is NULL.
-    //   error_code: the error code after the RPC.
+    //   cntl: contexts of the RPC
+    //   error_code: error code after the RPC.
     //   end_of_rpc: true if the RPC is about to destroyed.
     virtual void DestroyStreamUserData(SocketUniquePtr& sending_sock,
+                                       Controller* cntl,
                                        int error_code,
                                        bool end_of_rpc) = 0;
 };

test/brpc_http_rpc_protocol_unittest.cpp

@@ -1113,7 +1113,7 @@ TEST_F(HttpTest, http2_window_used_up) {
         } else {
             ASSERT_TRUE(st.ok());
         }
-        h2_req->DestroyStreamUserData(_h2_client_sock, 0, false);
+        h2_req->DestroyStreamUserData(_h2_client_sock, &cntl, 0, false);
     }
 }