Fix in RTMP:

1. Call done directly in StartConnect to avoid race with OnConnected in StopConnect
2. Dangling pointer in OnServerStreamCreated
3. Invalid _chunk_stream_id in closeStream and deleteStream

src/brpc/policy/rtmp_protocol.cpp

@@ -3445,7 +3445,7 @@ public:
              AMFInputStream* istream, Socket* socket);
     void Cancel();
 private:
-    RtmpClientStream* _stream;
+    butil::intrusive_ptr<RtmpClientStream> _stream;
     CallId _call_id;
 };
 
@@ -3509,7 +3509,7 @@ void OnServerStreamCreated::Run(bool error,
         // to avoid the race between OnStreamCreationDone and a failed OnStatus,
         // because the former function runs in another bthread and may run later
         // than OnStatus which needs to see the stream.
-        if (!ctx->AddClientStream(_stream)) {
+        if (!ctx->AddClientStream(_stream.get())) {
             cntl->SetFailed(EINVAL, "Fail to add client stream_id=%u", stream_id);
             break;
         }

src/brpc/rtmp.cpp

@@ -1013,9 +1013,6 @@ void RtmpConnect::StartConnect(
         return done(EINVAL, data);
     }
 
-    // Save to callback to call when RTMP connect is done.
-    ctx->SetConnectCallback(done, data);
-
     const RtmpClientOptions* _client_options = ctx->client_options();
     if (_client_options && _client_options->simplified_rtmp) {
         ctx->set_simplified_rtmp(true);
@@ -1025,9 +1022,11 @@ void RtmpConnect::StartConnect(
         }
         ctx->SetState(s->remote_side(), policy::RtmpContext::STATE_RECEIVED_S2);
         ctx->set_create_stream_with_play_or_publish(true);
-        ctx->OnConnected(0);
-        return;
+        return done(0, data);
     }
+
+    // Save to callback to call when RTMP connect is done.
+    ctx->SetConnectCallback(done, data);
         
     // Initiate the rtmp handshake.
     bool is_simple_handshake = false;
@@ -1786,7 +1785,7 @@ void RtmpClientStream::OnStopInternal() {
         return CallOnStop();
     }
 
-    if (!_rtmpsock->Failed()) {
+    if (!_rtmpsock->Failed() && _chunk_stream_id != 0) {
         // SRS requires closeStream which is sent over this stream.
         butil::IOBuf req_buf1;
         {