Handle rejected RTMP streams on response failure

src/brpc/policy/rtmp_protocol.cpp

@@ -2714,9 +2714,11 @@ bool RtmpChunkStream::OnCreateStream(const RtmpMessageHeader& mh,
         PLOG(WARNING) << socket->remote_side() << '[' << mh.stream_id
                       << "] Fail to respond createStream";
         // End the stream at server-side.
-        const bthread_id_t id = stream->_onfail_id;
-        if (id != INVALID_BTHREAD_ID) {
-            bthread_id_error(id, 0);
+        if (stream != nullptr) {
+            const bthread_id_t id = stream->_onfail_id;
+            if (id != INVALID_BTHREAD_ID) {
+                bthread_id_error(id, 0);
+            }
         }
         return false;
     }

test/brpc_rtmp_unittest.cpp

@@ -355,6 +355,14 @@ class PlayingDummyService : public brpc::RtmpService {
     int64_t _sleep_ms;
 };
 
+class RejectingRtmpService : public brpc::RtmpService {
+private:
+    brpc::RtmpServerStream* NewStream(
+        const brpc::RtmpConnectRequest&) override {
+        return nullptr;
+    }
+};
+
 class PublishStream : public brpc::RtmpServerStream {
 public:
     PublishStream(int64_t sleep_ms)
@@ -967,6 +975,43 @@ TEST(RtmpTest, abort_message_naming_own_chunk_stream) {
     ASSERT_EQ(brpc::PARSE_OK, ctx.Feed(&buf2, sock.get()).error());
 }
 
+TEST(RtmpTest, rejected_create_stream_with_response_write_failure) {
+    RejectingRtmpService rtmp_service;
+    brpc::Server server;
+    brpc::ServerOptions server_options;
+    server_options.rtmp_service = &rtmp_service;
+    ASSERT_EQ(0, server.Start(0, &server_options));
+
+    brpc::SocketId socket_id;
+    brpc::SocketOptions socket_options;
+    ASSERT_EQ(0, brpc::Socket::Create(socket_options, &socket_id));
+    brpc::SocketUniquePtr socket;
+    ASSERT_EQ(0, brpc::Socket::Address(socket_id, &socket));
+    ASSERT_EQ(0, socket->SetFailed());
+
+    brpc::policy::RtmpContext ctx(nullptr, &server);
+    brpc::policy::RtmpChunkStream chunk_stream(
+        &ctx, brpc::policy::RTMP_CONTROL_CHUNK_STREAM_ID);
+
+    std::string request;
+    google::protobuf::io::StringOutputStream zc_stream(&request);
+    brpc::AMFOutputStream ostream(&zc_stream);
+    brpc::WriteAMFNumber(1, &ostream);
+    brpc::AMFObject command;
+    brpc::WriteAMFObject(command, &ostream);
+    ASSERT_TRUE(ostream.good());
+
+    google::protobuf::io::ArrayInputStream input(
+        request.data(), request.size());
+    brpc::AMFInputStream istream(&input);
+    brpc::policy::RtmpMessageHeader message_header;
+    ASSERT_FALSE(chunk_stream.OnCreateStream(
+        message_header, &istream, socket.get()));
+
+    server.Stop(0);
+    server.Join();
+}
+
 TEST(RtmpTest, successfully_play_streams) {
     PlayingDummyService rtmp_service;
     brpc::Server server;