From 384682c6beefa993eea0dd091b8ca4fb6f02f5d2 Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 2 Apr 2021 13:51:26 -0700 Subject: [PATCH] Internal change PiperOrigin-RevId: 366507191 --- cpp/core/internal/mediums/webrtc/BUILD | 1 + .../mediums/webrtc/connection_flow.cc | 63 +++---- .../internal/mediums/webrtc/connection_flow.h | 9 +- .../mediums/webrtc/connection_flow_test.cc | 163 ++++++++++++++++++ .../webrtc/data_channel_observer_impl.cc | 24 ++- .../webrtc/data_channel_observer_impl.h | 14 +- .../webrtc/peer_connection_observer_impl.cc | 48 +----- .../webrtc/peer_connection_observer_impl.h | 20 +-- 8 files changed, 249 insertions(+), 93 deletions(-) diff --git a/cpp/core/internal/mediums/webrtc/BUILD b/cpp/core/internal/mediums/webrtc/BUILD index a310c2aa..294390ec 100644 --- a/cpp/core/internal/mediums/webrtc/BUILD +++ b/cpp/core/internal/mediums/webrtc/BUILD @@ -54,6 +54,7 @@ cc_library( cc_test( name = "webrtc_test", + timeout = "short", srcs = [ "connection_flow_test.cc", "peer_id_test.cc", diff --git a/cpp/core/internal/mediums/webrtc/connection_flow.cc b/cpp/core/internal/mediums/webrtc/connection_flow.cc index d90dace1..81b2ecf6 100644 --- a/cpp/core/internal/mediums/webrtc/connection_flow.cc +++ b/cpp/core/internal/mediums/webrtc/connection_flow.cc @@ -122,7 +122,7 @@ SessionDescriptionWrapper ConnectionFlow::CreateOffer() { data_channel_init.reliable = true; rtc::scoped_refptr data_channel = peer_connection_->CreateDataChannel(kDataChannelName, &data_channel_init); - data_channel->RegisterObserver(CreateDataChannelObserver(data_channel)); + RegisterDataChannelObserver(data_channel); auto success_future = new Future(); webrtc::PeerConnectionInterface::RTCOfferAnswerOptions options; @@ -168,6 +168,7 @@ SessionDescriptionWrapper ConnectionFlow::CreateAnswer() { bool ConnectionFlow::SetLocalSessionDescription(SessionDescriptionWrapper sdp) { MutexLock lock(&mutex_); + if (state_ == State::kEnded) return false; if (!sdp.IsValid()) return false; auto success_future = new Future(); @@ -252,8 +253,7 @@ bool ConnectionFlow::OnRemoteIceCandidatesReceived( } bool ConnectionFlow::Close() { - MutexLock lock(&mutex_); - return CloseLocked(); + return Close(/* close_peer_connection= */ true); } bool ConnectionFlow::InitPeerConnection(WebRtcMedium& webrtc_medium) { @@ -290,16 +290,19 @@ bool ConnectionFlow::InitPeerConnection(WebRtcMedium& webrtc_medium) { } void ConnectionFlow::OnSignalingStable() { - MutexLock lock(&mutex_); + OffloadFromSignalingThread([this] { + MutexLock lock(&mutex_); - if (state_ != State::kWaitingToConnect && state_ != State::kConnected) return; + if (state_ != State::kWaitingToConnect && state_ != State::kConnected) + return; - for (auto&& ice_candidate : cached_remote_ice_candidates_) { - if (!peer_connection_->AddIceCandidate(ice_candidate.get())) { - NEARBY_LOG(WARNING, "Unable to add remote ice candidate."); + for (auto&& ice_candidate : cached_remote_ice_candidates_) { + if (!peer_connection_->AddIceCandidate(ice_candidate.get())) { + NEARBY_LOG(WARNING, "Unable to add remote ice candidate."); + } } - } - cached_remote_ice_candidates_.clear(); + cached_remote_ice_candidates_.clear(); + }); } void ConnectionFlow::ProcessOnPeerConnectionChange( @@ -307,7 +310,11 @@ void ConnectionFlow::ProcessOnPeerConnectionChange( if (new_state == PeerConnectionState::kClosed || new_state == PeerConnectionState::kFailed || new_state == PeerConnectionState::kDisconnected) { - Close(); + // kClosed means that PeerConnection is already closed or + // is closing right now - PeerConnection::Close() triggered + // PeerConnectionObserver::OnConnectionChange(kClosed). + // We must not call PeerConnection::Close() again on that code path + Close(new_state != PeerConnectionState::kClosed); } } @@ -323,29 +330,25 @@ void ConnectionFlow::ProcessDataChannelConnected( data_channel_listener_.data_channel_created_cb(std::move(data_channel)); } -webrtc::DataChannelObserver* ConnectionFlow::CreateDataChannelObserver( +void ConnectionFlow::RegisterDataChannelObserver( rtc::scoped_refptr data_channel) { if (!data_channel_observer_) { - auto state_change_callback = [this, - data_channel{std::move(data_channel)}]() { + auto state_change_callback = [this, data_channel]() { if (data_channel->state() == webrtc::DataChannelInterface::DataState::kOpen) { OffloadFromSignalingThread( [this, data_channel{std::move(data_channel)}]() { ProcessDataChannelConnected(std::move(data_channel)); }); - } else if (data_channel->state() == - webrtc::DataChannelInterface::DataState::kClosed) { - data_channel->UnregisterObserver(); - data_channel_listener_.data_channel_closed_cb(); } }; data_channel_observer_ = absl::make_unique( - &data_channel_listener_, std::move(state_change_callback)); + data_channel, &data_channel_listener_, + std::move(state_change_callback)); + NEARBY_LOG(INFO, "Registered data channel observer"); + } else { + NEARBY_LOG(WARNING, "Data channel observer already exists"); } - - return reinterpret_cast( - data_channel_observer_.get()); } bool ConnectionFlow::TransitionState(State current_state, State new_state) { @@ -360,17 +363,19 @@ bool ConnectionFlow::TransitionState(State current_state, State new_state) { return true; } -bool ConnectionFlow::CloseLocked() { - NEARBY_LOG(INFO, "Closing WebRTC connection."); - if (state_ == State::kEnded) { - return false; +bool ConnectionFlow::Close(bool close_peer_connection) { + { + MutexLock lock(&mutex_); + if (state_ == State::kEnded) { + return false; + } + state_ = State::kEnded; } - state_ = State::kEnded; + NEARBY_LOG(INFO, "Closing WebRTC connection."); single_threaded_signaling_offloader_.Shutdown(); - peer_connection_observer_.DisconnectConnectionFlow(); - if (peer_connection_) peer_connection_->Close(); + if (peer_connection_ && close_peer_connection) peer_connection_->Close(); data_channel_observer_.reset(); NEARBY_LOG(INFO, "Closed WebRTC connection."); diff --git a/cpp/core/internal/mediums/webrtc/connection_flow.h b/cpp/core/internal/mediums/webrtc/connection_flow.h index 438a9401..41c5fd49 100644 --- a/cpp/core/internal/mediums/webrtc/connection_flow.h +++ b/cpp/core/internal/mediums/webrtc/connection_flow.h @@ -118,7 +118,7 @@ class ConnectionFlow { // Invoked when the peer connection indicates that signaling is stable. void OnSignalingStable() ABSL_LOCKS_EXCLUDED(mutex_); - webrtc::DataChannelObserver* CreateDataChannelObserver( + void RegisterDataChannelObserver( rtc::scoped_refptr data_channel); // Invoked upon changes in the state of peer connection, e.g. react to @@ -127,6 +127,11 @@ class ConnectionFlow { webrtc::PeerConnectionInterface::PeerConnectionState new_state) ABSL_LOCKS_EXCLUDED(mutex_); + // For tests only + webrtc::PeerConnectionInterface* GetPeerConnection() { + return peer_connection_.get(); + } + private: ConnectionFlow(LocalIceCandidateListener local_ice_candidate_listener, DataChannelListener data_channel_listener); @@ -148,7 +153,7 @@ class ConnectionFlow { ABSL_LOCKS_EXCLUDED(mutex_); void CloseAndNotifyLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - bool CloseLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + bool Close(bool close_peer_connection) ABSL_LOCKS_EXCLUDED(mutex_); void OffloadFromSignalingThread(Runnable runnable); diff --git a/cpp/core/internal/mediums/webrtc/connection_flow_test.cc b/cpp/core/internal/mediums/webrtc/connection_flow_test.cc index 96c9cca4..45e71632 100644 --- a/cpp/core/internal/mediums/webrtc/connection_flow_test.cc +++ b/cpp/core/internal/mediums/webrtc/connection_flow_test.cc @@ -239,6 +239,169 @@ TEST_F(ConnectionFlowTest, PeerConnectionTimeout) { EXPECT_EQ(flow2, nullptr); } +TEST_F(ConnectionFlowTest, TerminateAnswerer) { + WebRtcMedium webrtc_medium_offerer, webrtc_medium_answerer; + + Future message_received_future; + + Future> + offerer_data_channel_future; + Future> + answerer_data_channel_future; + + std::unique_ptr offerer, answerer; + + // Send Ice Candidates immediately when you retrieve them + offerer = ConnectionFlow::Create( + {.local_ice_candidate_found_cb = + [&answerer](const webrtc::IceCandidateInterface* candidate) { + std::vector> vec; + vec.push_back(CopyCandidate(candidate)); + // The callback might be alive while the objects in test are + // destroyed. + if (answerer) + answerer->OnRemoteIceCandidatesReceived(std::move(vec)); + }}, + {.data_channel_created_cb = + [&offerer_data_channel_future]( + rtc::scoped_refptr data_channel) { + offerer_data_channel_future.Set(std::move(data_channel)); + }}, + webrtc_medium_offerer); + ASSERT_NE(offerer, nullptr); + answerer = ConnectionFlow::Create( + {.local_ice_candidate_found_cb = + [&offerer](const webrtc::IceCandidateInterface* candidate) { + std::vector> vec; + vec.push_back(CopyCandidate(candidate)); + // The callback might be alive while the objects in test are + // destroyed. + if (offerer) + offerer->OnRemoteIceCandidatesReceived(std::move(vec)); + }}, + {.data_channel_created_cb = + [&answerer_data_channel_future]( + rtc::scoped_refptr data_channel) { + answerer_data_channel_future.Set(std::move(data_channel)); + }, + .data_channel_message_received_cb = + [&message_received_future](ByteArray bytes) { + message_received_future.Set(std::move(bytes)); + }}, + webrtc_medium_answerer); + ASSERT_NE(answerer, nullptr); + + // Create and send offer + SessionDescriptionWrapper offer = offerer->CreateOffer(); + EXPECT_EQ(offer.GetType(), webrtc::SdpType::kOffer); + EXPECT_TRUE(answerer->OnOfferReceived(offer)); + EXPECT_TRUE(offerer->SetLocalSessionDescription(std::move(offer))); + + // Create and send answer + SessionDescriptionWrapper answer = answerer->CreateAnswer(); + EXPECT_EQ(answer.GetType(), webrtc::SdpType::kAnswer); + EXPECT_TRUE(offerer->OnAnswerReceived(answer)); + EXPECT_TRUE(answerer->SetLocalSessionDescription(std::move(answer))); + + // Retrieve Data Channels + ExceptionOr> + offerer_channel = offerer_data_channel_future.Get(absl::Seconds(1)); + EXPECT_TRUE(offerer_channel.ok()); + ExceptionOr> + answerer_channel = answerer_data_channel_future.Get(absl::Seconds(1)); + EXPECT_TRUE(answerer_channel.ok()); + + answerer->GetPeerConnection()->Close(); + + // Send message on data channel + const char message[] = "Test"; + offerer_channel.result()->Send(webrtc::DataBuffer(message)); + ExceptionOr received_message = + message_received_future.Get(absl::Seconds(1)); + EXPECT_FALSE(received_message.ok()); +} + +TEST_F(ConnectionFlowTest, TerminateOfferer) { + WebRtcMedium webrtc_medium_offerer, webrtc_medium_answerer; + + Future message_received_future; + + Future> + offerer_data_channel_future; + Future> + answerer_data_channel_future; + + std::unique_ptr offerer, answerer; + + // Send Ice Candidates immediately when you retrieve them + offerer = ConnectionFlow::Create( + {.local_ice_candidate_found_cb = + [&answerer](const webrtc::IceCandidateInterface* candidate) { + std::vector> vec; + vec.push_back(CopyCandidate(candidate)); + // The callback might be alive while the objects in test are + // destroyed. + if (answerer) + answerer->OnRemoteIceCandidatesReceived(std::move(vec)); + }}, + {.data_channel_created_cb = + [&offerer_data_channel_future]( + rtc::scoped_refptr data_channel) { + offerer_data_channel_future.Set(std::move(data_channel)); + }}, + webrtc_medium_offerer); + ASSERT_NE(offerer, nullptr); + answerer = ConnectionFlow::Create( + {.local_ice_candidate_found_cb = + [&offerer](const webrtc::IceCandidateInterface* candidate) { + std::vector> vec; + vec.push_back(CopyCandidate(candidate)); + // The callback might be alive while the objects in test are + // destroyed. + if (offerer) + offerer->OnRemoteIceCandidatesReceived(std::move(vec)); + }}, + {.data_channel_created_cb = + [&answerer_data_channel_future]( + rtc::scoped_refptr data_channel) { + answerer_data_channel_future.Set(std::move(data_channel)); + }, + .data_channel_message_received_cb = + [&message_received_future](ByteArray bytes) { + message_received_future.Set(std::move(bytes)); + }}, + webrtc_medium_answerer); + ASSERT_NE(answerer, nullptr); + + // Create and send offer + SessionDescriptionWrapper offer = offerer->CreateOffer(); + EXPECT_EQ(offer.GetType(), webrtc::SdpType::kOffer); + EXPECT_TRUE(answerer->OnOfferReceived(offer)); + EXPECT_TRUE(offerer->SetLocalSessionDescription(std::move(offer))); + + // Create and send answer + SessionDescriptionWrapper answer = answerer->CreateAnswer(); + EXPECT_EQ(answer.GetType(), webrtc::SdpType::kAnswer); + EXPECT_TRUE(offerer->OnAnswerReceived(answer)); + EXPECT_TRUE(answerer->SetLocalSessionDescription(std::move(answer))); + + // Retrieve Data Channels + ExceptionOr> + offerer_channel = offerer_data_channel_future.Get(absl::Seconds(1)); + EXPECT_TRUE(offerer_channel.ok()); + ExceptionOr> + answerer_channel = answerer_data_channel_future.Get(absl::Seconds(1)); + EXPECT_TRUE(answerer_channel.ok()); + + offerer->GetPeerConnection()->Close(); + + // Send message on data channel + const char message[] = "Test"; + offerer_channel.result()->Send(webrtc::DataBuffer(message)); + ExceptionOr received_message = + message_received_future.Get(absl::Seconds(1)); + EXPECT_FALSE(received_message.ok()); +} } // namespace } // namespace mediums } // namespace connections diff --git a/cpp/core/internal/mediums/webrtc/data_channel_observer_impl.cc b/cpp/core/internal/mediums/webrtc/data_channel_observer_impl.cc index cba998d6..a382f6ba 100644 --- a/cpp/core/internal/mediums/webrtc/data_channel_observer_impl.cc +++ b/cpp/core/internal/mediums/webrtc/data_channel_observer_impl.cc @@ -20,12 +20,24 @@ namespace connections { namespace mediums { DataChannelObserverImpl::DataChannelObserverImpl( + rtc::scoped_refptr data_channel, DataChannelListener* data_channel_listener, DataChannelStateChangeCallback callback) : data_channel_listener_(data_channel_listener), - state_change_callback_(std::move(callback)) {} + state_change_callback_(std::move(callback)), + data_channel_{std::move(data_channel)} { + data_channel_->RegisterObserver(this); +} -void DataChannelObserverImpl::OnStateChange() { state_change_callback_(); } +DataChannelObserverImpl::~DataChannelObserverImpl() { Disconnect(); } + +void DataChannelObserverImpl::OnStateChange() { + if (data_channel_->state() == + webrtc::DataChannelInterface::DataState::kClosed) { + Disconnect(); + } + state_change_callback_(); +} void DataChannelObserverImpl::OnMessage(const webrtc::DataBuffer& buffer) { data_channel_listener_->data_channel_message_received_cb( @@ -36,6 +48,14 @@ void DataChannelObserverImpl::OnBufferedAmountChange(uint64_t sent_data_size) { data_channel_listener_->data_channel_buffered_amount_changed_cb(); } +void DataChannelObserverImpl::Disconnect() { + data_channel_->UnregisterObserver(); + if (data_channel_listener_) { + data_channel_listener_->data_channel_closed_cb(); + data_channel_listener_ = nullptr; + } +} + } // namespace mediums } // namespace connections } // namespace nearby diff --git a/cpp/core/internal/mediums/webrtc/data_channel_observer_impl.h b/cpp/core/internal/mediums/webrtc/data_channel_observer_impl.h index 4de3f346..c876f90f 100644 --- a/cpp/core/internal/mediums/webrtc/data_channel_observer_impl.h +++ b/cpp/core/internal/mediums/webrtc/data_channel_observer_impl.h @@ -27,9 +27,14 @@ class DataChannelObserverImpl : public webrtc::DataChannelObserver { public: using DataChannelStateChangeCallback = std::function; - ~DataChannelObserverImpl() override = default; - DataChannelObserverImpl(DataChannelListener* data_channel_listener, - DataChannelStateChangeCallback callback); + // Creates and registers an observer for |data_channel| + // The observer is unregistered in destructor or when |data_channel| + // is closed. + DataChannelObserverImpl( + rtc::scoped_refptr data_channel, + DataChannelListener* data_channel_listener, + DataChannelStateChangeCallback callback); + ~DataChannelObserverImpl() override; // webrtc::DataChannelObserver: void OnStateChange() override; @@ -37,8 +42,11 @@ class DataChannelObserverImpl : public webrtc::DataChannelObserver { void OnBufferedAmountChange(uint64_t sent_data_size) override; private: + void Disconnect(); + DataChannelListener* data_channel_listener_; DataChannelStateChangeCallback state_change_callback_; + rtc::scoped_refptr data_channel_; }; } // namespace mediums diff --git a/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.cc b/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.cc index 950b2dcb..8cc816cc 100644 --- a/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.cc +++ b/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.cc @@ -28,11 +28,6 @@ PeerConnectionObserverImpl::PeerConnectionObserverImpl( : connection_flow_(connection_flow), local_ice_candidate_listener_(std::move(local_ice_candidate_listener)) {} -PeerConnectionObserverImpl::~PeerConnectionObserverImpl() { - MutexLock lock(&mutex_); - connection_flow_ = nullptr; -} - void PeerConnectionObserverImpl::OnIceCandidate( const webrtc::IceCandidateInterface* candidate) { local_ice_candidate_listener_.local_ice_candidate_found_cb(candidate); @@ -41,31 +36,15 @@ void PeerConnectionObserverImpl::OnIceCandidate( void PeerConnectionObserverImpl::OnSignalingChange( webrtc::PeerConnectionInterface::SignalingState new_state) { NEARBY_LOG(INFO, "OnSignalingChange: %d", new_state); - - OffloadFromSignalingThread([this, new_state]() { - MutexLock lock(&mutex_); - if (new_state == webrtc::PeerConnectionInterface::SignalingState::kStable && - connection_flow_) { - connection_flow_->OnSignalingStable(); - } - }); + if (new_state == webrtc::PeerConnectionInterface::SignalingState::kStable) { + connection_flow_->OnSignalingStable(); + } } void PeerConnectionObserverImpl::OnDataChannel( rtc::scoped_refptr data_channel) { NEARBY_LOG(INFO, "OnDataChannel"); - - webrtc::DataChannelObserver* data_channel_observer = nullptr; - { - MutexLock lock(&mutex_); - if (!connection_flow_) { - return; - } - - data_channel_observer = - connection_flow_->CreateDataChannelObserver(data_channel); - } - data_channel->RegisterObserver(data_channel_observer); + connection_flow_->RegisterDataChannelObserver(std::move(data_channel)); } void PeerConnectionObserverImpl::OnIceGatheringChange( @@ -76,28 +55,13 @@ void PeerConnectionObserverImpl::OnIceGatheringChange( void PeerConnectionObserverImpl::OnConnectionChange( webrtc::PeerConnectionInterface::PeerConnectionState new_state) { NEARBY_LOG(INFO, "OnConnectionChange: %d", new_state); - - OffloadFromSignalingThread([this, new_state]() { - MutexLock lock(&mutex_); - if (connection_flow_) { - connection_flow_->ProcessOnPeerConnectionChange(new_state); - } - }); + connection_flow_->ProcessOnPeerConnectionChange(new_state); } -void PeerConnectionObserverImpl ::OnRenegotiationNeeded() { +void PeerConnectionObserverImpl::OnRenegotiationNeeded() { NEARBY_LOG(INFO, "OnRenegotiationNeeded"); } -void PeerConnectionObserverImpl::DisconnectConnectionFlow() { - MutexLock lock(&mutex_); - connection_flow_ = nullptr; -} - -void PeerConnectionObserverImpl::OffloadFromSignalingThread(Runnable runnable) { - single_threaded_signaling_offloader_.Execute(std::move(runnable)); -} - } // namespace mediums } // namespace connections } // namespace nearby diff --git a/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.h b/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.h index 7bcd86b1..045bbfe9 100644 --- a/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.h +++ b/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.h @@ -31,33 +31,23 @@ class PeerConnectionObserverImpl : public webrtc::PeerConnectionObserver { PeerConnectionObserverImpl( ConnectionFlow* connection_flow, LocalIceCandidateListener local_ice_candidate_listener); - ~PeerConnectionObserverImpl() override; + ~PeerConnectionObserverImpl() override = default; // webrtc::PeerConnectionObserver: void OnIceCandidate(const webrtc::IceCandidateInterface* candidate) override; void OnSignalingChange( - webrtc::PeerConnectionInterface::SignalingState new_state) override - ABSL_LOCKS_EXCLUDED(mutex_); + webrtc::PeerConnectionInterface::SignalingState new_state) override; void OnDataChannel( - rtc::scoped_refptr data_channel) override - ABSL_LOCKS_EXCLUDED(mutex_); + rtc::scoped_refptr data_channel) override; void OnIceGatheringChange( webrtc::PeerConnectionInterface::IceGatheringState new_state) override; void OnConnectionChange( - webrtc::PeerConnectionInterface::PeerConnectionState new_state) override - ABSL_LOCKS_EXCLUDED(mutex_); + webrtc::PeerConnectionInterface::PeerConnectionState new_state) override; void OnRenegotiationNeeded() override; - void DisconnectConnectionFlow() ABSL_LOCKS_EXCLUDED(mutex_); - private: - void OffloadFromSignalingThread(Runnable runnable); - - // NOTE: This must be a recursive mutex due to the call interactions. - RecursiveMutex mutex_; // protects access to connection_flow_ - ConnectionFlow* connection_flow_ ABSL_GUARDED_BY(mutex_); + ConnectionFlow* connection_flow_; LocalIceCandidateListener local_ice_candidate_listener_; - SingleThreadExecutor single_threaded_signaling_offloader_; }; } // namespace mediums