This is an automated email from the ASF dual-hosted git repository.
wasphin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new 829f462e Handle rejected RTMP streams on response failure (#3508)
829f462e is described below
commit 829f462e7d55915e51e2d393f099fd219675223c
Author: Xiaofeng Wang <[email protected]>
AuthorDate: Thu Sep 3 13:29:54 2026 +0800
Handle rejected RTMP streams on response failure (#3508)
* 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.
* 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.
---
src/brpc/policy/rtmp_protocol.cpp | 8 ++++---
test/brpc_rtmp_unittest.cpp | 45 +++++++++++++++++++++++++++++++++++++++
2 files changed, 50 insertions(+), 3 deletions(-)
diff --git a/src/brpc/policy/rtmp_protocol.cpp
b/src/brpc/policy/rtmp_protocol.cpp
index 8ad63fb9..d72684b1 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;
}
diff --git a/test/brpc_rtmp_unittest.cpp b/test/brpc_rtmp_unittest.cpp
index 9264a8e0..eb39f417 100644
--- a/test/brpc_rtmp_unittest.cpp
+++ b/test/brpc_rtmp_unittest.cpp
@@ -355,6 +355,14 @@ private:
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;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
