* Move the place of DestroyStreamCreator from Call::OnComplete to Controller::EndRPC

* Delete stream_creater() and make set_stream_creator can be called only once

src/brpc/controller.cpp

@@ -775,9 +775,7 @@ void Controller::Call::OnComplete(
         stream_user_data->DestroyStreamUserData(sending_sock, c, error_code, end_of_rpc);
         stream_user_data = NULL;
     }
-    if (end_of_rpc && c->stream_creator()) {
-        c->stream_creator()->DestroyStreamCreator(c);
-    }
+
     // Release the `Socket' we used to send/receive data
     sending_sock.reset(NULL);
 }
@@ -838,7 +836,9 @@ void Controller::EndRPC(const CompletionInfo& info) {
             CHECK(false) << "A previous non-backed-up call responded";
         }
     }
-    
+    if (_stream_creator) {
+        _stream_creator->DestroyStreamCreator(this);
+    }
     // Clear _error_text when the call succeeded, otherwise a successful
     // call with non-empty ErrorText may confuse user.
     if (!_error_code) {
@@ -1322,6 +1322,14 @@ void Controller::reset_rpc_dump_meta(RpcDumpMeta* meta) {
     _rpc_dump_meta = meta;
 }
 
+void Controller::set_stream_creator(StreamCreator* sc) {
+    if (_stream_creator) {
+        LOG(FATAL) << "A StreamCreator has been set previously";
+        return;
+    }
+    _stream_creator = sc;
+}
+
 ProgressiveAttachment*
 Controller::CreateProgressiveAttachment(StopStyle stop_style) {
     if (has_progressive_writer()) {

src/brpc/controller.h

@@ -248,10 +248,10 @@ public:
     void reset_rpc_dump_meta(RpcDumpMeta* meta);
     const RpcDumpMeta* rpc_dump_meta() { return _rpc_dump_meta; }
 
-    // Attach a StreamCreator to this RPC. Notice that controller never deletes
-    // the StreamCreator, you can do the deletion inside OnDestroyingStream.
-    void set_stream_creator(StreamCreator* sc) { _stream_creator = sc; }
-    StreamCreator* stream_creator() const { return _stream_creator; }
+    // Attach a StreamCreator to this RPC. Notice that the ownership of sc has
+    // been transferred to cntl, and sc->DestroyStreamCreator() would be called
+    // only once to destroy sc.
+    void set_stream_creator(StreamCreator* sc);
 
     // Make the RPC end when the HTTP response has complete headers and let
     // user read the remaining body by using ReadProgressiveAttachmentBy().

src/brpc/rtmp.cpp

@@ -1748,7 +1748,7 @@ void RtmpClientStream::OnFailedToCreateStream() {
 
 void RtmpClientStream::DestroyStreamUserData(SocketUniquePtr& sending_sock,
                                              Controller* cntl,
-                                             int error_code,
+                                             int /*error_code*/,
                                              bool end_of_rpc) {
     if (!end_of_rpc) {
         if (sending_sock) {
@@ -1760,7 +1760,7 @@ void RtmpClientStream::DestroyStreamUserData(SocketUniquePtr& sending_sock,
             }
         }
     } else {
-        // Always move sending_sock into _rtmpsock.
+        // Always move sending_sock into _rtmpsock at the end of rpc.
         // - If the RPC is successful, moving sending_sock prevents it from
         //   setfailed in Controller after calling this method.
         // - If the RPC is failed, OnStopInternal() can clean up the socket_map