From e5f80e9fdc8a1ca4859db0e0672c2f9419eafd41 Mon Sep 17 00:00:00 2001 From: Edwin Wu Date: Wed, 16 Oct 2024 01:15:18 -0700 Subject: [PATCH] Add WEB_RTC_NON_CELLULAR enum, IV PiperOrigin-RevId: 686408097 --- connections/implementation/mediums/webrtc.cc | 29 +++++++ connections/implementation/mediums/webrtc.h | 8 ++ .../implementation/mediums/webrtc/BUILD | 2 +- .../mediums/webrtc/connection_flow.cc | 23 ++++-- .../mediums/webrtc/connection_flow.h | 29 ++++++- .../mediums/webrtc/connection_flow_test.cc | 82 +++++++++++++------ internal/platform/feature_flags.h | 1 + internal/platform/webrtc.h | 8 +- 8 files changed, 143 insertions(+), 39 deletions(-) diff --git a/connections/implementation/mediums/webrtc.cc b/connections/implementation/mediums/webrtc.cc index 5b8082f5..80b6bd27 100644 --- a/connections/implementation/mediums/webrtc.cc +++ b/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 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 diff --git a/connections/implementation/mediums/webrtc.h b/connections/implementation/mediums/webrtc.h index 8e0796d7..88314843 100644 --- a/connections/implementation/mediums/webrtc.h +++ b/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 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> connection_flows_ ABSL_GUARDED_BY(mutex_); + + bool is_using_cellular_ ABSL_GUARDED_BY(mutex_) = true; }; } // namespace mediums diff --git a/connections/implementation/mediums/webrtc/BUILD b/connections/implementation/mediums/webrtc/BUILD index d00491c9..ff2f5ca0 100644 --- a/connections/implementation/mediums/webrtc/BUILD +++ b/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", diff --git a/connections/implementation/mediums/webrtc/connection_flow.cc b/connections/implementation/mediums/webrtc/connection_flow.cc index ef01a825..83a56fe6 100644 --- a/connections/implementation/mediums/webrtc/connection_flow.cc +++ b/connections/implementation/mediums/webrtc/connection_flow.cc @@ -18,14 +18,21 @@ #include #include +#include +#include #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::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::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 diff --git a/connections/implementation/mediums/webrtc/connection_flow.h b/connections/implementation/mediums/webrtc/connection_flow.h index 40fe9d0e..d1c0da5f 100644 --- a/connections/implementation/mediums/webrtc/connection_flow.h +++ b/connections/implementation/mediums/webrtc/connection_flow.h @@ -18,14 +18,20 @@ #ifndef NO_WEBRTC #include +#include +#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 + adapter_type_changed_cb = DefaultCallback(); + }; + // 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 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 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 @@ -223,6 +242,8 @@ class ConnectionFlow : public webrtc::PeerConnectionObserver { // the former is thread-safe. std::shared_ptr can_run_tasks_ = std::make_shared(); + AdapterTypeListener adapter_type_listener_; + friend class CreateSessionDescriptionObserverImpl; }; diff --git a/connections/implementation/mediums/webrtc/connection_flow_test.cc b/connections/implementation/mediums/webrtc/connection_flow_test.cc index f0e1d6bf..5d349fb6 100644 --- a/connections/implementation/mediums/webrtc/connection_flow_test.cc +++ b/connections/implementation/mediums/webrtc/connection_flow_test.cc @@ -15,21 +15,23 @@ #include "connections/implementation/mediums/webrtc/connection_flow.h" #include +#include +#include #include -#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 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 offerer = - ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(), - webrtc_medium_offerer); + std::unique_ptr offerer = ConnectionFlow::Create( + LocalIceCandidateListener(), DataChannelListener(), + ConnectionFlow::AdapterTypeListener(), webrtc_medium_offerer); ASSERT_NE(offerer, nullptr); - std::unique_ptr answerer = - ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(), - webrtc_medium_answerer); + std::unique_ptr 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 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 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 offerer = - ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(), - webrtc_medium_offerer); + std::unique_ptr offerer = ConnectionFlow::Create( + LocalIceCandidateListener(), DataChannelListener(), + ConnectionFlow::AdapterTypeListener(), webrtc_medium_offerer); ASSERT_NE(offerer, nullptr); - std::unique_ptr answerer = - ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(), - webrtc_medium_answerer); + std::unique_ptr 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 answerer = ConnectionFlow::Create( - LocalIceCandidateListener(), DataChannelListener(), medium); + std::unique_ptr 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 flow1 = ConnectionFlow::Create( - LocalIceCandidateListener(), DataChannelListener(), medium1); + std::unique_ptr 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 flow2 = ConnectionFlow::Create( - LocalIceCandidateListener(), DataChannelListener(), medium2); + std::unique_ptr 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); diff --git a/internal/platform/feature_flags.h b/internal/platform/feature_flags.h index 7dc76c49..f1237202 100644 --- a/internal/platform/feature_flags.h +++ b/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() { diff --git a/internal/platform/webrtc.h b/internal/platform/webrtc.h index 47873ddd..96a83f9b 100644 --- a/internal/platform/webrtc.h +++ b/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));