Internal change

PiperOrigin-RevId: 366507191

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",

cpp/core/internal/mediums/webrtc/connection_flow.cc

@@ -122,7 +122,7 @@ SessionDescriptionWrapper ConnectionFlow::CreateOffer() {
   data_channel_init.reliable = true;
   rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel =
       peer_connection_->CreateDataChannel(kDataChannelName, &data_channel_init);
-  data_channel->RegisterObserver(CreateDataChannelObserver(data_channel));
+  RegisterDataChannelObserver(data_channel);
 
   auto success_future = new Future<SessionDescriptionWrapper>();
   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<bool>();
@@ -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<webrtc::DataChannelInterface> 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<DataChannelObserverImpl>(
-        &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<webrtc::DataChannelObserver*>(
-      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.");

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<webrtc::DataChannelInterface> 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);
 

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<ByteArray> message_received_future;
+
+  Future<rtc::scoped_refptr<webrtc::DataChannelInterface>>
+      offerer_data_channel_future;
+  Future<rtc::scoped_refptr<webrtc::DataChannelInterface>>
+      answerer_data_channel_future;
+
+  std::unique_ptr<ConnectionFlow> 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<std::unique_ptr<webrtc::IceCandidateInterface>> 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<webrtc::DataChannelInterface> 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<std::unique_ptr<webrtc::IceCandidateInterface>> 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<webrtc::DataChannelInterface> 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<rtc::scoped_refptr<webrtc::DataChannelInterface>>
+      offerer_channel = offerer_data_channel_future.Get(absl::Seconds(1));
+  EXPECT_TRUE(offerer_channel.ok());
+  ExceptionOr<rtc::scoped_refptr<webrtc::DataChannelInterface>>
+      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<ByteArray> 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<ByteArray> message_received_future;
+
+  Future<rtc::scoped_refptr<webrtc::DataChannelInterface>>
+      offerer_data_channel_future;
+  Future<rtc::scoped_refptr<webrtc::DataChannelInterface>>
+      answerer_data_channel_future;
+
+  std::unique_ptr<ConnectionFlow> 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<std::unique_ptr<webrtc::IceCandidateInterface>> 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<webrtc::DataChannelInterface> 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<std::unique_ptr<webrtc::IceCandidateInterface>> 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<webrtc::DataChannelInterface> 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<rtc::scoped_refptr<webrtc::DataChannelInterface>>
+      offerer_channel = offerer_data_channel_future.Get(absl::Seconds(1));
+  EXPECT_TRUE(offerer_channel.ok());
+  ExceptionOr<rtc::scoped_refptr<webrtc::DataChannelInterface>>
+      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<ByteArray> received_message =
+      message_received_future.Get(absl::Seconds(1));
+  EXPECT_FALSE(received_message.ok());
+}
 }  // namespace
 }  // namespace mediums
 }  // namespace connections

cpp/core/internal/mediums/webrtc/data_channel_observer_impl.cc

@@ -20,12 +20,24 @@ namespace connections {
 namespace mediums {
 
 DataChannelObserverImpl::DataChannelObserverImpl(
+    rtc::scoped_refptr<webrtc::DataChannelInterface> 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

cpp/core/internal/mediums/webrtc/data_channel_observer_impl.h

@@ -27,9 +27,14 @@ class DataChannelObserverImpl : public webrtc::DataChannelObserver {
  public:
   using DataChannelStateChangeCallback = std::function<void()>;
 
-  ~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<webrtc::DataChannelInterface> 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<webrtc::DataChannelInterface> data_channel_;
 };
 
 }  // namespace mediums

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<webrtc::DataChannelInterface> 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

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<webrtc::DataChannelInterface> data_channel) override
-      ABSL_LOCKS_EXCLUDED(mutex_);
+      rtc::scoped_refptr<webrtc::DataChannelInterface> 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