Add WEB_RTC_NON_CELLULAR enum, IV

PiperOrigin-RevId: 686408097

connections/implementation/mediums/webrtc.cc

@@ -36,6 +36,7 @@
 #include "internal/platform/cancellation_flag_listener.h"
 #include "internal/platform/exception.h"
 #include "internal/platform/future.h"
+#include "internal/platform/feature_flags.h"
 #include "internal/platform/logging.h"
 #include "internal/platform/mutex_lock.h"
 #include "internal/platform/webrtc.h"
@@ -742,9 +743,32 @@ std::unique_ptr<ConnectionFlow> WebRtc::CreateConnectionFlow(
             });
           }},
       },
+      {
+          .adapter_type_changed_cb =
+              {[this](/*rtc::AdapterType*/ int adapter_type) {
+                OffloadFromThread(
+                    "rtc-adapter-type-changed", [this, adapter_type]() {
+                      if (FeatureFlags::GetInstance()
+                              .GetFlags()
+                              .support_web_rtc_non_cellular_medium) {
+                        AdapterTypeChangedHandler(adapter_type);
+                      }
+                    });
+              }},
+      },
       *medium_);
 }
 
+void WebRtc::AdapterTypeChangedHandler(/*rtc::AdapterType*/ int adapter_type) {
+  // TODO(edwinwu): Uncomment this once OSS supports WEB_RTC
+  // MutexLock lock(&mutex_);
+  // is_using_cellular_ = adapter_type == rtc::ADAPTER_TYPE_CELLULAR ||
+  //                      adapter_type == rtc::ADAPTER_TYPE_CELLULAR_2G ||
+  //                      adapter_type == rtc::ADAPTER_TYPE_CELLULAR_3G ||
+  //                      adapter_type == rtc::ADAPTER_TYPE_CELLULAR_4G ||
+  //                      adapter_type == rtc::ADAPTER_TYPE_CELLULAR_5G;
+}
+
 void WebRtc::RemoveConnectionFlow(const WebrtcPeerId& remote_peer_id) {
   if (!connection_flows_.erase(remote_peer_id.GetId())) {
     return;
@@ -764,6 +788,11 @@ void WebRtc::OffloadFromThread(const std::string& name, Runnable runnable) {
   single_thread_executor_.Execute(name, std::move(runnable));
 }
 
+bool WebRtc::IsUsingCellular() {
+  MutexLock lock(&mutex_);
+  return is_using_cellular_;
+}
+
 }  // namespace mediums
 }  // namespace connections
 }  // namespace nearby

connections/implementation/mediums/webrtc.h

@@ -90,6 +90,8 @@ class WebRtc {
       CancellationFlag* cancellation_flag, bool non_cellular)
       ABSL_LOCKS_EXCLUDED(mutex_);
 
+  bool IsUsingCellular() ABSL_LOCKS_EXCLUDED(mutex_);
+
  protected:
   // Use for unit tests only to inject a WebRtcMedium.
   explicit WebRtc(std::unique_ptr<WebRtcMedium> medium);
@@ -232,6 +234,10 @@ class WebRtc {
   void RestartTachyonReceiveMessages(const std::string& service_id)
       ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
+  // Runs on |single_thread_executor_|.
+  void AdapterTypeChangedHandler(/*rtc::AdapterType*/ int adapter_type)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
   void OffloadFromThread(const std::string& name, Runnable runnable);
 
   Mutex mutex_;
@@ -256,6 +262,8 @@ class WebRtc {
   // a unique ConnectionFlow.
   absl::flat_hash_map<std::string, std::unique_ptr<ConnectionFlow>>
       connection_flows_ ABSL_GUARDED_BY(mutex_);
+
+  bool is_using_cellular_ ABSL_GUARDED_BY(mutex_) = true;
 };
 
 }  // namespace mediums

connections/implementation/mediums/webrtc/BUILD

@@ -42,6 +42,7 @@ cc_library(
         "//internal/platform:types",
         "//proto/mediums:web_rtc_signaling_frames_cc_proto",
         # TODO: Support WebRTC
+        "@com_google_absl//absl/base:core_headers",
         "@com_google_absl//absl/functional:any_invocable",
         "@com_google_absl//absl/memory",
         "@com_google_absl//absl/time",
@@ -94,7 +95,6 @@ cc_test(
         "//internal/platform/implementation/g3",  # buildcleaner: keep
         "//third_party/protobuf",
         "//third_party/webrtc/files/stable/webrtc/api:libjingle_peerconnection_api",
-        "//third_party/webrtc/files/stable/webrtc/api:rtc_error",
         "//third_party/webrtc/files/stable/webrtc/api:scoped_refptr",
         "@com_github_protobuf_matchers//protobuf-matchers",
         "@com_google_absl//absl/time",

connections/implementation/mediums/webrtc/connection_flow.cc

@@ -18,14 +18,21 @@
 
 #include <iterator>
 #include <memory>
+#include <utility>
+#include <vector>
 
 #include "absl/memory/memory.h"
 #include "absl/time/time.h"
+#include "connections/implementation/mediums/webrtc/data_channel_listener.h"
+#include "connections/implementation/mediums/webrtc/local_ice_candidate_listener.h"
 #include "connections/implementation/mediums/webrtc/session_description_wrapper.h"
 #include "connections/implementation/mediums/webrtc/webrtc_socket_impl.h"
 #include "connections/implementation/mediums/webrtc_socket.h"
+#include "internal/platform/exception.h"
+#include "internal/platform/future.h"
 #include "internal/platform/logging.h"
 #include "internal/platform/mutex_lock.h"
+#include "internal/platform/runnable.h"
 #include "internal/platform/webrtc.h"
 #include "webrtc/api/data_channel_interface.h"
 #include "webrtc/api/jsep.h"
@@ -118,10 +125,11 @@ using PeerConnectionState =
 
 std::unique_ptr<ConnectionFlow> ConnectionFlow::Create(
     LocalIceCandidateListener local_ice_candidate_listener,
-    DataChannelListener data_channel_listener, WebRtcMedium& webrtc_medium) {
-  auto connection_flow = absl::WrapUnique(
-      new ConnectionFlow(std::move(local_ice_candidate_listener),
-                         std::move(data_channel_listener)));
+    DataChannelListener data_channel_listener,
+    AdapterTypeListener adapter_type_listener, WebRtcMedium& webrtc_medium) {
+  auto connection_flow = absl::WrapUnique(new ConnectionFlow(
+      std::move(local_ice_candidate_listener), std::move(data_channel_listener),
+      std::move(adapter_type_listener)));
   if (connection_flow->InitPeerConnection(webrtc_medium)) {
     return connection_flow;
   }
@@ -131,9 +139,11 @@ std::unique_ptr<ConnectionFlow> ConnectionFlow::Create(
 
 ConnectionFlow::ConnectionFlow(
     LocalIceCandidateListener local_ice_candidate_listener,
-    DataChannelListener data_channel_listener)
+    DataChannelListener data_channel_listener,
+    AdapterTypeListener adapter_type_listener)
     : data_channel_listener_(std::move(data_channel_listener)),
-      local_ice_candidate_listener_(std::move(local_ice_candidate_listener)) {}
+      local_ice_candidate_listener_(std::move(local_ice_candidate_listener)),
+      adapter_type_listener_(std::move(adapter_type_listener)) {}
 
 ConnectionFlow::~ConnectionFlow() {
   NEARBY_LOGS(INFO) << "~ConnectionFlow";
@@ -563,6 +573,7 @@ ConnectionFlow::GetAndResetPeerConnection() {
   MutexLock lock(&mutex_);
   return std::move(peer_connection_);
 }
+
 }  // namespace mediums
 }  // namespace connections
 }  // namespace nearby

connections/implementation/mediums/webrtc/connection_flow.h

@@ -18,14 +18,20 @@
 #ifndef NO_WEBRTC
 
 #include <memory>
+#include <vector>
 
+#include "absl/base/thread_annotations.h"
+#include "absl/functional/any_invocable.h"
+#include "absl/time/time.h"
 #include "connections/implementation/mediums/webrtc/data_channel_listener.h"
 #include "connections/implementation/mediums/webrtc/local_ice_candidate_listener.h"
 #include "connections/implementation/mediums/webrtc/session_description_wrapper.h"
 #include "connections/implementation/mediums/webrtc_socket.h"
-#include "internal/platform/runnable.h"
 #include "internal/platform/count_down_latch.h"
-#include "internal/platform/single_thread_executor.h"
+#include "internal/platform/future.h"
+#include "internal/platform/listeners.h"
+#include "internal/platform/mutex.h"
+#include "internal/platform/runnable.h"
 #include "internal/platform/webrtc.h"
 #include "webrtc/api/data_channel_interface.h"
 #include "webrtc/api/peer_connection_interface.h"
@@ -80,11 +86,20 @@ class ConnectionFlow : public webrtc::PeerConnectionObserver {
     kEnded,
   };
 
+  // The listener that notifies the AdapterType has been changed.
+  // TODO(edwinwu): replace param |int| to |rtc::AdapterType| once OSS supports
+  // WebRtc.
+  struct AdapterTypeListener {
+    absl::AnyInvocable<void(/*rtc::AdapterType*/ int adapter_type)>
+        adapter_type_changed_cb = DefaultCallback</*rtc::AdapterType*/ int>();
+  };
+
   // This method blocks on the creation of the peer connection object.
   // Can be called on any thread but never called on signaling thread.
   static std::unique_ptr<ConnectionFlow> Create(
       LocalIceCandidateListener local_ice_candidate_listener,
-      DataChannelListener data_channel_listener, WebRtcMedium& webrtc_medium);
+      DataChannelListener data_channel_listener,
+      AdapterTypeListener adapter_type_listener, WebRtcMedium& webrtc_medium);
   ~ConnectionFlow() override;
 
   // Create the offer that will be sent to the remote. Mirrors the behaviour of
@@ -134,13 +149,17 @@ class ConnectionFlow : public webrtc::PeerConnectionObserver {
   void OnConnectionChange(
       webrtc::PeerConnectionInterface::PeerConnectionState new_state) override;
   void OnRenegotiationNeeded() override;
+  // TODO(edwinwu): Implement once OSS supports WebRtc.
+  // void OnIceSelectedCandidatePairChanged(
+  //     const cricket::CandidatePairChangeEvent& event) override;
 
   // Public because it's used in tests too.
   rtc::scoped_refptr<webrtc::PeerConnectionInterface> GetPeerConnection();
 
  private:
   ConnectionFlow(LocalIceCandidateListener local_ice_candidate_listener,
-                 DataChannelListener data_channel_listener);
+                 DataChannelListener data_channel_listener,
+                 AdapterTypeListener adapter_type_listener);
 
   // Resets peer connection reference. Returns old value.
   rtc::scoped_refptr<webrtc::PeerConnectionInterface>
@@ -223,6 +242,8 @@ class ConnectionFlow : public webrtc::PeerConnectionObserver {
   // the former is thread-safe.
   std::shared_ptr<void> can_run_tasks_ = std::make_shared<int>();
 
+  AdapterTypeListener adapter_type_listener_;
+
   friend class CreateSessionDescriptionObserverImpl;
 };
 

connections/implementation/mediums/webrtc/connection_flow_test.cc

@@ -15,21 +15,23 @@
 #include "connections/implementation/mediums/webrtc/connection_flow.h"
 
 #include <memory>
+#include <string>
+#include <utility>
 #include <vector>
 
-#include "gmock/gmock.h"
-#include "protobuf-matchers/protocol-buffer-matchers.h"
 #include "gtest/gtest.h"
 #include "absl/time/time.h"
+#include "connections/implementation/mediums/webrtc/data_channel_listener.h"
+#include "connections/implementation/mediums/webrtc/local_ice_candidate_listener.h"
 #include "connections/implementation/mediums/webrtc/session_description_wrapper.h"
 #include "connections/implementation/mediums/webrtc_socket.h"
 #include "internal/platform/byte_array.h"
 #include "internal/platform/count_down_latch.h"
+#include "internal/platform/exception.h"
+#include "internal/platform/future.h"
 #include "internal/platform/medium_environment.h"
 #include "internal/platform/webrtc.h"
-#include "webrtc/api/data_channel_interface.h"
 #include "webrtc/api/jsep.h"
-#include "webrtc/api/rtc_error.h"
 #include "webrtc/api/scoped_refptr.h"
 
 namespace nearby {
@@ -78,6 +80,10 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) {
            [&offerer_socket_future](WebRtcSocketWrapper socket) {
              offerer_socket_future.Set(std::move(socket));
            }},
+      {.adapter_type_changed_cb =
+           [](/*rtc::AdapterType*/ int adapter_type) {
+             // Do nothing
+           }},
       webrtc_medium_offerer);
   ASSERT_NE(offerer, nullptr);
   answerer = ConnectionFlow::Create(
@@ -94,6 +100,10 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) {
            [&answerer_socket_future](WebRtcSocketWrapper socket) {
              answerer_socket_future.Set(std::move(socket));
            }},
+      {.adapter_type_changed_cb =
+           [](/*rtc::AdapterType*/ int adapter_type) {
+             // Do nothing
+           }},
       webrtc_medium_answerer);
   ASSERT_NE(answerer, nullptr);
 
@@ -133,7 +143,8 @@ TEST_F(ConnectionFlowTest, CreateAnswerBeforeOfferReceived) {
   WebRtcMedium webrtc_medium;
 
   std::unique_ptr<ConnectionFlow> answerer = ConnectionFlow::Create(
-      LocalIceCandidateListener(), DataChannelListener(), webrtc_medium);
+      LocalIceCandidateListener(), DataChannelListener(),
+      ConnectionFlow::AdapterTypeListener(), webrtc_medium);
   ASSERT_NE(answerer, nullptr);
 
   SessionDescriptionWrapper answer = answerer->CreateAnswer();
@@ -143,13 +154,13 @@ TEST_F(ConnectionFlowTest, CreateAnswerBeforeOfferReceived) {
 TEST_F(ConnectionFlowTest, SetAnswerBeforeOffer) {
   WebRtcMedium webrtc_medium_offerer, webrtc_medium_answerer;
 
-  std::unique_ptr<ConnectionFlow> offerer =
-      ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(),
-                             webrtc_medium_offerer);
+  std::unique_ptr<ConnectionFlow> offerer = ConnectionFlow::Create(
+      LocalIceCandidateListener(), DataChannelListener(),
+      ConnectionFlow::AdapterTypeListener(), webrtc_medium_offerer);
   ASSERT_NE(offerer, nullptr);
-  std::unique_ptr<ConnectionFlow> answerer =
-      ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(),
-                             webrtc_medium_answerer);
+  std::unique_ptr<ConnectionFlow> answerer = ConnectionFlow::Create(
+      LocalIceCandidateListener(), DataChannelListener(),
+      ConnectionFlow::AdapterTypeListener(), webrtc_medium_answerer);
   ASSERT_NE(answerer, nullptr);
 
   SessionDescriptionWrapper offer = offerer->CreateOffer();
@@ -168,7 +179,8 @@ TEST_F(ConnectionFlowTest, CannotCreateOfferAfterClose) {
   WebRtcMedium webrtc_medium;
 
   std::unique_ptr<ConnectionFlow> offerer = ConnectionFlow::Create(
-      LocalIceCandidateListener(), DataChannelListener(), webrtc_medium);
+      LocalIceCandidateListener(), DataChannelListener(),
+      ConnectionFlow::AdapterTypeListener(), webrtc_medium);
   ASSERT_NE(offerer, nullptr);
 
   EXPECT_TRUE(offerer->CloseIfNotConnected());
@@ -180,7 +192,8 @@ TEST_F(ConnectionFlowTest, CannotSetSessionDescriptionAfterClose) {
   WebRtcMedium webrtc_medium;
 
   std::unique_ptr<ConnectionFlow> offerer = ConnectionFlow::Create(
-      LocalIceCandidateListener(), DataChannelListener(), webrtc_medium);
+      LocalIceCandidateListener(), DataChannelListener(),
+      ConnectionFlow::AdapterTypeListener(), webrtc_medium);
   ASSERT_NE(offerer, nullptr);
 
   SessionDescriptionWrapper offer = offerer->CreateOffer();
@@ -195,13 +208,13 @@ TEST_F(ConnectionFlowTest, CannotSetSessionDescriptionAfterClose) {
 TEST_F(ConnectionFlowTest, CannotReceiveOfferAfterClose) {
   WebRtcMedium webrtc_medium_offerer, webrtc_medium_answerer;
 
-  std::unique_ptr<ConnectionFlow> offerer =
-      ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(),
-                             webrtc_medium_offerer);
+  std::unique_ptr<ConnectionFlow> offerer = ConnectionFlow::Create(
+      LocalIceCandidateListener(), DataChannelListener(),
+      ConnectionFlow::AdapterTypeListener(), webrtc_medium_offerer);
   ASSERT_NE(offerer, nullptr);
-  std::unique_ptr<ConnectionFlow> answerer =
-      ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(),
-                             webrtc_medium_answerer);
+  std::unique_ptr<ConnectionFlow> answerer = ConnectionFlow::Create(
+      LocalIceCandidateListener(), DataChannelListener(),
+      ConnectionFlow::AdapterTypeListener(), webrtc_medium_answerer);
   ASSERT_NE(answerer, nullptr);
 
   EXPECT_TRUE(answerer->CloseIfNotConnected());
@@ -218,8 +231,9 @@ TEST_F(ConnectionFlowTest, NullPeerConnection) {
       /*use_valid_peer_connection=*/false);
 
   WebRtcMedium medium;
-  std::unique_ptr<ConnectionFlow> answerer = ConnectionFlow::Create(
-      LocalIceCandidateListener(), DataChannelListener(), medium);
+  std::unique_ptr<ConnectionFlow> answerer =
+      ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(),
+                             ConnectionFlow::AdapterTypeListener(), medium);
   EXPECT_EQ(answerer, nullptr);
 }
 
@@ -227,15 +241,17 @@ TEST_F(ConnectionFlowTest, PeerConnectionTimeout) {
   MediumEnvironment::Instance().SetUseValidPeerConnection(
       /*use_valid_peer_connection=*/true);
   WebRtcMedium medium1;
-  std::unique_ptr<ConnectionFlow> flow1 = ConnectionFlow::Create(
-      LocalIceCandidateListener(), DataChannelListener(), medium1);
+  std::unique_ptr<ConnectionFlow> flow1 =
+      ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(),
+                             ConnectionFlow::AdapterTypeListener(), medium1);
   EXPECT_NE(flow1, nullptr);
 
   // Attempt to trigger the 2.5s peer connection timeout.
   MediumEnvironment::Instance().SetPeerConnectionLatency(absl::Seconds(5));
   WebRtcMedium medium2;
-  std::unique_ptr<ConnectionFlow> flow2 = ConnectionFlow::Create(
-      LocalIceCandidateListener(), DataChannelListener(), medium2);
+  std::unique_ptr<ConnectionFlow> flow2 =
+      ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(),
+                             ConnectionFlow::AdapterTypeListener(), medium2);
   EXPECT_EQ(flow2, nullptr);
 }
 
@@ -263,6 +279,10 @@ TEST_F(ConnectionFlowTest, TerminateAnswerer) {
            [&offerer_socket_future](WebRtcSocketWrapper socket) {
              offerer_socket_future.Set(std::move(socket));
            }},
+      {.adapter_type_changed_cb =
+           [](/*rtc::AdapterType*/ int adapter_type) {
+             // Do nothing
+           }},
       webrtc_medium_offerer);
   ASSERT_NE(offerer, nullptr);
   answerer = ConnectionFlow::Create(
@@ -279,6 +299,10 @@ TEST_F(ConnectionFlowTest, TerminateAnswerer) {
            [&answerer_socket_future](WebRtcSocketWrapper wrapper) {
              answerer_socket_future.Set(std::move(wrapper));
            }},
+      {.adapter_type_changed_cb =
+           [](/*rtc::AdapterType*/ int adapter_type) {
+             // Do nothing
+           }},
       webrtc_medium_answerer);
   ASSERT_NE(answerer, nullptr);
 
@@ -344,6 +368,10 @@ TEST_F(ConnectionFlowTest, TerminateOfferer) {
            [&offerer_socket_future](WebRtcSocketWrapper socket) {
              offerer_socket_future.Set(std::move(socket));
            }},
+      {.adapter_type_changed_cb =
+           [](/*rtc::AdapterType*/ int adapter_type) {
+             // Do nothing
+           }},
       webrtc_medium_offerer);
   ASSERT_NE(offerer, nullptr);
   answerer = ConnectionFlow::Create(
@@ -360,6 +388,10 @@ TEST_F(ConnectionFlowTest, TerminateOfferer) {
            [&answerer_socket_future](WebRtcSocketWrapper wrapper) {
              answerer_socket_future.Set(std::move(wrapper));
            }},
+      {.adapter_type_changed_cb =
+           [](/*rtc::AdapterType*/ int adapter_type) {
+             // Do nothing
+           }},
       webrtc_medium_answerer);
   ASSERT_NE(answerer, nullptr);
 

internal/platform/feature_flags.h

@@ -114,6 +114,7 @@ class FeatureFlags {
     // from triggering an OutOfMemory error.
     std::uint32_t connection_max_frame_length = 1048576;
     std::uint32_t blocking_queue_stream_queue_capacity = 10;
+    bool support_web_rtc_non_cellular_medium = false;
   };
 
   static const FeatureFlags& GetInstance() {

internal/platform/webrtc.h

@@ -24,6 +24,7 @@
 
 #include "absl/strings/string_view.h"
 #include "internal/platform/byte_array.h"
+#include "internal/platform/feature_flags.h"
 #include "internal/platform/implementation/platform.h"
 #include "internal/platform/implementation/webrtc.h"
 #include "webrtc/api/peer_connection_interface.h"
@@ -87,9 +88,10 @@ class WebRtcMedium {
   // |callback|.
   void CreatePeerConnection(webrtc::PeerConnectionObserver* observer,
                             PeerConnectionCallback callback) {
-    // TODO(edwinwu): Add a flag to control this and add support for
-    // non-cellular networks.
-    if (non_cellular_) {
+    if (FeatureFlags::GetInstance()
+            .GetFlags()
+            .support_web_rtc_non_cellular_medium) {
+      // TODO(edwinwu): Add support for non-cellular networks.
       impl_->CreatePeerConnection(std::nullopt, observer, std::move(callback));
     } else {
       impl_->CreatePeerConnection(observer, std::move(callback));