admin/webrtc_m130
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rewrite PeerConnectionMessageHandler to not use rtc::MessageHandler

Browse Source 
Bug: webrtc:9702
Change-Id: I92390262b4794b1061702663621a9a4db22d367f
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/272023
Reviewed-by: Henrik Boström <hbos@webrtc.org>
Commit-Queue: Danil Chapovalov <danilchap@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#37836}
This commit is contained in:
Danil Chapovalov 2022-08-17 14:58:40 +02:00 committed by WebRTC LUCI CQ
parent 2cfc1af78a
commit 5d37ba29de
4 changed files with 38 additions and 146 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
pc/BUILD.gn
View File

@ -1525,10 +1525,9 @@ rtc_library("peer_connection_message_handler") {
"../api:rtc_error",
"../api:scoped_refptr",
"../api:sequence_checker",
"../rtc_base",
"../api/task_queue",
"../api/task_queue:pending_task_safety_flag",
"../rtc_base:checks",
"../rtc_base:location",
"../rtc_base:threading",
]
}

153
pc/peer_connection_message_handler.cc
View File

@ -10,7 +10,6 @@
#include "pc/peer_connection_message_handler.h"
#include <list>
#include <utility>
#include "api/jsep.h"
@ -19,164 +18,68 @@
#include "api/scoped_refptr.h"
#include "api/sequence_checker.h"
#include "api/stats_types.h"
#include "api/task_queue/pending_task_safety_flag.h"
#include "pc/legacy_stats_collector_interface.h"
#include "rtc_base/checks.h"
#include "rtc_base/location.h"
namespace webrtc {
namespace {
enum {
MSG_SET_SESSIONDESCRIPTION_SUCCESS = 0,
MSG_SET_SESSIONDESCRIPTION_FAILED,
MSG_CREATE_SESSIONDESCRIPTION_FAILED,
MSG_GETSTATS,
MSG_REPORT_USAGE_PATTERN,
};
struct SetSessionDescriptionMsg : public rtc::MessageData {
explicit SetSessionDescriptionMsg(
webrtc::SetSessionDescriptionObserver* observer)
: observer(observer) {}
rtc::scoped_refptr<webrtc::SetSessionDescriptionObserver> observer;
RTCError error;
};
struct CreateSessionDescriptionMsg : public rtc::MessageData {
explicit CreateSessionDescriptionMsg(
webrtc::CreateSessionDescriptionObserver* observer)
: observer(observer) {}
rtc::scoped_refptr<webrtc::CreateSessionDescriptionObserver> observer;
RTCError error;
};
struct LegacyGetStatsMsg : public rtc::MessageData {
LegacyGetStatsMsg(webrtc::StatsObserver* observer,
LegacyStatsCollectorInterface* stats,
webrtc::MediaStreamTrackInterface* track)
: observer(observer), stats(stats), track(track) {}
rtc::scoped_refptr<webrtc::StatsObserver> observer;
LegacyStatsCollectorInterface* stats;
rtc::scoped_refptr<webrtc::MediaStreamTrackInterface> track;
};
struct RequestUsagePatternMsg : public rtc::MessageData {
explicit RequestUsagePatternMsg(std::function<void()> func)
: function(func) {}
std::function<void()> function;
};
template <typename T>
rtc::scoped_refptr<T> WrapScoped(T* ptr) {
return rtc::scoped_refptr<T>(ptr);
}
} // namespace
PeerConnectionMessageHandler::~PeerConnectionMessageHandler() {
// Process all pending notifications in the message queue. If we don't do
// this, requests will linger and not know they succeeded or failed.
rtc::MessageList list;
signaling_thread()->Clear(this, rtc::MQID_ANY, &list);
for (auto& msg : list) {
if (msg.message_id == MSG_CREATE_SESSIONDESCRIPTION_FAILED) {
// Processing CreateOffer() and CreateAnswer() messages ensures their
// observers are invoked even if the PeerConnection is destroyed early.
OnMessage(&msg);
} else {
// TODO(hbos): Consider processing all pending messages. This would mean
// that SetLocalDescription() and SetRemoteDescription() observers are
// informed of successes and failures; this is currently NOT the case.
delete msg.pdata;
}
}
}
void PeerConnectionMessageHandler::OnMessage(rtc::Message* msg) {
RTC_DCHECK_RUN_ON(signaling_thread());
switch (msg->message_id) {
case MSG_SET_SESSIONDESCRIPTION_SUCCESS: {
SetSessionDescriptionMsg* param =
static_cast<SetSessionDescriptionMsg*>(msg->pdata);
param->observer->OnSuccess();
delete param;
break;
}
case MSG_SET_SESSIONDESCRIPTION_FAILED: {
SetSessionDescriptionMsg* param =
static_cast<SetSessionDescriptionMsg*>(msg->pdata);
param->observer->OnFailure(std::move(param->error));
delete param;
break;
}
case MSG_CREATE_SESSIONDESCRIPTION_FAILED: {
CreateSessionDescriptionMsg* param =
static_cast<CreateSessionDescriptionMsg*>(msg->pdata);
param->observer->OnFailure(std::move(param->error));
delete param;
break;
}
case MSG_GETSTATS: {
LegacyGetStatsMsg* param = static_cast<LegacyGetStatsMsg*>(msg->pdata);
StatsReports reports;
param->stats->GetStats(param->track.get(), &reports);
param->observer->OnComplete(reports);
delete param;
break;
}
case MSG_REPORT_USAGE_PATTERN: {
RequestUsagePatternMsg* param =
static_cast<RequestUsagePatternMsg*>(msg->pdata);
param->function();
delete param;
break;
}
default:
RTC_DCHECK_NOTREACHED() << "Not implemented";
break;
}
}
void PeerConnectionMessageHandler::PostSetSessionDescriptionSuccess(
SetSessionDescriptionObserver* observer) {
SetSessionDescriptionMsg* msg = new SetSessionDescriptionMsg(observer);
signaling_thread()->Post(RTC_FROM_HERE, this,
MSG_SET_SESSIONDESCRIPTION_SUCCESS, msg);
signaling_thread_->PostTask(
SafeTask(safety_.flag(),
[observer = WrapScoped(observer)] { observer->OnSuccess(); }));
}
void PeerConnectionMessageHandler::PostSetSessionDescriptionFailure(
SetSessionDescriptionObserver* observer,
RTCError&& error) {
RTC_DCHECK(!error.ok());
SetSessionDescriptionMsg* msg = new SetSessionDescriptionMsg(observer);
msg->error = std::move(error);
signaling_thread()->Post(RTC_FROM_HERE, this,
MSG_SET_SESSIONDESCRIPTION_FAILED, msg);
signaling_thread_->PostTask(SafeTask(
safety_.flag(),
[observer = WrapScoped(observer), error = std::move(error)]() mutable {
observer->OnFailure(std::move(error));
}));
}
void PeerConnectionMessageHandler::PostCreateSessionDescriptionFailure(
CreateSessionDescriptionObserver* observer,
RTCError error) {
RTC_DCHECK(!error.ok());
CreateSessionDescriptionMsg* msg = new CreateSessionDescriptionMsg(observer);
msg->error = std::move(error);
signaling_thread()->Post(RTC_FROM_HERE, this,
MSG_CREATE_SESSIONDESCRIPTION_FAILED, msg);
// Do not protect this task with the safety_.flag() to ensure
// observer is invoked even if the PeerConnection is destroyed early.
signaling_thread_->PostTask(
[observer = WrapScoped(observer), error = std::move(error)]() mutable {
observer->OnFailure(std::move(error));
});
}
void PeerConnectionMessageHandler::PostGetStats(
StatsObserver* observer,
LegacyStatsCollectorInterface* legacy_stats,
MediaStreamTrackInterface* track) {
signaling_thread()->Post(
RTC_FROM_HERE, this, MSG_GETSTATS,
new LegacyGetStatsMsg(observer, legacy_stats, track));
signaling_thread_->PostTask(
SafeTask(safety_.flag(), [observer = WrapScoped(observer), legacy_stats,
track = WrapScoped(track)] {
StatsReports reports;
legacy_stats->GetStats(track.get(), &reports);
observer->OnComplete(reports);
}));
}
void PeerConnectionMessageHandler::RequestUsagePatternReport(
std::function<void()> func,
int delay_ms) {
signaling_thread()->PostDelayed(RTC_FROM_HERE, delay_ms, this,
MSG_REPORT_USAGE_PATTERN,
new RequestUsagePatternMsg(func));
signaling_thread_->PostDelayedTask(SafeTask(safety_.flag(), std::move(func)),
TimeDelta::Millis(delay_ms));
}
} // namespace webrtc

22
pc/peer_connection_message_handler.h
View File

@ -18,27 +18,18 @@
#include "api/peer_connection_interface.h"
#include "api/rtc_error.h"
#include "api/stats_types.h"
#include "api/task_queue/pending_task_safety_flag.h"
#include "api/task_queue/task_queue_base.h"
#include "pc/legacy_stats_collector_interface.h"
#include "rtc_base/message_handler.h"
#include "rtc_base/thread.h"
#include "rtc_base/thread_message.h"
namespace webrtc {
class CreateSessionDescriptionObserver;
class SetSessionDescriptionObserver;
class StatsCollectorInterface;
class StatsObserver;
class MediaStreamTrackInterface;
class PeerConnectionMessageHandler : public rtc::MessageHandler {
class PeerConnectionMessageHandler {
public:
explicit PeerConnectionMessageHandler(rtc::Thread* signaling_thread)
: signaling_thread_(signaling_thread) {}
~PeerConnectionMessageHandler();
~PeerConnectionMessageHandler() = default;
// Implements MessageHandler.
void OnMessage(rtc::Message* msg) override;
void PostSetSessionDescriptionSuccess(
SetSessionDescriptionObserver* observer);
void PostSetSessionDescriptionFailure(SetSessionDescriptionObserver* observer,
@ -52,9 +43,8 @@ class PeerConnectionMessageHandler : public rtc::MessageHandler {
void RequestUsagePatternReport(std::function<void()>, int delay_ms);
private:
rtc::Thread* signaling_thread() const { return signaling_thread_; }
rtc::Thread* const signaling_thread_;
ScopedTaskSafety safety_;
TaskQueueBase* const signaling_thread_;
};
} // namespace webrtc

4
pc/peer_connection_signaling_unittest.cc
View File

@ -580,7 +580,7 @@ TEST_P(PeerConnectionSignalingTest, CreateOffersAndShutdown) {
// We expect to have received a notification now even if the PeerConnection
// was terminated. The offer creation may or may not have succeeded, but we
// must have received a notification.
EXPECT_TRUE(observer->called());
EXPECT_TRUE_WAIT(observer->called(), kWaitTimeout);
}
}
@ -594,7 +594,7 @@ TEST_P(PeerConnectionSignalingTest, CloseCreateOfferAndShutdown) {
caller->pc()->Close();
caller->pc()->CreateOffer(observer.get(), RTCOfferAnswerOptions());
caller.reset(nullptr);
EXPECT_TRUE(observer->called());
EXPECT_TRUE_WAIT(observer->called(), kWaitTimeout);
}
TEST_P(PeerConnectionSignalingTest,