From 8fafd3ef09818e1983945aa8be9663713c79c142 Mon Sep 17 00:00:00 2001
From: hai007 <hais@google.com>
Date: Mon, 8 Mar 2021 16:38:14 -0800
Subject: [PATCH] Internal change

PiperOrigin-RevId: 361686543
---
 .../mediums/webrtc/connection_flow.cc         |  2 +-
 .../webrtc/peer_connection_observer_impl.cc   | 22 ++++++++++++++-----
 .../webrtc/peer_connection_observer_impl.h    | 15 ++++++++-----
 3 files changed, 27 insertions(+), 12 deletions(-)

diff --git a/cpp/core/internal/mediums/webrtc/connection_flow.cc b/cpp/core/internal/mediums/webrtc/connection_flow.cc
index 15b25609..ff8582c9 100644
--- a/cpp/core/internal/mediums/webrtc/connection_flow.cc
+++ b/cpp/core/internal/mediums/webrtc/connection_flow.cc
@@ -350,7 +350,7 @@ bool ConnectionFlow::CloseLocked() {
   state_ = State::kEnded;
 
   single_threaded_signaling_offloader_.Shutdown();
-  peer_connection_observer_.Shutdown();
+  peer_connection_observer_.DisconnectConnectionFlow();
 
   if (peer_connection_) peer_connection_->Close();
 
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 75be6c7c..950b2dcb 100644
--- a/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.cc
+++ b/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.cc
@@ -29,7 +29,8 @@ PeerConnectionObserverImpl::PeerConnectionObserverImpl(
       local_ice_candidate_listener_(std::move(local_ice_candidate_listener)) {}
 
 PeerConnectionObserverImpl::~PeerConnectionObserverImpl() {
-    Shutdown();
+  MutexLock lock(&mutex_);
+  connection_flow_ = nullptr;
 }
 
 void PeerConnectionObserverImpl::OnIceCandidate(
@@ -42,6 +43,7 @@ void PeerConnectionObserverImpl::OnSignalingChange(
   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();
@@ -53,10 +55,17 @@ void PeerConnectionObserverImpl::OnDataChannel(
     rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel) {
   NEARBY_LOG(INFO, "OnDataChannel");
 
-  if (connection_flow_) {
-    data_channel->RegisterObserver(
-        connection_flow_->CreateDataChannelObserver(data_channel));
+  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);
 }
 
 void PeerConnectionObserverImpl::OnIceGatheringChange(
@@ -69,6 +78,7 @@ void PeerConnectionObserverImpl::OnConnectionChange(
   NEARBY_LOG(INFO, "OnConnectionChange: %d", new_state);
 
   OffloadFromSignalingThread([this, new_state]() {
+    MutexLock lock(&mutex_);
     if (connection_flow_) {
       connection_flow_->ProcessOnPeerConnectionChange(new_state);
     }
@@ -79,8 +89,8 @@ void PeerConnectionObserverImpl ::OnRenegotiationNeeded() {
   NEARBY_LOG(INFO, "OnRenegotiationNeeded");
 }
 
-void PeerConnectionObserverImpl::Shutdown() {
-  single_threaded_signaling_offloader_.Shutdown();
+void PeerConnectionObserverImpl::DisconnectConnectionFlow() {
+  MutexLock lock(&mutex_);
   connection_flow_ = nullptr;
 }
 
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 a2906c20..7bcd86b1 100644
--- a/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.h
+++ b/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.h
@@ -36,21 +36,26 @@ class PeerConnectionObserverImpl : public webrtc::PeerConnectionObserver {
   // webrtc::PeerConnectionObserver:
   void OnIceCandidate(const webrtc::IceCandidateInterface* candidate) override;
   void OnSignalingChange(
-      webrtc::PeerConnectionInterface::SignalingState new_state) override;
+      webrtc::PeerConnectionInterface::SignalingState new_state) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
   void OnDataChannel(
-      rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel) override;
+      rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
   void OnIceGatheringChange(
       webrtc::PeerConnectionInterface::IceGatheringState new_state) override;
   void OnConnectionChange(
-      webrtc::PeerConnectionInterface::PeerConnectionState new_state) override;
+      webrtc::PeerConnectionInterface::PeerConnectionState new_state) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
   void OnRenegotiationNeeded() override;
 
-  void Shutdown();
+  void DisconnectConnectionFlow() ABSL_LOCKS_EXCLUDED(mutex_);
 
  private:
   void OffloadFromSignalingThread(Runnable runnable);
 
-  ConnectionFlow* volatile connection_flow_;
+  // 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_);
   LocalIceCandidateListener local_ice_candidate_listener_;
   SingleThreadExecutor single_threaded_signaling_offloader_;
 };