wasphin commented on code in PR #3508:
URL: https://github.com/apache/brpc/pull/3508#discussion_r3915427470
##########
test/brpc_rtmp_unittest.cpp:
##########
@@ -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;
Review Comment:
RtmpMessageHeader has a default constructor that initializes timestamp,
message_length, and message_type to 0 and stream_id to
RTMP_CONTROL_MESSAGE_STREAM_ID, so this declaration is fully initialized.
##########
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;
+ }
+};
Review Comment:
Keeping this private is consistent with the existing PlayingDummyService and
PublishService test implementations in this file. NewStream() is public in the
RtmpService interface, and private access on the override does not affect
virtual dispatch through the base class.
##########
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;
Review Comment:
Added a deterministic regression test in ef067d8. It uses a service whose
NewStream() returns nullptr and an already failed socket so the
rejection-response write failure is exercised without relying on a
client-disconnect race.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
