From f2f48c4528d63d7bd45a3eb884d967e4114adb8f Mon Sep 17 00:00:00 2001
From: Wang Xiaofeng <wasphin@gmail.com>
Date: Sun, 30 Aug 2026 22:53:58 +0800
Subject: [PATCH 1/2] Handle rejected RTMP streams on response failure

Avoid accessing the stream failure callback when NewStream rejects a
createStream request and writing the rejection response also fails.
---
 src/brpc/policy/rtmp_protocol.cpp | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/src/brpc/policy/rtmp_protocol.cpp b/src/brpc/policy/rtmp_protocol.cpp
index 8ad63fb958..d72684b1ed 100644
--- a/src/brpc/policy/rtmp_protocol.cpp
+++ b/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;
     }

From ef067d89e302684e12a23995a641a626bd7f9ca2 Mon Sep 17 00:00:00 2001
From: Wang Xiaofeng <wasphin@gmail.com>
Date: Tue, 1 Sep 2026 22:56:13 +0800
Subject: [PATCH 2/2] Test rejected RTMP stream response failure

Cover the case where an RTMP service rejects stream creation and writing
the rejection response also fails. Use an already failed socket to trigger
the write error deterministically.
---
 test/brpc_rtmp_unittest.cpp | 45 +++++++++++++++++++++++++++++++++++++
 1 file changed, 45 insertions(+)

diff --git a/test/brpc_rtmp_unittest.cpp b/test/brpc_rtmp_unittest.cpp
index 9264a8e0f7..eb39f41796 100644
--- a/test/brpc_rtmp_unittest.cpp
+++ b/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;