diff --git a/connections/implementation/BUILD b/connections/implementation/BUILD index 61c37dc0..dea9acf3 100644 --- a/connections/implementation/BUILD +++ b/connections/implementation/BUILD @@ -155,7 +155,8 @@ cc_library( "//connections/implementation/flags:connections_flags", "//connections/implementation/mediums", "//connections/implementation/mediums:utils", - "//connections/implementation/mediums:webrtc_utils", + "//connections/implementation/mediums:webrtc_peer_id", + "//connections/implementation/mediums:webrtc_socket", "//connections/implementation/mediums/advertisements:dct_advertisement", "//connections/implementation/mediums/advertisements:util", "//connections/implementation/mediums/ble:ble_advertisement_header", @@ -315,10 +316,9 @@ cc_test( ":internal_test", ":types", "//connections:core_types", - "//connections/implementation/analytics", "//connections/implementation/flags:connections_flags", "//connections/implementation/mediums", - "//connections/implementation/mediums:webrtc_utils", + "//connections/implementation/mediums:webrtc_peer_id", "//connections/implementation/proto:offline_wire_formats_cc_proto", "//connections/v3:v3_types", "//internal/analytics:mock_event_logger", diff --git a/connections/implementation/mediums/BUILD b/connections/implementation/mediums/BUILD index 0d8aea37..e215226d 100644 --- a/connections/implementation/mediums/BUILD +++ b/connections/implementation/mediums/BUILD @@ -50,7 +50,8 @@ cc_library( ], deps = [ ":utils", - ":webrtc_utils", + ":webrtc_peer_id", + ":webrtc_socket", "//connections:core_types", "//connections/implementation:types", "//connections/implementation/flags:connections_flags", @@ -71,7 +72,6 @@ cc_library( "//internal/platform/flags:platform_flags", "//internal/platform/implementation:comm", "//internal/platform/implementation:platform", - "//internal/platform/implementation:wifi_utils", "//proto/mediums:web_rtc_signaling_frames_cc_proto", # "//third_party/webrtc/files/stable/webrtc/api:create_peerconnection_factory", # buildcleaner: keep # "//third_party/webrtc/files/stable/webrtc/api:jsep", @@ -90,15 +90,9 @@ cc_library( ) cc_library( - name = "webrtc_utils", - srcs = [ - "webrtc_peer_id.cc", - ], - hdrs = [ - "webrtc_peer_id.h", - "webrtc_socket.h", - "webrtc_socket_stub.h", - ], + name = "webrtc_peer_id", + srcs = ["webrtc_peer_id.cc"], + hdrs = ["webrtc_peer_id.h"], visibility = [ "//connections/implementation:__pkg__", "//connections/implementation/mediums:__pkg__", @@ -106,7 +100,6 @@ cc_library( ], deps = [ ":utils", - "//connections/implementation/mediums/webrtc:data_types", "//internal/platform:base", "@com_google_absl//absl/strings", ], @@ -141,6 +134,16 @@ cc_library( ], ) +cc_library( + name = "webrtc_socket", + hdrs = ["webrtc_socket.h"], + visibility = ["//connections/implementation:__subpackages__"], + deps = [ + "//internal/platform:base", + "@com_google_absl//absl/strings:string_view", + ], +) + cc_test( name = "core_internal_mediums_test", size = "small", @@ -197,7 +200,8 @@ cc_test( ], deps = [ ":mediums", - ":webrtc_utils", + ":webrtc_peer_id", + ":webrtc_socket", "//internal/platform:base", "//internal/platform:cancellation_flag", "//internal/platform:comm", diff --git a/connections/implementation/mediums/webrtc.cc b/connections/implementation/mediums/webrtc.cc index c0fc4a56..1c412e8c 100644 --- a/connections/implementation/mediums/webrtc.cc +++ b/connections/implementation/mediums/webrtc.cc @@ -207,13 +207,13 @@ void WebRtc::StopAcceptingConnections(const std::string& service_id) { << service_id; } -ErrorOr WebRtc::Connect( +ErrorOr> WebRtc::Connect( const std::string& service_id, const WebrtcPeerId& remote_peer_id, const LocationHint& location_hint, CancellationFlag* cancellation_flag, bool non_cellular) { service_id_to_connect_attempts_count_map_[service_id] = 1; medium_->SetNonCellular(non_cellular); - ErrorOr wrapper_result = { + ErrorOr> wrapper_result = { Error(OperationResultCode::DETAIL_UNKNOWN)}; while (service_id_to_connect_attempts_count_map_[service_id] <= kConnectAttemptsLimit) { @@ -242,12 +242,12 @@ ErrorOr WebRtc::Connect( return {Error(wrapper_result.error().operation_result_code().value())}; } -ErrorOr WebRtc::AttemptToConnect( +ErrorOr> WebRtc::AttemptToConnect( const std::string& service_id, const WebrtcPeerId& remote_peer_id, const LocationHint& location_hint, CancellationFlag* cancellation_flag) { ConnectionRequestInfo info = ConnectionRequestInfo(); info.self_peer_id = WebrtcPeerId::FromRandom(); - Future socket_future = info.socket_future; + Future> socket_future = info.socket_future; // `listener` will go out of scope at the end of `AttemptToConnect`, and this // is expected. This `listener` is tied to `socket_future` which we block on @@ -329,7 +329,7 @@ ErrorOr WebRtc::AttemptToConnect( // Wait for the connection to go through. Don't hold the mutex here so that // we're not blocking necessary operations. - ExceptionOr socket_result = + ExceptionOr> socket_result = socket_future.Get(kDataChannelTimeout); { @@ -660,9 +660,9 @@ void WebRtc::RestartTachyonReceiveMessages(const std::string& service_id) { << service_id; } -void WebRtc::ProcessDataChannelOpen(const std::string& service_id, - const WebrtcPeerId& remote_peer_id, - WebRtcSocketWrapper socket_wrapper) { +void WebRtc::ProcessDataChannelOpen( + const std::string& service_id, const WebrtcPeerId& remote_peer_id, + std::shared_ptr socket_wrapper) { MutexLock lock(&mutex_); // Notify the client of the newly formed socket. @@ -683,7 +683,7 @@ void WebRtc::ProcessDataChannelOpen(const std::string& service_id, } // No one to handle the newly created DataChannel, so we'll just close it. - socket_wrapper.Close(); + socket_wrapper->Close(); LOG(INFO) << "Ignoring new DataChannel because we are not accepting " "connections for service " << service_id; @@ -719,7 +719,8 @@ std::unique_ptr WebRtc::CreateConnectionFlow( }}}, { .data_channel_open_cb = {[this, service_id, remote_peer_id]( - WebRtcSocketWrapper socket_wrapper) { + std::shared_ptr + socket_wrapper) { OffloadFromThread( "rtc-channel-created", [this, service_id, remote_peer_id, socket_wrapper]() { diff --git a/connections/implementation/mediums/webrtc.h b/connections/implementation/mediums/webrtc.h index 66b84b45..47fcbc52 100644 --- a/connections/implementation/mediums/webrtc.h +++ b/connections/implementation/mediums/webrtc.h @@ -50,7 +50,7 @@ class WebRtc { public: // Callback that is invoked when a new connection is accepted. using AcceptedConnectionCallback = absl::AnyInvocable; + const std::string& service_id, std::shared_ptr socket)>; WebRtc(); ~WebRtc(); @@ -85,7 +85,7 @@ class WebRtc { // Initiates a WebRtc connection with peer device identified by |peer_id| // with internal retry for maximum attempts of kConnectAttemptsLimit. // Runs on @MainThread. - ErrorOr Connect( + ErrorOr> Connect( const std::string& service_id, const WebrtcPeerId& peer_id, const location::nearby::connections::LocationHint& location_hint, CancellationFlag* cancellation_flag, bool non_cellular) @@ -145,13 +145,13 @@ class WebRtc { // The pending DataChannel future. Our client will be blocked on this while // they wait for us to set up the channel over Tachyon. - Future socket_future; + Future> socket_future; }; // Attempt to initiates a WebRtc connection with peer device identified by // |peer_id|. // Runs on @MainThread. - ErrorOr AttemptToConnect( + ErrorOr> AttemptToConnect( const std::string& service_id, const WebrtcPeerId& peer_id, const location::nearby::connections::LocationHint& location_hint, CancellationFlag* cancellation_flag) ABSL_LOCKS_EXCLUDED(mutex_); @@ -214,7 +214,7 @@ class WebRtc { // Runs on |single_thread_executor_|. void ProcessDataChannelOpen(const std::string& service_id, const WebrtcPeerId& remote_peer_id, - WebRtcSocketWrapper socket_wrapper) + std::shared_ptr socket_wrapper) ABSL_LOCKS_EXCLUDED(mutex_); // Runs on |single_thread_executor_|. diff --git a/connections/implementation/mediums/webrtc/BUILD b/connections/implementation/mediums/webrtc/BUILD index a2b92747..3c20133b 100644 --- a/connections/implementation/mediums/webrtc/BUILD +++ b/connections/implementation/mediums/webrtc/BUILD @@ -37,9 +37,10 @@ cc_library( "//connections/implementation:__subpackages__", ], deps = [ - ":data_types", + ":webrtc_socket_impl", "//connections:core_types", - "//connections/implementation/mediums:webrtc_utils", + "//connections/implementation/mediums:webrtc_peer_id", + "//connections/implementation/mediums:webrtc_socket", "//internal/platform:base", "//internal/platform:comm", "//internal/platform:logging", @@ -57,24 +58,20 @@ cc_library( ) cc_library( - name = "data_types", - srcs = [ - "webrtc_socket_impl.cc", - ], - hdrs = [ - "webrtc_socket_impl.h", - ], - copts = [ - "-DCORE_ADAPTER_DLL", - "-DNO_WEBRTC", - ], + name = "webrtc_socket_impl", + srcs = ["webrtc_socket_impl.cc"], + hdrs = ["webrtc_socket_impl.h"], visibility = [ "//connections/implementation:__subpackages__", ], deps = [ + "//connections/implementation/mediums:webrtc_socket", "//internal/platform:base", "//internal/platform:logging", "//internal/platform:types", + # "//third_party/webrtc/files/stable/webrtc/api:data_channel_interface", + # "//third_party/webrtc/files/stable/webrtc/api:scoped_refptr", + "@com_google_absl//absl/functional:any_invocable", "@com_google_absl//absl/strings:string_view", ], ) @@ -92,9 +89,10 @@ cc_test( "requires-net:external", ], deps = [ - ":data_types", ":webrtc", - "//connections/implementation/mediums:webrtc_utils", + ":webrtc_socket_impl", + "//connections/implementation/mediums:webrtc_peer_id", + "//connections/implementation/mediums:webrtc_socket", "//internal/platform:base", "//internal/platform:comm", "//internal/platform:test_util", diff --git a/connections/implementation/mediums/webrtc/connection_flow.cc b/connections/implementation/mediums/webrtc/connection_flow.cc index 1d32ae17..bff8e441 100644 --- a/connections/implementation/mediums/webrtc/connection_flow.cc +++ b/connections/implementation/mediums/webrtc/connection_flow.cc @@ -412,10 +412,11 @@ void ConnectionFlow::OnSignalingStable() { void ConnectionFlow::CreateSocketFromDataChannel( webrtc::scoped_refptr data_channel) { LOG(INFO) << "Creating data channel socket"; - auto socket = - std::make_unique("WebRtcSocket", std::move(data_channel)); + auto socket = std::make_shared("WebRtcSocket", + std::move(data_channel)); + socket_ = socket; socket->SetSocketListener({ - .socket_ready_cb = {[this](WebRtcSocket* socket) { + .socket_ready_cb = {[this](WebRtcSocketImpl* socket) { CHECK(IsRunningOnSignalingThread()); if (!TransitionState(State::kWaitingToConnect, State::kConnected)) { LOG(ERROR) << "Data channel socket is open but connection " @@ -424,14 +425,13 @@ void ConnectionFlow::CreateSocketFromDataChannel( return; } // Pass socket wrapper by copy on purpose - data_channel_listener_.data_channel_open_cb(socket_wrapper_); + data_channel_listener_.data_channel_open_cb(socket_); }}, .socket_closed_cb = - [this](WebRtcSocket*) { + [this](WebRtcSocketImpl*) { data_channel_listener_.data_channel_closed_cb(); }, }); - socket_wrapper_ = WebRtcSocketWrapper(std::move(socket)); } void ConnectionFlow::OnIceCandidate(const webrtc::IceCandidate* candidate) { @@ -512,7 +512,7 @@ bool ConnectionFlow::CloseOnSignalingThread() { // Close the socket wrapper before terminating the PeerConnection // since the teardown process of the PC may close threads that are // otherwise depended upon by objects kept alive by the socket_wrapper. - if (socket_wrapper_.IsValid()) socket_wrapper_.Close(); + if (socket_ && socket_->IsValid()) socket_->Close(); // This prevents other tasks from queuing on the signaling thread for this // object. diff --git a/connections/implementation/mediums/webrtc/connection_flow.h b/connections/implementation/mediums/webrtc/connection_flow.h index e0986786..73e8294e 100644 --- a/connections/implementation/mediums/webrtc/connection_flow.h +++ b/connections/implementation/mediums/webrtc/connection_flow.h @@ -222,7 +222,7 @@ class ConnectionFlow : public webrtc::PeerConnectionObserver { // Used to hold a reference to the WebRtcSocket while the data channel is // connecting. - WebRtcSocketWrapper socket_wrapper_; + std::shared_ptr socket_; std::vector> cached_remote_ice_candidates_; diff --git a/connections/implementation/mediums/webrtc/connection_flow_test.cc b/connections/implementation/mediums/webrtc/connection_flow_test.cc index 64a7766d..e08aa5ae 100644 --- a/connections/implementation/mediums/webrtc/connection_flow_test.cc +++ b/connections/implementation/mediums/webrtc/connection_flow_test.cc @@ -61,7 +61,8 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) { Future message_received_future; - Future offerer_socket_future, answerer_socket_future; + Future> offerer_socket_future, + answerer_socket_future; std::unique_ptr offerer, answerer; @@ -77,7 +78,7 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) { answerer->OnRemoteIceCandidatesReceived(std::move(vec)); }}, {.data_channel_open_cb = - [&offerer_socket_future](WebRtcSocketWrapper socket) { + [&offerer_socket_future](std::shared_ptr socket) { offerer_socket_future.Set(std::move(socket)); }}, {.adapter_type_changed_cb = @@ -97,7 +98,7 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) { offerer->OnRemoteIceCandidatesReceived(std::move(vec)); }}, {.data_channel_open_cb = - [&answerer_socket_future](WebRtcSocketWrapper socket) { + [&answerer_socket_future](std::shared_ptr socket) { answerer_socket_future.Set(std::move(socket)); }}, {.adapter_type_changed_cb = @@ -122,18 +123,18 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) { EXPECT_TRUE(answerer->SetLocalSessionDescription(std::move(answer))); // Retrieve Data Channels - ExceptionOr offerer_socket = + ExceptionOr> offerer_socket = offerer_socket_future.Get(absl::Seconds(1)); EXPECT_TRUE(offerer_socket.ok()); - ExceptionOr answerer_socket = + ExceptionOr> answerer_socket = answerer_socket_future.Get(absl::Seconds(1)); EXPECT_TRUE(answerer_socket.ok()); // Send message on data channel absl::string_view message = "Test"; - offerer_socket.result().GetImpl().GetOutputStream().Write(message); + offerer_socket.result()->GetOutputStream().Write(message); ExceptionOr received_message = - answerer_socket.result().GetImpl().GetInputStream().Read(4); + answerer_socket.result()->GetInputStream().Read(4); EXPECT_TRUE(received_message.ok()); EXPECT_EQ(received_message.result(), ByteArray{message.data()}); } @@ -259,7 +260,8 @@ TEST_F(ConnectionFlowTest, TerminateAnswerer) { Future message_received_future; - Future offerer_socket_future, answerer_socket_future; + Future> offerer_socket_future, + answerer_socket_future; std::unique_ptr offerer, answerer; @@ -275,7 +277,7 @@ TEST_F(ConnectionFlowTest, TerminateAnswerer) { answerer->OnRemoteIceCandidatesReceived(std::move(vec)); }}, {.data_channel_open_cb = - [&offerer_socket_future](WebRtcSocketWrapper socket) { + [&offerer_socket_future](std::shared_ptr socket) { offerer_socket_future.Set(std::move(socket)); }}, {.adapter_type_changed_cb = @@ -295,7 +297,7 @@ TEST_F(ConnectionFlowTest, TerminateAnswerer) { offerer->OnRemoteIceCandidatesReceived(std::move(vec)); }}, {.data_channel_open_cb = - [&answerer_socket_future](WebRtcSocketWrapper wrapper) { + [&answerer_socket_future](std::shared_ptr wrapper) { answerer_socket_future.Set(std::move(wrapper)); }}, {.adapter_type_changed_cb = @@ -321,10 +323,10 @@ TEST_F(ConnectionFlowTest, TerminateAnswerer) { EXPECT_TRUE(answerer->SetLocalSessionDescription(std::move(answer))); // Retrieve Data Channels - ExceptionOr offerer_socket = + ExceptionOr> offerer_socket = offerer_socket_future.Get(absl::Seconds(1)); EXPECT_TRUE(offerer_socket.ok()); - ExceptionOr answerer_socket = + ExceptionOr> answerer_socket = answerer_socket_future.Get(absl::Seconds(1)); EXPECT_TRUE(offerer_socket.ok()); @@ -338,9 +340,9 @@ TEST_F(ConnectionFlowTest, TerminateAnswerer) { // Send message on data channel absl::string_view message = "Test"; - offerer_socket.result().GetOutputStream().Write(message); + offerer_socket.result()->GetOutputStream().Write(message); ExceptionOr received_message = - answerer_socket.result().GetInputStream().Read(4); + answerer_socket.result()->GetInputStream().Read(4); EXPECT_TRUE(received_message.GetResult().Empty()); } @@ -349,7 +351,8 @@ TEST_F(ConnectionFlowTest, TerminateOfferer) { Future message_received_future; - Future offerer_socket_future, answerer_socket_future; + Future> offerer_socket_future, + answerer_socket_future; std::unique_ptr offerer, answerer; @@ -365,7 +368,7 @@ TEST_F(ConnectionFlowTest, TerminateOfferer) { answerer->OnRemoteIceCandidatesReceived(std::move(vec)); }}, {.data_channel_open_cb = - [&offerer_socket_future](WebRtcSocketWrapper socket) { + [&offerer_socket_future](std::shared_ptr socket) { offerer_socket_future.Set(std::move(socket)); }}, {.adapter_type_changed_cb = @@ -386,7 +389,7 @@ TEST_F(ConnectionFlowTest, TerminateOfferer) { offerer->OnRemoteIceCandidatesReceived(std::move(vec)); }}, {.data_channel_open_cb = - [&answerer_socket_future](WebRtcSocketWrapper wrapper) { + [&answerer_socket_future](std::shared_ptr wrapper) { answerer_socket_future.Set(std::move(wrapper)); }}, {.adapter_type_changed_cb = @@ -412,10 +415,10 @@ TEST_F(ConnectionFlowTest, TerminateOfferer) { EXPECT_TRUE(answerer->SetLocalSessionDescription(std::move(answer))); // Retrieve Data Channels - ExceptionOr offerer_socket = + ExceptionOr> offerer_socket = offerer_socket_future.Get(absl::Seconds(1)); EXPECT_TRUE(offerer_socket.ok()); - ExceptionOr answerer_socket = + ExceptionOr> answerer_socket = answerer_socket_future.Get(absl::Seconds(1)); EXPECT_TRUE(offerer_socket.ok()); @@ -429,9 +432,9 @@ TEST_F(ConnectionFlowTest, TerminateOfferer) { // Send message on data channel absl::string_view message = "Test"; - offerer_socket.result().GetOutputStream().Write(message); + offerer_socket.result()->GetOutputStream().Write(message); ExceptionOr received_message = - answerer_socket.result().GetInputStream().Read(4); + answerer_socket.result()->GetInputStream().Read(4); EXPECT_TRUE(received_message.GetResult().Empty()); } diff --git a/connections/implementation/mediums/webrtc/data_channel_listener.h b/connections/implementation/mediums/webrtc/data_channel_listener.h index 17e6a0dc..ec679940 100644 --- a/connections/implementation/mediums/webrtc/data_channel_listener.h +++ b/connections/implementation/mediums/webrtc/data_channel_listener.h @@ -17,6 +17,8 @@ #ifndef NO_WEBRTC +#include + #include "absl/functional/any_invocable.h" #include "connections/implementation/mediums/webrtc_socket.h" @@ -28,8 +30,8 @@ namespace mediums { struct DataChannelListener { // Called when the data channel is open and the socket wrapper is ready to // read and write. - absl::AnyInvocable data_channel_open_cb = - [](WebRtcSocketWrapper) {}; + absl::AnyInvocable)> data_channel_open_cb = + [](std::shared_ptr) {}; // Called when the data channel is closed. absl::AnyInvocable data_channel_closed_cb = []() {}; @@ -39,6 +41,6 @@ struct DataChannelListener { } // namespace connections } // namespace nearby -#endif +#endif // NO_WEBRTC #endif // CORE_INTERNAL_MEDIUMS_WEBRTC_DATA_CHANNEL_LISTENER_H_ diff --git a/connections/implementation/mediums/webrtc/webrtc_socket_impl.cc b/connections/implementation/mediums/webrtc/webrtc_socket_impl.cc index aca93c67..c34160c5 100644 --- a/connections/implementation/mediums/webrtc/webrtc_socket_impl.cc +++ b/connections/implementation/mediums/webrtc/webrtc_socket_impl.cc @@ -12,6 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. +#ifndef NO_WEBRTC + +#include "connections/implementation/mediums/webrtc/webrtc_socket_impl.h" + #include #include #include @@ -21,19 +25,20 @@ #include "internal/platform/byte_array.h" #include "internal/platform/exception.h" #include "internal/platform/input_stream.h" -#include "internal/platform/pipe.h" -#ifndef NO_WEBRTC - -#include "connections/implementation/mediums/webrtc/webrtc_socket_impl.h" #include "internal/platform/logging.h" #include "internal/platform/mutex_lock.h" +#include "internal/platform/output_stream.h" +#include "internal/platform/pipe.h" +#include "internal/platform/runnable.h" +#include "webrtc/api/data_channel_interface.h" +#include "webrtc/api/scoped_refptr.h" namespace nearby { namespace connections { namespace mediums { // OutputStreamImpl -Exception WebRtcSocket::OutputStreamImpl::Write(absl::string_view data) { +Exception WebRtcSocketImpl::OutputStreamImpl::Write(absl::string_view data) { if (data.size() > kMaxDataSize) { LOG(WARNING) << "Sending data larger than 1MB"; return {Exception::kIo}; @@ -53,18 +58,18 @@ Exception WebRtcSocket::OutputStreamImpl::Write(absl::string_view data) { return {Exception::kSuccess}; } -Exception WebRtcSocket::OutputStreamImpl::Flush() { +Exception WebRtcSocketImpl::OutputStreamImpl::Flush() { // Java implementation is empty. return {Exception::kSuccess}; } -Exception WebRtcSocket::OutputStreamImpl::Close() { +Exception WebRtcSocketImpl::OutputStreamImpl::Close() { socket_->Close(); return {Exception::kSuccess}; } // WebRtcSocket -WebRtcSocket::WebRtcSocket( +WebRtcSocketImpl::WebRtcSocketImpl( const std::string& name, webrtc::scoped_refptr data_channel) : name_(name), data_channel_(std::move(data_channel)) { @@ -73,7 +78,7 @@ WebRtcSocket::WebRtcSocket( data_channel_->RegisterObserver(this); } -WebRtcSocket::~WebRtcSocket() { +WebRtcSocketImpl::~WebRtcSocketImpl() { LOG(INFO) << "WebRtcSocket::~WebRtcSocket(" << name_ << ") this: " << this; if (!IsClosed()) { @@ -85,16 +90,16 @@ WebRtcSocket::~WebRtcSocket() { << " done"; } -InputStream& WebRtcSocket::GetInputStream() { return *pipe_input_; } +InputStream& WebRtcSocketImpl::GetInputStream() { return *pipe_input_; } -OutputStream& WebRtcSocket::GetOutputStream() { return output_stream_; } +OutputStream& WebRtcSocketImpl::GetOutputStream() { return output_stream_; } -Exception WebRtcSocket::Close() { +Exception WebRtcSocketImpl::Close() { LOG(INFO) << "WebRtcSocket::Close(" << name_ << ") this: " << this; if (closed_.Set(true)) return {Exception::kSuccess}; ClosePipe(); - // NOTE: This call blocks and triggers a state change on the siginaling thread + // NOTE: This call blocks and triggers a state change on the signaling thread // to 'closing' but does not block until 'closed' is sent so the data channel // is not fully closed when this call is done. data_channel_->Close(); @@ -102,7 +107,7 @@ Exception WebRtcSocket::Close() { return {Exception::kSuccess}; } -void WebRtcSocket::OnStateChange() { +void WebRtcSocketImpl::OnStateChange() { // Running on the signaling thread right now. LOG(ERROR) << "WebRtcSocket::OnStateChange() webrtc data channel state: " << webrtc::DataChannelInterface::DataStateString( @@ -131,7 +136,7 @@ void WebRtcSocket::OnStateChange() { break; } } -void WebRtcSocket::OnMessage(const webrtc::DataBuffer& buffer) { +void WebRtcSocketImpl::OnMessage(const webrtc::DataBuffer& buffer) { // This is a data channel callback on the signaling thread, lets off load so // we don't block signaling. OffloadFromSignalingThread( @@ -147,20 +152,20 @@ void WebRtcSocket::OnMessage(const webrtc::DataBuffer& buffer) { }); } -void WebRtcSocket::OnBufferedAmountChange(uint64_t sent_data_size) { +void WebRtcSocketImpl::OnBufferedAmountChange(uint64_t sent_data_size) { // This is a data channel callback on the signaling thread, lets off load so // we don't block signaling. OffloadFromSignalingThread([this] { WakeUpWriter(); }); } -bool WebRtcSocket::SendMessage(const ByteArray& data) { +bool WebRtcSocketImpl::SendMessage(const ByteArray& data) { return data_channel_->Send( webrtc::DataBuffer(std::string(data.data(), data.size()))); } -bool WebRtcSocket::IsClosed() { return closed_.Get(); } +bool WebRtcSocketImpl::IsClosed() { return closed_.Get(); } -void WebRtcSocket::ClosePipe() { +void WebRtcSocketImpl::ClosePipe() { LOG(INFO) << "WebRtcSocket::ClosePipe(" << name_ << ") this: " << this; // This is thread-safe to close these sockets even if a read or write is in // process on another thread, Close will wait for the exclusive mutex before @@ -173,16 +178,16 @@ void WebRtcSocket::ClosePipe() { } // Must not be called on signalling thread. -void WebRtcSocket::WakeUpWriter() { +void WebRtcSocketImpl::WakeUpWriter() { MutexLock lock(&backpressure_mutex_); buffer_variable_.Notify(); } -void WebRtcSocket::SetSocketListener(SocketListener&& listener) { +void WebRtcSocketImpl::SetSocketListener(SocketListener&& listener) { socket_listener_ = std::move(listener); } -void WebRtcSocket::BlockUntilSufficientSpaceInBuffer(int length) { +void WebRtcSocketImpl::BlockUntilSufficientSpaceInBuffer(int length) { MutexLock lock(&backpressure_mutex_); while (!IsClosed() && (data_channel_->buffered_amount() + length > kMaxDataSize)) { @@ -191,7 +196,7 @@ void WebRtcSocket::BlockUntilSufficientSpaceInBuffer(int length) { } } -void WebRtcSocket::OffloadFromSignalingThread(Runnable runnable) { +void WebRtcSocketImpl::OffloadFromSignalingThread(Runnable runnable) { single_thread_executor_.Execute(std::move(runnable)); } @@ -199,4 +204,4 @@ void WebRtcSocket::OffloadFromSignalingThread(Runnable runnable) { } // namespace connections } // namespace nearby -#endif +#endif // NO_WEBRTC diff --git a/connections/implementation/mediums/webrtc/webrtc_socket_impl.h b/connections/implementation/mediums/webrtc/webrtc_socket_impl.h index 109367b7..071522ef 100644 --- a/connections/implementation/mediums/webrtc/webrtc_socket_impl.h +++ b/connections/implementation/mediums/webrtc/webrtc_socket_impl.h @@ -15,23 +15,27 @@ #ifndef CORE_INTERNAL_MEDIUMS_WEBRTC_WEBRTC_SOCKET_IMPL_H_ #define CORE_INTERNAL_MEDIUMS_WEBRTC_WEBRTC_SOCKET_IMPL_H_ -#include -#include -#include - -#include "absl/strings/string_view.h" -#include "internal/platform/byte_array.h" -#include "internal/platform/exception.h" -#include "internal/platform/listeners.h" -#include "internal/platform/runnable.h" #ifndef NO_WEBRTC + +#include +#include +#include + +#include "absl/functional/any_invocable.h" +#include "absl/strings/string_view.h" +#include "connections/implementation/mediums/webrtc_socket.h" #include "internal/platform/atomic_boolean.h" +#include "internal/platform/byte_array.h" #include "internal/platform/condition_variable.h" +#include "internal/platform/exception.h" #include "internal/platform/input_stream.h" +#include "internal/platform/listeners.h" #include "internal/platform/mutex.h" #include "internal/platform/output_stream.h" +#include "internal/platform/runnable.h" #include "internal/platform/single_thread_executor.h" -#include "internal/platform/socket.h" +#include "webrtc/api/data_channel_interface.h" +#include "webrtc/api/scoped_refptr.h" namespace nearby { namespace connections { @@ -45,20 +49,22 @@ constexpr int kMaxDataSize = 1 * 1024 * 1024; // // Messages are buffered here to prevent the data channel from overflowing, // which could lead to data loss. -class WebRtcSocket : public Socket, public webrtc::DataChannelObserver { +class WebRtcSocketImpl : public WebRtcSocket, + public webrtc::DataChannelObserver { public: - WebRtcSocket( + WebRtcSocketImpl( const std::string& name, webrtc::scoped_refptr data_channel); - ~WebRtcSocket() override; + ~WebRtcSocketImpl() override; - WebRtcSocket(const WebRtcSocket& other) = delete; - WebRtcSocket& operator=(const WebRtcSocket& other) = delete; + WebRtcSocketImpl(const WebRtcSocketImpl& other) = delete; + WebRtcSocketImpl& operator=(const WebRtcSocketImpl& other) = delete; - // Overrides for nearby::Socket: + // Overrides for WebRtcSocket: InputStream& GetInputStream() override; OutputStream& GetOutputStream() override; Exception Close() override; + bool IsValid() const override { return true; } // webrtc::DataChannelObserver: void OnStateChange() override; @@ -67,10 +73,10 @@ class WebRtcSocket : public Socket, public webrtc::DataChannelObserver { // Listener class the gets called when the socket is ready or closed struct SocketListener { - absl::AnyInvocable socket_ready_cb = - DefaultCallback(); - absl::AnyInvocable socket_closed_cb = - DefaultCallback(); + absl::AnyInvocable socket_ready_cb = + DefaultCallback(); + absl::AnyInvocable socket_closed_cb = + DefaultCallback(); }; void SetSocketListener(SocketListener&& listener); @@ -78,7 +84,8 @@ class WebRtcSocket : public Socket, public webrtc::DataChannelObserver { private: class OutputStreamImpl : public OutputStream { public: - explicit OutputStreamImpl(WebRtcSocket* const socket) : socket_(socket) {} + explicit OutputStreamImpl(WebRtcSocketImpl* const socket) + : socket_(socket) {} ~OutputStreamImpl() override = default; OutputStreamImpl(const OutputStreamImpl& other) = delete; @@ -91,7 +98,7 @@ class WebRtcSocket : public Socket, public webrtc::DataChannelObserver { private: // |this| OutputStreamImpl is owned by |socket_|. - WebRtcSocket* const socket_; + WebRtcSocketImpl* const socket_; }; void WakeUpWriter(); @@ -124,6 +131,6 @@ class WebRtcSocket : public Socket, public webrtc::DataChannelObserver { } // namespace connections } // namespace nearby -#endif +#endif // NO_WEBRTC #endif // CORE_INTERNAL_MEDIUMS_WEBRTC_WEBRTC_SOCKET_IMPL_H_ diff --git a/connections/implementation/mediums/webrtc/webrtc_socket_impl_test.cc b/connections/implementation/mediums/webrtc/webrtc_socket_impl_test.cc index 78ec1771..b7f46c3a 100644 --- a/connections/implementation/mediums/webrtc/webrtc_socket_impl_test.cc +++ b/connections/implementation/mediums/webrtc/webrtc_socket_impl_test.cc @@ -64,7 +64,7 @@ TEST(WebRtcSocketTest, ReadFromSocket) { const char* message = "message"; webrtc::scoped_refptr mock_data_channel( new MockDataChannel()); - WebRtcSocket webrtc_socket(kSocketName, mock_data_channel); + WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel); webrtc_socket.OnMessage(webrtc::DataBuffer{message}); ExceptionOr result = webrtc_socket.GetInputStream().Read(7); @@ -75,7 +75,7 @@ TEST(WebRtcSocketTest, ReadFromSocket) { TEST(WebRtcSocketTest, ReadMultipleMessages) { webrtc::scoped_refptr mock_data_channel( new MockDataChannel()); - WebRtcSocket webrtc_socket(kSocketName, mock_data_channel); + WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel); webrtc_socket.OnMessage(webrtc::DataBuffer{"Me"}); webrtc_socket.OnMessage(webrtc::DataBuffer{"ssa"}); @@ -101,7 +101,7 @@ TEST(WebRtcSocketTest, WriteToSocket) { absl::string_view kMessage{"Message"}; webrtc::scoped_refptr mock_data_channel( new MockDataChannel()); - WebRtcSocket webrtc_socket(kSocketName, mock_data_channel); + WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel); EXPECT_CALL(*mock_data_channel, Send(testing::_)) .WillRepeatedly(testing::Return(true)); @@ -112,7 +112,7 @@ TEST(WebRtcSocketTest, SendDataBiggerThanMax) { std::string kMessage(kMaxDataSize + 1, '0'); webrtc::scoped_refptr mock_data_channel( new MockDataChannel()); - WebRtcSocket webrtc_socket(kSocketName, mock_data_channel); + WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel); EXPECT_CALL(*mock_data_channel, Send(testing::_)).Times(0); EXPECT_EQ(webrtc_socket.GetOutputStream().Write(kMessage), @@ -123,7 +123,7 @@ TEST(WebRtcSocketTest, WriteToDataChannelFails) { absl::string_view kMessage{"Message"}; webrtc::scoped_refptr mock_data_channel( new MockDataChannel()); - WebRtcSocket webrtc_socket(kSocketName, mock_data_channel); + WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel); ON_CALL(*mock_data_channel, Send(testing::_)) .WillByDefault(testing::Return(false)); @@ -134,14 +134,14 @@ TEST(WebRtcSocketTest, WriteToDataChannelFails) { TEST(WebRtcSocketTest, Close) { webrtc::scoped_refptr mock_data_channel( new MockDataChannel()); - WebRtcSocket webrtc_socket(kSocketName, mock_data_channel); + WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel); EXPECT_CALL(*mock_data_channel, Close()); int socket_closed_cb_called = 0; webrtc_socket.SetSocketListener( - {.socket_closed_cb = [&](WebRtcSocket* socket) { + {.socket_closed_cb = [&](WebRtcSocketImpl* socket) { socket_closed_cb_called++; }}); webrtc_socket.Close(); @@ -160,7 +160,7 @@ TEST(WebRtcSocketTest, WriteOnClosedChannel) { absl::string_view kMessage{"Message"}; webrtc::scoped_refptr mock_data_channel( new MockDataChannel()); - WebRtcSocket webrtc_socket(kSocketName, mock_data_channel); + WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel); webrtc_socket.Close(); EXPECT_CALL(*mock_data_channel, Send(testing::_)).Times(0); @@ -172,7 +172,7 @@ TEST(WebRtcSocketTest, ReadFromClosedChannel) { absl::string_view kMessage{"Message"}; webrtc::scoped_refptr mock_data_channel( new MockDataChannel()); - WebRtcSocket webrtc_socket(kSocketName, mock_data_channel); + WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel); ON_CALL(*mock_data_channel, Send(testing::_)) .WillByDefault(testing::Return(true)); @@ -185,7 +185,7 @@ TEST(WebRtcSocketTest, ReadFromClosedChannel) { TEST(WebRtcSocketTest, DataChannelCloseEventCleansUp) { webrtc::scoped_refptr mock_data_channel( new MockDataChannel()); - WebRtcSocket webrtc_socket(kSocketName, mock_data_channel); + WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel); ON_CALL(*mock_data_channel, state()) .WillByDefault( @@ -203,12 +203,12 @@ TEST(WebRtcSocketTest, DataChannelCloseEventCleansUp) { TEST(WebRtcSocketTest, OpenStateTriggersCallback) { webrtc::scoped_refptr mock_data_channel( new MockDataChannel()); - WebRtcSocket webrtc_socket(kSocketName, mock_data_channel); + WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel); int socket_ready_cb_called = 0; webrtc_socket.SetSocketListener( - {.socket_ready_cb = [&](WebRtcSocket* socket) { + {.socket_ready_cb = [&](WebRtcSocketImpl* socket) { socket_ready_cb_called++; }}); @@ -224,12 +224,12 @@ TEST(WebRtcSocketTest, OpenStateTriggersCallback) { TEST(WebRtcSocketTest, CloseStateTriggersCallback) { webrtc::scoped_refptr mock_data_channel( new MockDataChannel()); - WebRtcSocket webrtc_socket(kSocketName, mock_data_channel); + WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel); int socket_closed_cb_called = 0; webrtc_socket.SetSocketListener( - {.socket_closed_cb = [&](WebRtcSocket* socket) { + {.socket_closed_cb = [&](WebRtcSocketImpl* socket) { socket_closed_cb_called++; }}); diff --git a/connections/implementation/mediums/webrtc_socket.h b/connections/implementation/mediums/webrtc_socket.h index 724e247a..3881c2f0 100644 --- a/connections/implementation/mediums/webrtc_socket.h +++ b/connections/implementation/mediums/webrtc_socket.h @@ -1,4 +1,4 @@ -// Copyright 2020 Google LLC +// Copyright 2026 Google LLC // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -15,44 +15,57 @@ #ifndef CORE_INTERNAL_MEDIUMS_WEBRTC_SOCKET_H_ #define CORE_INTERNAL_MEDIUMS_WEBRTC_SOCKET_H_ +#include + +#include "absl/strings/string_view.h" +#include "internal/platform/byte_array.h" #include "internal/platform/exception.h" -#ifndef NO_WEBRTC - -#include - -#include "connections/implementation/mediums/webrtc/webrtc_socket_impl.h" +#include "internal/platform/input_stream.h" +#include "internal/platform/output_stream.h" +#include "internal/platform/socket.h" namespace nearby { namespace connections { namespace mediums { -class WebRtcSocketWrapper final { +// A base implementation that creates a non-working WebRtcSocket that can be +// used as a placeholder when WebRTC is disabled. +class WebRtcSocket : public Socket { public: - WebRtcSocketWrapper() = default; - WebRtcSocketWrapper(const WebRtcSocketWrapper&) = default; - WebRtcSocketWrapper& operator=(const WebRtcSocketWrapper&) = default; - explicit WebRtcSocketWrapper(std::unique_ptr socket) - : impl_(socket.release()) {} - ~WebRtcSocketWrapper() = default; + ~WebRtcSocket() override = default; - InputStream& GetInputStream() { return impl_->GetInputStream(); } + InputStream& GetInputStream() override { return fake_input_stream_; } - OutputStream& GetOutputStream() { return impl_->GetOutputStream(); } + OutputStream& GetOutputStream() override { return fake_output_stream_; } - Exception Close() { return impl_->Close(); } + Exception Close() override { return {Exception::kSuccess}; } - bool IsValid() const { return impl_ != nullptr; } - - WebRtcSocket& GetImpl() { return *impl_; } + virtual bool IsValid() const { return false; } private: - std::shared_ptr impl_; + class FakeInputStream : public InputStream { + public: + ExceptionOr Read(std::int64_t size) override { + return {Exception::kSuccess}; + } + Exception Close() override { return {Exception::kSuccess}; } + }; + + class FakeOutputStream : public OutputStream { + public: + Exception Write(absl::string_view data) override { + return {Exception::kSuccess}; + } + Exception Flush() override { return {Exception::kSuccess}; } + Exception Close() override { return {Exception::kSuccess}; } + }; + + FakeInputStream fake_input_stream_; + FakeOutputStream fake_output_stream_; }; } // namespace mediums } // namespace connections } // namespace nearby -#endif - -#endif // CORE_INTERNAL_MEDIUMS_WEBRTC_WEBRTC_SOCKET_H_ +#endif // CORE_INTERNAL_MEDIUMS_WEBRTC_SOCKET_H_ diff --git a/connections/implementation/mediums/webrtc_socket_stub.h b/connections/implementation/mediums/webrtc_socket_stub.h deleted file mode 100644 index f3f104ef..00000000 --- a/connections/implementation/mediums/webrtc_socket_stub.h +++ /dev/null @@ -1,71 +0,0 @@ -// Copyright 2020 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// https://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#ifndef CORE_INTERNAL_MEDIUMS_WEBRTC_SOCKET_STUB_H_ -#define CORE_INTERNAL_MEDIUMS_WEBRTC_SOCKET_STUB_H_ - -#ifdef NO_WEBRTC - -#include - -#include "internal/platform/input_stream.h" -#include "internal/platform/output_stream.h" - -namespace nearby { -namespace connections { -namespace mediums { -class FakeInputStream : public InputStream { - public: - ExceptionOr Read(std::int64_t size) { - return {Exception::kSuccess}; - } - Exception Close() { return {Exception::kSuccess}; } -}; - -class FakeOutputStream : public OutputStream { - public: - Exception Write(absl::string_view data) override { - return {Exception::kSuccess}; - } - Exception Flush() override { return {Exception::kSuccess}; } - Exception Close() override { return {Exception::kSuccess}; } -}; - -class WebRtcSocketWrapper final { - public: - WebRtcSocketWrapper() = default; - WebRtcSocketWrapper(const WebRtcSocketWrapper&) = default; - WebRtcSocketWrapper& operator=(const WebRtcSocketWrapper&) = default; - ~WebRtcSocketWrapper() = default; - - InputStream& GetInputStream() { return fake_input_stream_; } - - OutputStream& GetOutputStream() { return fake_output_stream_; } - - void Close() {} - - bool IsValid() const { return false; } - - private: - FakeInputStream fake_input_stream_; - FakeOutputStream fake_output_stream_; -}; - -} // namespace mediums -} // namespace connections -} // namespace nearby - -#endif - -#endif // CORE_INTERNAL_MEDIUMS_WEBRTC_SOCKET_STUB_H_ diff --git a/connections/implementation/mediums/webrtc_stub.cc b/connections/implementation/mediums/webrtc_stub.cc index 385fc882..b50a7bf4 100644 --- a/connections/implementation/mediums/webrtc_stub.cc +++ b/connections/implementation/mediums/webrtc_stub.cc @@ -19,7 +19,7 @@ #include #include -#include "connections/implementation/mediums/webrtc_socket_stub.h" +#include "connections/implementation/mediums/webrtc_socket.h" #include "internal/platform/cancelable_alarm.h" #include "internal/platform/expected.h" #include "internal/platform/future.h" @@ -52,7 +52,7 @@ bool WebRtc::StartAcceptingConnections(const std::string& service_id, void WebRtc::StopAcceptingConnections(const std::string& service_id) {} -ErrorOr WebRtc::Connect( +ErrorOr> WebRtc::Connect( const std::string& service_id, const WebrtcPeerId& remote_peer_id, const LocationHint& location_hint, CancellationFlag* cancellation_flag) { return {Error(OperationResultCode::DETAIL_UNKNOWN)}; diff --git a/connections/implementation/mediums/webrtc_stub.h b/connections/implementation/mediums/webrtc_stub.h index 832fc38f..ef5ddfb9 100644 --- a/connections/implementation/mediums/webrtc_stub.h +++ b/connections/implementation/mediums/webrtc_stub.h @@ -23,7 +23,7 @@ #include #include "connections/implementation/mediums/webrtc_peer_id.h" -#include "connections/implementation/mediums/webrtc_socket_stub.h" +#include "connections/implementation/mediums/webrtc_socket.h" #include "connections/implementation/proto/offline_wire_formats.pb.h" #include "internal/platform/cancellation_flag.h" #include "internal/platform/expected.h" @@ -38,7 +38,7 @@ class WebRtc { public: // Callback that is invoked when a new connection is accepted. using AcceptedConnectionCallback = - absl::AnyInvocable; + absl::AnyInvocable socket)>; WebRtc(); ~WebRtc(); @@ -69,7 +69,7 @@ class WebRtc { // Initiates a WebRtc connection with peer device identified by |peer_id| // with internal retry for maximum attempts of kConnectAttemptsLimit. // Runs on @MainThread. - ErrorOr Connect( + ErrorOr> Connect( const std::string& service_id, const WebrtcPeerId& peer_id, const location::nearby::connections::LocationHint& location_hint, CancellationFlag* cancellation_flag); diff --git a/connections/implementation/mediums/webrtc_test.cc b/connections/implementation/mediums/webrtc_test.cc index 061dbd4c..fbd4da27 100644 --- a/connections/implementation/mediums/webrtc_test.cc +++ b/connections/implementation/mediums/webrtc_test.cc @@ -61,7 +61,7 @@ class TestWebRtc : public WebRtc { class WebRtcTest : public ::testing::TestWithParam { protected: using MockAcceptedCallback = testing::MockFunction; + const std::string& service_id, std::shared_ptr socket)>; MediumEnvironment& env_{MediumEnvironment::Instance()}; }; @@ -73,7 +73,7 @@ TEST_P(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) { WebRtcTestParams params = GetParam(); env_.SetFeatureFlags(params.feature_flags); WebRtc receiver, sender; - WebRtcSocketWrapper receiver_socket; + std::shared_ptr receiver_socket; const WebrtcPeerId self_id("self_id"); const std::string service_id("NearbySharing"); LocationHint location_hint; @@ -82,18 +82,19 @@ TEST_P(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) { receiver.StartAcceptingConnections( service_id, self_id, location_hint, - [&receiver_socket, connected](const std::string& service_id, - WebRtcSocketWrapper wrapper) mutable { + [&receiver_socket, connected]( + const std::string& service_id, + std::shared_ptr wrapper) mutable { receiver_socket = wrapper; - connected.Set(receiver_socket.IsValid()); + connected.Set(receiver_socket->IsValid()); }, params.non_cellular); CancellationFlag flag; - ErrorOr sender_socket_result = sender.Connect( + ErrorOr> sender_socket_result = sender.Connect( service_id, self_id, location_hint, &flag, params.non_cellular); EXPECT_TRUE(sender_socket_result.has_value()); - EXPECT_TRUE(sender_socket_result.value().IsValid()); + EXPECT_TRUE(sender_socket_result.value()->IsValid()); ExceptionOr devices_connected = connected.Get(); ASSERT_TRUE(devices_connected.ok()); @@ -102,9 +103,9 @@ TEST_P(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) { // Only shuts down signaling channel. receiver.StopAcceptingConnections(service_id); - sender_socket_result.value().GetOutputStream().Write(message); + sender_socket_result.value()->GetOutputStream().Write(message); ExceptionOr received_msg = - receiver_socket.GetInputStream().Read(/*size=*/32); + receiver_socket->GetInputStream().Read(/*size=*/32); ASSERT_TRUE(received_msg.ok()); EXPECT_EQ(message, received_msg.result().AsStringView()); env_.Stop(); @@ -115,7 +116,7 @@ TEST_P(WebRtcTest, CanCancelConnect) { WebRtcTestParams params = GetParam(); env_.SetFeatureFlags(params.feature_flags); WebRtc receiver, sender; - WebRtcSocketWrapper receiver_socket; + std::shared_ptr receiver_socket; const WebrtcPeerId self_id("self_id"); const std::string service_id("NearbySharing"); LocationHint location_hint; @@ -124,32 +125,33 @@ TEST_P(WebRtcTest, CanCancelConnect) { receiver.StartAcceptingConnections( service_id, self_id, location_hint, - [&receiver_socket, connected](const std::string& service_id, - WebRtcSocketWrapper wrapper) mutable { + [&receiver_socket, connected]( + const std::string& service_id, + std::shared_ptr wrapper) mutable { receiver_socket = wrapper; - connected.Set(receiver_socket.IsValid()); + connected.Set(receiver_socket->IsValid()); }, params.non_cellular); CancellationFlag flag(true); - ErrorOr sender_socket_result = sender.Connect( + ErrorOr> sender_socket_result = sender.Connect( service_id, self_id, location_hint, &flag, params.non_cellular); // If FeatureFlag is disabled, Cancelled is false as no-op. if (!params.feature_flags.enable_cancellation_flag) { EXPECT_TRUE(sender_socket_result.has_value()); - EXPECT_TRUE(sender_socket_result.value().IsValid()); + EXPECT_TRUE(sender_socket_result.value()->IsValid()); ExceptionOr devices_connected = connected.Get(); ASSERT_TRUE(devices_connected.ok()); EXPECT_TRUE(devices_connected.result()); - sender_socket_result.value().GetOutputStream().Write(message); + sender_socket_result.value()->GetOutputStream().Write(message); ExceptionOr received_msg = - receiver_socket.GetInputStream().Read(/*size=*/32); + receiver_socket->GetInputStream().Read(/*size=*/32); ASSERT_TRUE(received_msg.ok()); EXPECT_EQ(message, received_msg.result().AsStringView()); - receiver_socket.Close(); + receiver_socket->Close(); } else { EXPECT_TRUE(sender_socket_result.has_error()); } @@ -200,7 +202,7 @@ TEST_P(WebRtcTest, Connect_NoPeer) { ASSERT_TRUE(webrtc.IsAvailable()); CancellationFlag flag; - ErrorOr wrapper_1_result = webrtc.Connect( + ErrorOr> wrapper_1_result = webrtc.Connect( service_id, peer_id, location_hint, &flag, params.non_cellular); EXPECT_TRUE(wrapper_1_result.has_error()); @@ -225,7 +227,7 @@ TEST_P(WebRtcTest, StartAcceptingConnection_ThenConnect) { service_id, self_id, location_hint, mock_accepted_callback_.AsStdFunction(), params.non_cellular)); CancellationFlag flag; - ErrorOr wrapper_result = + ErrorOr> wrapper_result = webrtc.Connect(service_id, WebrtcPeerId("random_peer_id"), location_hint, &flag, params.non_cellular); EXPECT_TRUE(webrtc.IsAcceptingConnections(service_id)); @@ -262,7 +264,7 @@ TEST_P(WebRtcTest, StartAndStopAcceptingConnections) { TEST_P(WebRtcTest, ConnectTwice) { env_.Start({.webrtc_enabled = true}); WebRtc receiver, sender, device_c; - WebRtcSocketWrapper receiver_socket; + std::shared_ptr receiver_socket; WebRtcTestParams params = GetParam(); const WebrtcPeerId self_id("self_id"), other_id("other_id"); const std::string service_id("NearbySharing"); @@ -272,45 +274,47 @@ TEST_P(WebRtcTest, ConnectTwice) { receiver.StartAcceptingConnections( service_id, self_id, location_hint, - [&receiver_socket, connected](const std::string& service_id, - WebRtcSocketWrapper wrapper) mutable { + [&receiver_socket, connected]( + const std::string& service_id, + std::shared_ptr wrapper) mutable { receiver_socket = wrapper; - connected.Set(receiver_socket.IsValid()); + connected.Set(receiver_socket->IsValid()); }, params.non_cellular); device_c.StartAcceptingConnections( service_id, other_id, location_hint, - [](const std::string& service_id, WebRtcSocketWrapper wrapper) {}, + [](const std::string& service_id, std::shared_ptr wrapper) { + }, params.non_cellular); CancellationFlag flag; - ErrorOr sender_socket_result = sender.Connect( + ErrorOr> sender_socket_result = sender.Connect( service_id, self_id, location_hint, &flag, params.non_cellular); EXPECT_TRUE(sender_socket_result.has_value()); - EXPECT_TRUE(sender_socket_result.value().IsValid()); + EXPECT_TRUE(sender_socket_result.value()->IsValid()); ExceptionOr devices_connected = connected.Get(); ASSERT_TRUE(devices_connected.ok()); EXPECT_TRUE(devices_connected.result()); - ErrorOr socket_result = sender.Connect( + ErrorOr> socket_result = sender.Connect( service_id, other_id, location_hint, &flag, params.non_cellular); EXPECT_TRUE(socket_result.has_value()); - EXPECT_TRUE(socket_result.value().IsValid()); - socket_result.value().Close(); + EXPECT_TRUE(socket_result.value()->IsValid()); + socket_result.value()->Close(); - EXPECT_TRUE(receiver_socket.IsValid()); + EXPECT_TRUE(receiver_socket->IsValid()); EXPECT_TRUE(sender_socket_result.has_value()); - EXPECT_TRUE(sender_socket_result.value().IsValid()); + EXPECT_TRUE(sender_socket_result.value()->IsValid()); - sender_socket_result.value().GetOutputStream().Write(message); + sender_socket_result.value()->GetOutputStream().Write(message); ExceptionOr received_msg = - receiver_socket.GetInputStream().Read(/*size=*/32); + receiver_socket->GetInputStream().Read(/*size=*/32); ASSERT_TRUE(received_msg.ok()); EXPECT_EQ(message, received_msg.result().AsStringView()); - receiver_socket.Close(); + receiver_socket->Close(); env_.Stop(); } @@ -319,7 +323,7 @@ TEST_P(WebRtcTest, ConnectTwice) { TEST_P(WebRtcTest, ConnectBothDevicesAndAbort) { env_.Start({.webrtc_enabled = true}); WebRtc receiver, sender; - WebRtcSocketWrapper receiver_socket, sender_socket; + std::shared_ptr receiver_socket, sender_socket; WebRtcTestParams params = GetParam(); const WebrtcPeerId self_id("self_id"); const std::string service_id("NearbySharing"); @@ -328,24 +332,25 @@ TEST_P(WebRtcTest, ConnectBothDevicesAndAbort) { receiver.StartAcceptingConnections( service_id, self_id, location_hint, - [&receiver_socket, connected](const std::string& service_id, - WebRtcSocketWrapper wrapper) mutable { + [&receiver_socket, connected]( + const std::string& service_id, + std::shared_ptr wrapper) mutable { receiver_socket = wrapper; - connected.Set(receiver_socket.IsValid()); + connected.Set(receiver_socket->IsValid()); }, params.non_cellular); CancellationFlag flag; - ErrorOr sender_socket_result = sender.Connect( + ErrorOr> sender_socket_result = sender.Connect( service_id, self_id, location_hint, &flag, params.non_cellular); EXPECT_TRUE(sender_socket_result.has_value()); - EXPECT_TRUE(sender_socket_result.value().IsValid()); + EXPECT_TRUE(sender_socket_result.value()->IsValid()); ExceptionOr devices_connected = connected.Get(); ASSERT_TRUE(devices_connected.ok()); EXPECT_TRUE(devices_connected.result()); - receiver_socket.Close(); + receiver_socket->Close(); env_.Stop(); } @@ -354,7 +359,7 @@ TEST_P(WebRtcTest, ConnectBothDevicesAndAbort) { TEST_P(WebRtcTest, ConnectBothDevicesAndSendData) { env_.Start({.webrtc_enabled = true}); WebRtc receiver, sender; - WebRtcSocketWrapper receiver_socket; + std::shared_ptr receiver_socket; WebRtcTestParams params = GetParam(); const WebrtcPeerId self_id("self_id"); const std::string service_id("NearbySharing"); @@ -364,30 +369,31 @@ TEST_P(WebRtcTest, ConnectBothDevicesAndSendData) { receiver.StartAcceptingConnections( service_id, self_id, location_hint, - [&receiver_socket, connected](const std::string& service_id, - WebRtcSocketWrapper wrapper) mutable { + [&receiver_socket, connected]( + const std::string& service_id, + std::shared_ptr wrapper) mutable { receiver_socket = wrapper; - connected.Set(receiver_socket.IsValid()); + connected.Set(receiver_socket->IsValid()); }, params.non_cellular); CancellationFlag flag; - ErrorOr sender_socket_result = sender.Connect( + ErrorOr> sender_socket_result = sender.Connect( service_id, self_id, location_hint, &flag, params.non_cellular); EXPECT_TRUE(sender_socket_result.has_value()); - EXPECT_TRUE(sender_socket_result.value().IsValid()); + EXPECT_TRUE(sender_socket_result.value()->IsValid()); ExceptionOr devices_connected = connected.Get(); ASSERT_TRUE(devices_connected.ok()); EXPECT_TRUE(devices_connected.result()); - sender_socket_result.value().GetOutputStream().Write(message); + sender_socket_result.value()->GetOutputStream().Write(message); ExceptionOr received_msg = - receiver_socket.GetInputStream().Read(/*size=*/32); + receiver_socket->GetInputStream().Read(/*size=*/32); ASSERT_TRUE(received_msg.ok()); EXPECT_EQ(message, received_msg.result().AsStringView()); - receiver_socket.Close(); + receiver_socket->Close(); env_.Stop(); } @@ -405,7 +411,7 @@ TEST_P(WebRtcTest, Connect_NullPeerConnection) { ASSERT_TRUE(webrtc.IsAvailable()); CancellationFlag flag; - ErrorOr wrapper_result = + ErrorOr> wrapper_result = webrtc.Connect(service_id, WebrtcPeerId("random_peer_id"), location_hint, &flag, params.non_cellular); EXPECT_TRUE(wrapper_result.has_error()); @@ -456,7 +462,7 @@ TEST_P(WebRtcTest, CancelDuringConnect) { .enable_cancellation_flag = true, }); - WebRtcSocketWrapper receiver_socket, sender_socket; + std::shared_ptr receiver_socket, sender_socket; const WebrtcPeerId self_id("self_id"); const std::string service_id("NearbySharing"); LocationHint location_hint; @@ -479,14 +485,15 @@ TEST_P(WebRtcTest, CancelDuringConnect) { receiver->StartAcceptingConnections( service_id, self_id, location_hint, - [&receiver_socket, connected](const std::string& service_id, - WebRtcSocketWrapper wrapper) mutable { + [&receiver_socket, connected]( + const std::string& service_id, + std::shared_ptr wrapper) mutable { receiver_socket = wrapper; - connected.Set(receiver_socket.IsValid()); + connected.Set(receiver_socket->IsValid()); }, params.non_cellular); - ErrorOr sender_socket_result = sender->Connect( + ErrorOr> sender_socket_result = sender->Connect( service_id, self_id, location_hint, &sender_flag, params.non_cellular); // Since the flag was cancelled during the initial `AttemptToConnect`, except @@ -512,7 +519,7 @@ TEST_P(WebRtcTest, CancelBeforeConnect) { .enable_cancellation_flag = true, }); - WebRtcSocketWrapper receiver_socket; + std::shared_ptr receiver_socket; const WebrtcPeerId self_id("self_id"); const std::string service_id("NearbySharing"); LocationHint location_hint; @@ -528,14 +535,15 @@ TEST_P(WebRtcTest, CancelBeforeConnect) { receiver->StartAcceptingConnections( service_id, self_id, location_hint, - [&receiver_socket, connected](const std::string& service_id, - WebRtcSocketWrapper wrapper) mutable { + [&receiver_socket, connected]( + const std::string& service_id, + std::shared_ptr wrapper) mutable { receiver_socket = wrapper; - connected.Set(receiver_socket.IsValid()); + connected.Set(receiver_socket->IsValid()); }, params.non_cellular); - ErrorOr sender_socket_result = sender->Connect( + ErrorOr> sender_socket_result = sender->Connect( service_id, self_id, location_hint, &sender_flag, params.non_cellular); // Expect an invalid socket from stopping during the first attempt to connect, @@ -558,7 +566,7 @@ TEST_P(WebRtcTest, CancelDuringConnect_MultipleConnect) { .enable_cancellation_flag = true, }); - WebRtcSocketWrapper receiver_socket; + std::shared_ptr receiver_socket; const WebrtcPeerId self_id("self_id"); const std::string ns_service_id("NearbySharing"); const std::string ph_service_id("PhoneHub"); @@ -576,18 +584,19 @@ TEST_P(WebRtcTest, CancelDuringConnect_MultipleConnect) { receiver->StartAcceptingConnections( ns_service_id, self_id, location_hint, - [&receiver_socket, connected](const std::string& ns_service_id, - WebRtcSocketWrapper wrapper) mutable { + [&receiver_socket, connected]( + const std::string& ns_service_id, + std::shared_ptr wrapper) mutable { receiver_socket = wrapper; - connected.Set(receiver_socket.IsValid()); + connected.Set(receiver_socket->IsValid()); }, params.non_cellular); // Simulate a successful connect for the endpoint of NearbySharing. - ErrorOr sender_socket_result = sender->Connect( + ErrorOr> sender_socket_result = sender->Connect( ns_service_id, self_id, location_hint, &flag, params.non_cellular); EXPECT_TRUE(sender_socket_result.has_value()); - EXPECT_TRUE(sender_socket_result.value().IsValid()); + EXPECT_TRUE(sender_socket_result.value()->IsValid()); // Calls `CancellationFlag::Cancel` during a call to `GetSignalingMessenger` // to simulate the cancellation occuring during an `AttemptToConnect` for the diff --git a/connections/implementation/p2p_cluster_pcp_handler.h b/connections/implementation/p2p_cluster_pcp_handler.h index a3ac0891..7dac0be2 100644 --- a/connections/implementation/p2p_cluster_pcp_handler.h +++ b/connections/implementation/p2p_cluster_pcp_handler.h @@ -56,7 +56,6 @@ #include "internal/platform/nsd_service_info.h" #include "internal/platform/wifi_lan.h" #ifdef NO_WEBRTC -#include "connections/implementation/mediums/webrtc_socket_stub.h" #include "connections/implementation/mediums/webrtc_stub.h" #else #include "connections/implementation/mediums/webrtc.h" diff --git a/connections/implementation/webrtc_bwu_handler.cc b/connections/implementation/webrtc_bwu_handler.cc index 50a38814..5ec2c1c0 100644 --- a/connections/implementation/webrtc_bwu_handler.cc +++ b/connections/implementation/webrtc_bwu_handler.cc @@ -60,10 +60,10 @@ LocationHint BuildLocationHint(const std::string& location) { } // namespace WebrtcBwuHandler::WebrtcIncomingSocket::WebrtcIncomingSocket( - const std::string& name, mediums::WebRtcSocketWrapper socket) - : name_(name), socket_(socket) {} + const std::string& name, std::shared_ptr socket) + : name_(name), socket_(std::move(socket)) {} -void WebrtcBwuHandler::WebrtcIncomingSocket::Close() { socket_.Close(); } +void WebrtcBwuHandler::WebrtcIncomingSocket::Close() { socket_->Close(); } std::string WebrtcBwuHandler::WebrtcIncomingSocket::ToString() { return name_; } @@ -93,9 +93,10 @@ WebrtcBwuHandler::CreateUpgradedEndpointChannel( << peer_id.GetId() << ", location hint " << location_hint.location(); - ErrorOr socket_result = webrtc_.Connect( - service_id, peer_id, location_hint, - client->GetCancellationFlag(endpoint_id), client->GetWebRtcNonCellular()); + ErrorOr> socket_result = + webrtc_.Connect(service_id, peer_id, location_hint, + client->GetCancellationFlag(endpoint_id), + client->GetWebRtcNonCellular()); if (socket_result.has_error()) { LOG(ERROR) << "WebRtcBwuHandler failed to connect to remote peer (" << peer_id.GetId() << ") on endpoint " << endpoint_id @@ -111,7 +112,7 @@ WebrtcBwuHandler::CreateUpgradedEndpointChannel( auto channel = std::make_unique( service_id, /*channel_name=*/service_id, socket_result.value()); if (channel == nullptr) { - socket_result.value().Close(); + socket_result.value()->Close(); LOG(ERROR) << "WebRtcBwuHandler failed to create new EndpointChannel for " "outgoing socket, aborting upgrade."; return {Error( @@ -164,11 +165,11 @@ std::string WebrtcBwuHandler::HandleInitializeUpgradedMediumForEndpoint( // for this socket. void WebrtcBwuHandler::OnIncomingWebrtcConnection( ClientProxy* client, const std::string& upgrade_service_id, - mediums::WebRtcSocketWrapper socket) { + std::shared_ptr socket) { auto channel = std::make_unique( upgrade_service_id, /*channel_name=*/upgrade_service_id, socket); - auto webrtc_socket = - std::make_unique(upgrade_service_id, socket); + auto webrtc_socket = std::make_unique( + upgrade_service_id, std::move(socket)); std::unique_ptr connection( new IncomingSocketConnection{std::move(webrtc_socket), std::move(channel)}); diff --git a/connections/implementation/webrtc_bwu_handler.h b/connections/implementation/webrtc_bwu_handler.h index 5220ef18..c4918ad2 100644 --- a/connections/implementation/webrtc_bwu_handler.h +++ b/connections/implementation/webrtc_bwu_handler.h @@ -44,15 +44,15 @@ class WebrtcBwuHandler : public BaseBwuHandler { private: class WebrtcIncomingSocket : public BwuHandler::IncomingSocket { public: - explicit WebrtcIncomingSocket(const std::string& name, - mediums::WebRtcSocketWrapper socket); + explicit WebrtcIncomingSocket( + const std::string& name, std::shared_ptr socket); std::string ToString() override; void Close() override; private: std::string name_; - mediums::WebRtcSocketWrapper socket_; + std::shared_ptr socket_; }; // BwuHandler implementation: @@ -74,9 +74,9 @@ class WebrtcBwuHandler : public BaseBwuHandler { void HandleRevertInitiatorStateForService( const std::string& upgrade_service_id) final; - void OnIncomingWebrtcConnection(ClientProxy* client, - const std::string& upgrade_service_id, - mediums::WebRtcSocketWrapper socket); + void OnIncomingWebrtcConnection( + ClientProxy* client, const std::string& upgrade_service_id, + std::shared_ptr socket); Mediums& mediums_; mediums::WebRtc& webrtc_{mediums_.GetWebRtc()}; diff --git a/connections/implementation/webrtc_bwu_handler_stub.cc b/connections/implementation/webrtc_bwu_handler_stub.cc index 619b7647..d116d100 100644 --- a/connections/implementation/webrtc_bwu_handler_stub.cc +++ b/connections/implementation/webrtc_bwu_handler_stub.cc @@ -36,8 +36,8 @@ using ::location::nearby::proto::connections::OperationResultCode; } // namespace WebrtcBwuHandler::WebrtcIncomingSocket::WebrtcIncomingSocket( - const std::string& name, mediums::WebRtcSocketWrapper socket) - : name_(name), socket_(socket) {} + const std::string& name, std::shared_ptr socket) + : name_(name), socket_(std::move(socket)) {} void WebrtcBwuHandler::WebrtcIncomingSocket::Close() {} @@ -76,7 +76,7 @@ std::string WebrtcBwuHandler::HandleInitializeUpgradedMediumForEndpoint( // for this socket. void WebrtcBwuHandler::OnIncomingWebrtcConnection( ClientProxy* client, const std::string& upgrade_service_id, - mediums::WebRtcSocketWrapper socket) {} + std::shared_ptr socket) {} } // namespace connections } // namespace nearby diff --git a/connections/implementation/webrtc_bwu_handler_stub.h b/connections/implementation/webrtc_bwu_handler_stub.h index 6ee6635f..7c959cd9 100644 --- a/connections/implementation/webrtc_bwu_handler_stub.h +++ b/connections/implementation/webrtc_bwu_handler_stub.h @@ -23,11 +23,7 @@ #include "connections/implementation/client_proxy.h" #include "connections/implementation/endpoint_channel_manager.h" #include "connections/implementation/mediums/mediums.h" -#ifdef NO_WEBRTC -#include "connections/implementation/mediums/webrtc_socket_stub.h" -#else #include "connections/implementation/mediums/webrtc_socket.h" -#endif #include "internal/platform/expected.h" namespace nearby { @@ -44,15 +40,15 @@ class WebrtcBwuHandler : public BaseBwuHandler { private: class WebrtcIncomingSocket : public BwuHandler::IncomingSocket { public: - explicit WebrtcIncomingSocket(const std::string& name, - mediums::WebRtcSocketWrapper socket); + explicit WebrtcIncomingSocket( + const std::string& name, std::shared_ptr socket); std::string ToString() override; void Close() override; private: std::string name_; - mediums::WebRtcSocketWrapper socket_; + std::shared_ptr socket_; }; // BwuHandler implementation: @@ -74,9 +70,9 @@ class WebrtcBwuHandler : public BaseBwuHandler { void HandleRevertInitiatorStateForService( const std::string& upgrade_service_id) final; - void OnIncomingWebrtcConnection(ClientProxy* client, - const std::string& upgrade_service_id, - mediums::WebRtcSocketWrapper socket); + void OnIncomingWebrtcConnection( + ClientProxy* client, const std::string& upgrade_service_id, + std::shared_ptr socket); Mediums& mediums_; mediums::WebRtc& webrtc_{mediums_.GetWebRtc()}; diff --git a/connections/implementation/webrtc_endpoint_channel.cc b/connections/implementation/webrtc_endpoint_channel.cc index b22501bf..a95c4027 100644 --- a/connections/implementation/webrtc_endpoint_channel.cc +++ b/connections/implementation/webrtc_endpoint_channel.cc @@ -14,16 +14,21 @@ #include "connections/implementation/webrtc_endpoint_channel.h" +#include #include +#include + +#include "connections/implementation/base_endpoint_channel.h" +#include "connections/implementation/mediums/webrtc_socket.h" namespace nearby { namespace connections { WebRtcEndpointChannel::WebRtcEndpointChannel( const std::string& service_id, const std::string& channel_name, - mediums::WebRtcSocketWrapper socket) - : BaseEndpointChannel(service_id, channel_name, &socket.GetInputStream(), - &socket.GetOutputStream()), + std::shared_ptr socket) + : BaseEndpointChannel(service_id, channel_name, &socket->GetInputStream(), + &socket->GetOutputStream()), webrtc_socket_(std::move(socket)) {} location::nearby::proto::connections::Medium WebRtcEndpointChannel::GetMedium() @@ -31,7 +36,7 @@ location::nearby::proto::connections::Medium WebRtcEndpointChannel::GetMedium() return location::nearby::proto::connections::Medium::WEB_RTC; } -void WebRtcEndpointChannel::CloseImpl() { webrtc_socket_.Close(); } +void WebRtcEndpointChannel::CloseImpl() { webrtc_socket_->Close(); } } // namespace connections } // namespace nearby diff --git a/connections/implementation/webrtc_endpoint_channel.h b/connections/implementation/webrtc_endpoint_channel.h index ae0eb551..dd11bdf6 100644 --- a/connections/implementation/webrtc_endpoint_channel.h +++ b/connections/implementation/webrtc_endpoint_channel.h @@ -15,14 +15,11 @@ #ifndef CORE_INTERNAL_WEBRTC_ENDPOINT_CHANNEL_H_ #define CORE_INTERNAL_WEBRTC_ENDPOINT_CHANNEL_H_ +#include #include #include "connections/implementation/base_endpoint_channel.h" -#ifdef NO_WEBRTC -#include "connections/implementation/mediums/webrtc_socket_stub.h" -#else #include "connections/implementation/mediums/webrtc_socket.h" -#endif namespace nearby { namespace connections { @@ -31,14 +28,14 @@ class WebRtcEndpointChannel final : public BaseEndpointChannel { public: WebRtcEndpointChannel(const std::string& service_id, const std::string& channel_name, - mediums::WebRtcSocketWrapper webrtc_socket); + std::shared_ptr socket); location::nearby::proto::connections::Medium GetMedium() const override; private: void CloseImpl() override; - mediums::WebRtcSocketWrapper webrtc_socket_; + std::shared_ptr webrtc_socket_; }; } // namespace connections