From 999cdd99034676cc7f02799d5dc2f16e8d4b9712 Mon Sep 17 00:00:00 2001 From: Alexey Polyudov Date: Tue, 7 Jul 2020 12:34:48 -0700 Subject: [PATCH] Roll forward to cl/320013226 Signed-off-by: Alexey Polyudov Change-Id: I8be5378519ca952da43c40fba96141a7f8517748 --- cpp/core_v2/internal/mediums/webrtc.cc | 92 ++++-- cpp/core_v2/internal/mediums/webrtc.h | 15 +- cpp/core_v2/internal/mediums/webrtc_test.cc | 129 +++++++- cpp/core_v2/internal/mediums/wifi_lan.cc | 33 +- cpp/core_v2/internal/mediums/wifi_lan.h | 6 +- cpp/core_v2/internal/mediums/wifi_lan_test.cc | 103 ++++++- .../internal/offline_service_controller.cc | 14 - .../internal/offline_service_controller.h | 14 - .../offline_service_controller_test.cc | 14 - .../internal/offline_simulation_user.cc | 14 - .../internal/offline_simulation_user.h | 14 - .../internal/p2p_cluster_pcp_handler.cc | 2 +- cpp/platform_v2/base/medium_environment.cc | 116 +++++-- cpp/platform_v2/base/medium_environment.h | 41 ++- cpp/platform_v2/impl/g3/wifi_lan.cc | 283 ++++++++++++++++-- cpp/platform_v2/impl/g3/wifi_lan.h | 133 +++++++- cpp/platform_v2/public/BUILD | 1 + cpp/platform_v2/public/cancelable.h | 4 +- cpp/platform_v2/public/cancelable_alarm.h | 5 + .../public/cancelable_alarm_test.cc | 54 ++++ cpp/platform_v2/public/wifi_lan.cc | 6 +- cpp/platform_v2/public/wifi_lan.h | 4 +- cpp/platform_v2/public/wifi_lan_test.cc | 124 ++++++-- proto/error_code_enums.proto | 32 +- 24 files changed, 1026 insertions(+), 227 deletions(-) create mode 100644 cpp/platform_v2/public/cancelable_alarm_test.cc diff --git a/cpp/core_v2/internal/mediums/webrtc.cc b/cpp/core_v2/internal/mediums/webrtc.cc index 32a4ec0e..4b9510c7 100644 --- a/cpp/core_v2/internal/mediums/webrtc.cc +++ b/cpp/core_v2/internal/mediums/webrtc.cc @@ -7,11 +7,13 @@ #include "core_v2/internal/mediums/webrtc/signaling_frames.h" #include "platform_v2/base/byte_array.h" #include "platform_v2/base/listeners.h" +#include "platform_v2/public/cancelable_alarm.h" #include "platform_v2/public/future.h" #include "platform_v2/public/logging.h" #include "platform_v2/public/mutex_lock.h" #include "location/nearby/mediums/proto/web_rtc_signaling_frames.pb.h" #include "absl/strings/str_cat.h" +#include "absl/time/time.h" #include "webrtc/api/jsep.h" namespace location { @@ -22,19 +24,22 @@ namespace mediums { namespace { // The maximum amount of time to wait to connect to a data channel via WebRTC. -// TODO(himanshujaju): Should this be configurable per platform? constexpr absl::Duration kDataChannelTimeout = absl::Milliseconds(5000); +// Delay between restarting signaling messenger to receive messages. +constexpr absl::Duration kRestartReceiveMessagesDuration = absl::Seconds(60); + } // namespace WebRtc::WebRtc() = default; WebRtc::~WebRtc() { + // This ensures that all pending callbacks are run before we reset the medium + // and we are not accepting new runnables. + restart_receive_messages_executor_.Shutdown(); single_thread_executor_.Shutdown(); - { - MutexLock lock(&mutex_); - Disconnect(); - } + + Disconnect(); } bool WebRtc::IsAvailable() { return medium_.IsValid(); } @@ -70,6 +75,11 @@ bool WebRtc::StartAcceptingConnections(const PeerId& self_id, if (!InitWebRtcFlow(Role::kOfferer, self_id)) return false; + restart_receive_messages_alarm_ = CancelableAlarm( + "restart_receiving_messages_webrtc", + std::bind(&WebRtc::RestartReceiveMessages, this), + kRestartReceiveMessagesDuration, &restart_receive_messages_executor_); + SessionDescriptionWrapper offer = connection_flow_->CreateOffer(); pending_local_offer_ = webrtc_frames::EncodeOffer(self_id, offer.GetSdp()); if (!SetLocalSessionDescription(std::move(offer))) { @@ -89,23 +99,25 @@ bool WebRtc::StartAcceptingConnections(const PeerId& self_id, } WebRtcSocketWrapper WebRtc::Connect(const PeerId& peer_id) { - MutexLock lock(&mutex_); - if (!IsAvailable()) { Disconnect(); return WebRtcSocketWrapper(); } - if (role_ != Role::kNone) { - NEARBY_LOG(WARNING, - "Cannot connect with WebRtc because we are already acting as %d", - role_); - return WebRtcSocketWrapper(); - } + { + MutexLock lock(&mutex_); + if (role_ != Role::kNone) { + NEARBY_LOG( + WARNING, + "Cannot connect with WebRtc because we are already acting as %d", + role_); + return WebRtcSocketWrapper(); + } - peer_id_ = peer_id; - if (!InitWebRtcFlow(Role::kAnswerer, PeerId::FromRandom())) { - return WebRtcSocketWrapper(); + peer_id_ = peer_id; + if (!InitWebRtcFlow(Role::kAnswerer, PeerId::FromRandom())) { + return WebRtcSocketWrapper(); + } } NEARBY_LOG(INFO, "Attempting to make a WebRTC connection to %s.", @@ -116,6 +128,9 @@ WebRtcSocketWrapper WebRtc::Connect(const PeerId& peer_id) { // The two devices have discovered each other, hence we have a timeout for // establishing the transport channel. + // NOTE - We should not hold |mutex_| while waiting for the data channel since + // it would block incoming signaling messages from being processed, resulting + // in a timeout in creating the socket. ExceptionOr result = socket_future.Get(kDataChannelTimeout); if (result.ok()) return result.result(); @@ -186,7 +201,8 @@ WebRtcSocketWrapper WebRtc::CreateWebRtcSocketWrapper( } auto socket = std::make_unique("WebRtcSocket", data_channel); - socket->SetOnSocketClosedListener({std::bind(&WebRtc::Disconnect, this)}); + socket->SetOnSocketClosedListener( + {[this]() { OffloadFromSignalingThread([this]() { Disconnect(); }); }}); return WebRtcSocketWrapper(std::move(socket)); } @@ -217,7 +233,7 @@ bool WebRtc::InitWebRtcFlow(Role role, const PeerId& self_id) { if (!signaling_messenger_->IsValid() || !signaling_messenger_->StartReceivingMessages( signaling_message_callback)) { - Disconnect(); + DisconnectLocked(); return false; } @@ -393,7 +409,7 @@ void WebRtc::SendAnswerToPeer() { void WebRtc::LogAndDisconnect(const std::string& error_message) { NEARBY_LOG(WARNING, "Disconnecting WebRTC : %s", error_message.c_str()); - Disconnect(); + DisconnectLocked(); } void WebRtc::LogAndShutdownSignaling(const std::string& error_message) { @@ -408,6 +424,11 @@ void WebRtc::ShutdownSignaling() { pending_local_offer_ = ByteArray(); pending_local_ice_candidates_.clear(); + if (restart_receive_messages_alarm_.IsValid()) { + restart_receive_messages_alarm_.Cancel(); + restart_receive_messages_alarm_ = CancelableAlarm(); + } + if (signaling_messenger_) { signaling_messenger_->StopReceivingMessages(); signaling_messenger_.reset(); @@ -417,6 +438,11 @@ void WebRtc::ShutdownSignaling() { } void WebRtc::Disconnect() { + MutexLock lock(&mutex_); + DisconnectLocked(); +} + +void WebRtc::DisconnectLocked() { ShutdownSignaling(); ShutdownWebRtcSocket(); ShutdownIceCandidateCollection(); @@ -440,6 +466,34 @@ void WebRtc::OffloadFromSignalingThread(Runnable runnable) { single_thread_executor_.Execute(std::move(runnable)); } +void WebRtc::RestartReceiveMessages() { + if (!IsAcceptingConnections()) { + NEARBY_LOG(INFO, + "Skipping restart since we are not accepting connections."); + return; + } + + NEARBY_LOG(INFO, "Restarting listening for receiving signaling messages."); + { + MutexLock lock(&mutex_); + signaling_messenger_->StopReceivingMessages(); + + signaling_messenger_ = medium_.GetSignalingMessenger(self_id_.GetId()); + + auto signaling_message_callback = [this](ByteArray message) { + OffloadFromSignalingThread([this, message{std::move(message)}]() { + ProcessSignalingMessage(message); + }); + }; + + if (!signaling_messenger_->IsValid() || + !signaling_messenger_->StartReceivingMessages( + signaling_message_callback)) { + DisconnectLocked(); + } + } +} + } // namespace mediums } // namespace connections } // namespace nearby diff --git a/cpp/core_v2/internal/mediums/webrtc.h b/cpp/core_v2/internal/mediums/webrtc.h index 27612269..1322b5bb 100644 --- a/cpp/core_v2/internal/mediums/webrtc.h +++ b/cpp/core_v2/internal/mediums/webrtc.h @@ -10,9 +10,12 @@ #include "core_v2/internal/mediums/webrtc/peer_id.h" #include "core_v2/internal/mediums/webrtc/webrtc_socket.h" #include "core_v2/internal/mediums/webrtc/webrtc_socket_wrapper.h" +#include "platform_v2/public/cancelable_alarm.h" +#include "platform_v2/public/scheduled_executor.h" #include "platform_v2/base/byte_array.h" #include "platform_v2/base/listeners.h" #include "platform_v2/base/runnable.h" +#include "platform_v2/public/atomic_boolean.h" #include "platform_v2/public/future.h" #include "platform_v2/public/mutex.h" #include "platform_v2/public/single_thread_executor.h" @@ -112,8 +115,11 @@ class WebRtc { void LogAndDisconnect(const std::string& error_message) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + // Runs on @MainThread. + void Disconnect() ABSL_LOCKS_EXCLUDED(mutex_); + // Runs on @MainThread and |single_thread_executor_|. - void Disconnect() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + void DisconnectLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); void LogAndShutdownSignaling(const std::string& error_message) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); @@ -129,6 +135,9 @@ class WebRtc { void OffloadFromSignalingThread(Runnable runnable); + // Runs on |restart_receive_messages_executor_|. + void RestartReceiveMessages() ABSL_LOCKS_EXCLUDED(mutex_); + Mutex mutex_; Role role_ ABSL_GUARDED_BY(mutex_) = Role::kNone; @@ -145,6 +154,10 @@ class WebRtc { WebRtcSocketWrapper socket_ ABSL_GUARDED_BY(mutex_); SingleThreadExecutor single_thread_executor_; + + // Restarts the signaling messenger for receiving messages. + ScheduledExecutor restart_receive_messages_executor_; + CancelableAlarm restart_receive_messages_alarm_; }; } // namespace mediums diff --git a/cpp/core_v2/internal/mediums/webrtc_test.cc b/cpp/core_v2/internal/mediums/webrtc_test.cc index 140571f4..9b4f8399 100644 --- a/cpp/core_v2/internal/mediums/webrtc_test.cc +++ b/cpp/core_v2/internal/mediums/webrtc_test.cc @@ -89,28 +89,139 @@ TEST(WebRtcTest, StartAndStopAcceptingConnections) { EXPECT_FALSE(webrtc.IsAcceptingConnections()); } -// Tests the flow when the device calls StartAcceptingConnections() after -// calling Connect() without disconnecting in between. -TEST(WebRtcTest, Connect_ThenStartAcceptingConnections) { - // TODO(himanshujaju) - Complete the test. -} - // Tests the flow when the device tries to connect to two different peers // without disconnecting in between. TEST(WebRtcTest, ConnectTwice) { - // TODO(himanshujaju) - Complete the test. + WebRtc receiver, sender, device_c; + WebRtcSocketWrapper receiver_socket, sender_socket; + const PeerId self_id("self_id"), other_id("other_id"); + Future connected; + ByteArray message("message xyz"); + + receiver.StartAcceptingConnections( + self_id, + {[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable { + receiver_socket = wrapper; + connected.Set(receiver_socket.IsValid()); + }}); + + using MockAcceptedCallback = + testing::MockFunction; + testing::StrictMock mock_accepted_callback_; + device_c.StartAcceptingConnections(other_id, + {mock_accepted_callback_.AsStdFunction()}); + + sender_socket = sender.Connect(self_id); + EXPECT_TRUE(sender_socket.IsValid()); + + ExceptionOr devices_connected = connected.Get(); + ASSERT_TRUE(devices_connected.ok()); + EXPECT_TRUE(devices_connected.result()); + + WebRtcSocketWrapper socket = sender.Connect(other_id); + EXPECT_FALSE(socket.IsValid()); + + EXPECT_TRUE(receiver_socket.IsValid()); + EXPECT_TRUE(sender_socket.IsValid()); + + sender_socket.GetOutputStream().Write(message); + ExceptionOr received_msg = + receiver_socket.GetInputStream().Read(/*size=*/32); + ASSERT_TRUE(received_msg.ok()); + EXPECT_EQ(message, received_msg.result()); + + receiver_socket.Close(); } // Tests the flow when the two devices exchange SDP messages and connect to each // other but disconnect before being able to send/receive the actual data. TEST(WebRtcTest, ConnectBothDevicesAndAbort) { - // TODO(himanshujaju) - Complete the test. + WebRtc receiver, sender; + WebRtcSocketWrapper receiver_socket, sender_socket; + const PeerId self_id("self_id"); + Future connected; + ByteArray message("message xyz"); + + receiver.StartAcceptingConnections( + self_id, + {[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable { + receiver_socket = wrapper; + connected.Set(receiver_socket.IsValid()); + }}); + + sender_socket = sender.Connect(self_id); + EXPECT_TRUE(sender_socket.IsValid()); + + ExceptionOr devices_connected = connected.Get(); + ASSERT_TRUE(devices_connected.ok()); + EXPECT_TRUE(devices_connected.result()); + + receiver_socket.Close(); } // Tests the flow when the two devices exchange SDP messages and connect to each // other and the actual data is exchanged successfully between the devices. TEST(WebRtcTest, ConnectBothDevicesAndSendData) { - // TODO(himanshujaju) - Complete the test. + WebRtc receiver, sender; + WebRtcSocketWrapper receiver_socket, sender_socket; + const PeerId self_id("self_id"); + Future connected; + ByteArray message("message"); + + receiver.StartAcceptingConnections( + self_id, + {[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable { + receiver_socket = wrapper; + connected.Set(receiver_socket.IsValid()); + }}); + + sender_socket = sender.Connect(self_id); + EXPECT_TRUE(sender_socket.IsValid()); + + ExceptionOr devices_connected = connected.Get(); + ASSERT_TRUE(devices_connected.ok()); + EXPECT_TRUE(devices_connected.result()); + + sender_socket.GetOutputStream().Write(message); + ExceptionOr received_msg = + receiver_socket.GetInputStream().Read(/*size=*/32); + ASSERT_TRUE(received_msg.ok()); + EXPECT_EQ(message, received_msg.result()); + + receiver_socket.Close(); +} + +// Tests the flow when the two devices exchange SDP messages and connect to each +// other but the signaling channel is closed before sending the data. +TEST(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) { + WebRtc receiver, sender; + WebRtcSocketWrapper receiver_socket, sender_socket; + const PeerId self_id("self_id"); + Future connected; + ByteArray message("message xyz"); + + receiver.StartAcceptingConnections( + self_id, + {[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable { + receiver_socket = wrapper; + connected.Set(receiver_socket.IsValid()); + }}); + + sender_socket = sender.Connect(self_id); + EXPECT_TRUE(sender_socket.IsValid()); + + ExceptionOr devices_connected = connected.Get(); + ASSERT_TRUE(devices_connected.ok()); + EXPECT_TRUE(devices_connected.result()); + + // Only shuts down signaling channel. + receiver.StopAcceptingConnections(); + + sender_socket.GetOutputStream().Write(message); + ExceptionOr received_msg = + receiver_socket.GetInputStream().Read(/*size=*/32); + ASSERT_TRUE(received_msg.ok()); + EXPECT_EQ(message, received_msg.result()); } } // namespace diff --git a/cpp/core_v2/internal/mediums/wifi_lan.cc b/cpp/core_v2/internal/mediums/wifi_lan.cc index 894c4b9c..1983137f 100644 --- a/cpp/core_v2/internal/mediums/wifi_lan.cc +++ b/cpp/core_v2/internal/mediums/wifi_lan.cc @@ -43,24 +43,28 @@ bool WifiLan::StartAdvertising(const std::string& service_id, return false; } - NEARBY_LOG(INFO, "Turned on WifiLan advertising with service info name=%s", - wifi_lan_service_info_name.c_str()); + NEARBY_LOGS(INFO) << "Turned on WifiLan advertising with service info name=" + << wifi_lan_service_info_name + << ", service id=" << service_id; advertising_info_.service_id = service_id; return true; } -void WifiLan::StopAdvertising(const std::string& service_id) { +bool WifiLan::StopAdvertising(const std::string& service_id) { MutexLock lock(&mutex_); if (!IsAdvertisingLocked()) { NEARBY_LOG(INFO, "Can't turn off WifiLan advertising; it is already off"); - return; + return false; } - medium_.StopAdvertising(advertising_info_.service_id); + NEARBY_LOG(INFO, "Turned off WifiLan advertising with service id=%s", + service_id.c_str()); + bool ret = medium_.StopAdvertising(advertising_info_.service_id); // Reset our bundle of advertising state to mark that we're no longer // advertising. advertising_info_.Clear(); + return ret; } bool WifiLan::IsAdvertising() { @@ -103,23 +107,28 @@ bool WifiLan::StartDiscovery(const std::string& service_id, return false; } + NEARBY_LOG(INFO, "Turned on WifiLan discovering with service id=%s", + service_id.c_str()); // Mark the fact that we're currently performing a WifiLan discovering. discovering_info_.service_id = service_id; return true; } -void WifiLan::StopDiscovery(const std::string& service_id) { +bool WifiLan::StopDiscovery(const std::string& service_id) { MutexLock lock(&mutex_); if (!IsDiscoveringLocked(service_id)) { NEARBY_LOG(INFO, "Can't turn off WifiLan discovering because we never started " "discovering."); - return; + return false; } - medium_.StopDiscovery(service_id); + NEARBY_LOG(INFO, "Turned off WifiLan discovering with service id=%s", + service_id.c_str()); + bool ret = medium_.StopDiscovery(service_id); discovering_info_.Clear(); + return ret; } bool WifiLan::IsDiscovering(const std::string& service_id) { @@ -169,20 +178,22 @@ bool WifiLan::StartAcceptingConnections(const std::string& service_id, return true; } -void WifiLan::StopAcceptingConnections(const std::string& service_id) { +bool WifiLan::StopAcceptingConnections(const std::string& service_id) { MutexLock lock(&mutex_); if (!IsAcceptingConnectionsLocked(service_id)) { NEARBY_LOG(INFO, "Can't stop accepting WifiLan connections because it was never " "started."); - return; + return false; } - medium_.StopAcceptingConnections(accepting_connections_info_.service_id); + bool ret = + medium_.StopAcceptingConnections(accepting_connections_info_.service_id); // Reset our bundle of accepting connections state to mark that we're no // longer accepting connections. accepting_connections_info_.Clear(); + return ret; } bool WifiLan::IsAcceptingConnections(const std::string& service_id) { diff --git a/cpp/core_v2/internal/mediums/wifi_lan.h b/cpp/core_v2/internal/mediums/wifi_lan.h index 196cc2cd..16884a5d 100644 --- a/cpp/core_v2/internal/mediums/wifi_lan.h +++ b/cpp/core_v2/internal/mediums/wifi_lan.h @@ -30,7 +30,7 @@ class WifiLan { // Disables WifiLan advertising, and restores service info name to // what they were before the call to StartAdvertising(). - void StopAdvertising(const std::string& service_id) + bool StopAdvertising(const std::string& service_id) ABSL_LOCKS_EXCLUDED(mutex_); bool IsAdvertising() ABSL_LOCKS_EXCLUDED(mutex_); @@ -43,7 +43,7 @@ class WifiLan { ABSL_LOCKS_EXCLUDED(mutex_); // Disables WifiLan discovery mode. - void StopDiscovery(const std::string& service_id) ABSL_LOCKS_EXCLUDED(mutex_); + bool StopDiscovery(const std::string& service_id) ABSL_LOCKS_EXCLUDED(mutex_); bool IsDiscovering(const std::string& service_id) ABSL_LOCKS_EXCLUDED(mutex_); @@ -54,7 +54,7 @@ class WifiLan { ABSL_LOCKS_EXCLUDED(mutex_); // Closes socket corresponding to a service id. - void StopAcceptingConnections(const std::string& service_id) + bool StopAcceptingConnections(const std::string& service_id) ABSL_LOCKS_EXCLUDED(mutex_); bool IsAcceptingConnections(const std::string& service_id) diff --git a/cpp/core_v2/internal/mediums/wifi_lan_test.cc b/cpp/core_v2/internal/mediums/wifi_lan_test.cc index 545d6c3b..24e64d02 100644 --- a/cpp/core_v2/internal/mediums/wifi_lan_test.cc +++ b/cpp/core_v2/internal/mediums/wifi_lan_test.cc @@ -3,6 +3,8 @@ #include #include "platform_v2/base/medium_environment.h" +#include "platform_v2/public/count_down_latch.h" +#include "platform_v2/public/logging.h" #include "platform_v2/public/wifi_lan.h" #include "gmock/gmock.h" #include "gtest/gtest.h" @@ -12,11 +14,11 @@ namespace nearby { namespace connections { namespace { +constexpr absl::Duration kWaitDuration = absl::Milliseconds(1000); constexpr absl::string_view kServiceID{"com.google.location.nearby.apps.test"}; constexpr absl::string_view kServiceInfoName{ "Simulated WifiLan service encrypted string #1"}; -// TODO(edwinwu): Continue writing more tests after medium_environment is done. class WifiLanTest : public ::testing::Test { protected: using DiscoveredServiceCallback = WifiLanMedium::DiscoveredServiceCallback; @@ -30,6 +32,8 @@ TEST_F(WifiLanTest, CanConstructValidObject) { env_.Start(); WifiLan wifi_lan_a; WifiLan wifi_lan_b; + std::string service_id(kServiceID); + std::string service_name{kServiceInfoName}; EXPECT_TRUE(wifi_lan_a.IsAvailable()); EXPECT_TRUE(wifi_lan_b.IsAvailable()); @@ -38,9 +42,100 @@ TEST_F(WifiLanTest, CanConstructValidObject) { TEST_F(WifiLanTest, CanStartAdvertising) { env_.Start(); - WifiLan wifi_lan; - EXPECT_TRUE(wifi_lan.StartAdvertising(std::string(kServiceID), - std::string(kServiceInfoName))); + WifiLan wifi_lan_a; + WifiLan wifi_lan_b; + std::string service_id(kServiceID); + std::string service_name{kServiceInfoName}; + CountDownLatch found_latch(1); + + wifi_lan_b.StartDiscovery( + service_id, DiscoveredServiceCallback{ + .service_discovered_cb = + [&found_latch](WifiLanService& service, + const std::string& service_id) { + found_latch.CountDown(); + }, + }); + + EXPECT_TRUE(wifi_lan_a.StartAdvertising(service_id, service_name)); + EXPECT_TRUE(found_latch.Await(kWaitDuration).result()); + EXPECT_TRUE(wifi_lan_a.StopAdvertising(service_id)); + EXPECT_TRUE(wifi_lan_b.StopDiscovery(service_id)); + env_.Stop(); +} + +TEST_F(WifiLanTest, CanStartDiscovery) { + env_.Start(); + WifiLan wifi_lan_a; + WifiLan wifi_lan_b; + std::string service_id(kServiceID); + std::string service_name{kServiceInfoName}; + CountDownLatch accept_latch(1); + CountDownLatch lost_latch(1); + + wifi_lan_b.StartAdvertising(service_id, service_name); + + EXPECT_TRUE(wifi_lan_a.StartDiscovery( + service_id, { + .service_discovered_cb = + [&accept_latch](WifiLanService& service, + const std::string& service_id) { + accept_latch.CountDown(); + }, + .service_lost_cb = + [&lost_latch](WifiLanService& service, + const std::string& service_id) { + lost_latch.CountDown(); + }, + })); + EXPECT_TRUE(accept_latch.Await(kWaitDuration).result()); + wifi_lan_b.StopAdvertising(service_id); + EXPECT_TRUE(lost_latch.Await(kWaitDuration).result()); + EXPECT_TRUE(wifi_lan_a.StopDiscovery(service_id)); + env_.Stop(); +} + +TEST_F(WifiLanTest, CanStartAcceptingConnectionsAndConnect) { + env_.Start(); + WifiLan wifi_lan_a; + WifiLan wifi_lan_b; + std::string service_id(kServiceID); + std::string service_name{kServiceInfoName}; + CountDownLatch found_latch(1); + CountDownLatch accept_latch(1); + + wifi_lan_a.StartAdvertising(service_id, service_name); + wifi_lan_a.StartAcceptingConnections( + service_id, + { + .accepted_cb = [&accept_latch]( + WifiLanSocket socket, + const std::string&) { accept_latch.CountDown(); }, + }); + WifiLanService discovered_service; + wifi_lan_b.StartDiscovery( + service_id, + { + .service_discovered_cb = + [&found_latch, &discovered_service]( + WifiLanService& service, const std::string& service_id) { + discovered_service = service; + NEARBY_LOG(INFO, "Discovered service=%p [impl=%p]", &service, + &service.GetImpl()); + found_latch.CountDown(); + }, + }); + + EXPECT_TRUE(found_latch.Await(kWaitDuration).result()); + ASSERT_TRUE(discovered_service.IsValid()); + + WifiLanSocket socket = + wifi_lan_b.Connect(discovered_service, service_id); + + EXPECT_TRUE(accept_latch.Await(kWaitDuration).result()); + EXPECT_TRUE(socket.IsValid()); + wifi_lan_b.StopDiscovery(service_id); + wifi_lan_a.StopAdvertising(service_id); env_.Stop(); } diff --git a/cpp/core_v2/internal/offline_service_controller.cc b/cpp/core_v2/internal/offline_service_controller.cc index 7465fc96..249c97b8 100644 --- a/cpp/core_v2/internal/offline_service_controller.cc +++ b/cpp/core_v2/internal/offline_service_controller.cc @@ -1,17 +1,3 @@ -// 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. - #include "core_v2/internal/offline_service_controller.h" #include diff --git a/cpp/core_v2/internal/offline_service_controller.h b/cpp/core_v2/internal/offline_service_controller.h index a4855db2..bcb6e2c7 100644 --- a/cpp/core_v2/internal/offline_service_controller.h +++ b/cpp/core_v2/internal/offline_service_controller.h @@ -1,17 +1,3 @@ -// 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_V2_INTERNAL_OFFLINE_SERVICE_CONTROLLER_H_ #define CORE_V2_INTERNAL_OFFLINE_SERVICE_CONTROLLER_H_ diff --git a/cpp/core_v2/internal/offline_service_controller_test.cc b/cpp/core_v2/internal/offline_service_controller_test.cc index 2d4487ea..260dd527 100644 --- a/cpp/core_v2/internal/offline_service_controller_test.cc +++ b/cpp/core_v2/internal/offline_service_controller_test.cc @@ -1,17 +1,3 @@ -// 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. - #include "core_v2/internal/offline_service_controller.h" #include "core_v2/internal/offline_simulation_user.h" diff --git a/cpp/core_v2/internal/offline_simulation_user.cc b/cpp/core_v2/internal/offline_simulation_user.cc index 1a58f117..6ed65174 100644 --- a/cpp/core_v2/internal/offline_simulation_user.cc +++ b/cpp/core_v2/internal/offline_simulation_user.cc @@ -1,17 +1,3 @@ -// 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. - #include "core_v2/internal/offline_simulation_user.h" #include "core_v2/listeners.h" diff --git a/cpp/core_v2/internal/offline_simulation_user.h b/cpp/core_v2/internal/offline_simulation_user.h index 27a41d56..4c00d8eb 100644 --- a/cpp/core_v2/internal/offline_simulation_user.h +++ b/cpp/core_v2/internal/offline_simulation_user.h @@ -1,17 +1,3 @@ -// 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_V2_INTERNAL_OFFLINE_SIMULATION_USER_H_ #define CORE_V2_INTERNAL_OFFLINE_SIMULATION_USER_H_ diff --git a/cpp/core_v2/internal/p2p_cluster_pcp_handler.cc b/cpp/core_v2/internal/p2p_cluster_pcp_handler.cc index 41d612d7..62ab997f 100644 --- a/cpp/core_v2/internal/p2p_cluster_pcp_handler.cc +++ b/cpp/core_v2/internal/p2p_cluster_pcp_handler.cc @@ -588,7 +588,7 @@ proto::connections::Medium P2pClusterPcpHandler::StartWifiLanAdvertising( service_id.c_str()); if (!wifi_lan_medium_.StartAcceptingConnections( service_id, {.accepted_cb = [this, client, local_endpoint_name]( - WifiLanSocket& socket, + WifiLanSocket socket, const std::string& service_id) { if (!socket.IsValid()) { NEARBY_LOG(ERROR, "Invalid socket in accept callback: name=%s", diff --git a/cpp/platform_v2/base/medium_environment.cc b/cpp/platform_v2/base/medium_environment.cc index d2905ba4..164af945 100644 --- a/cpp/platform_v2/base/medium_environment.cc +++ b/cpp/platform_v2/base/medium_environment.cc @@ -152,32 +152,19 @@ void MediumEnvironment::OnWifiLanServiceStateChanged( WifiLanMediumContext& info, api::WifiLanService& service, const std::string& service_id, bool enabled) { if (!enabled_) return; - auto item = info.services.find(&service); - if (item == info.services.end()) { - NEARBY_LOG(INFO, - "G3 OnWifiLanServiceStateChanged [service impl=%p]: new service", - &service); - info.services.emplace(&service, service.GetName()); - if (enabled) { - RunOnMediumEnvironmentThread([&info, &service, service_id]() { - info.discovery_callback.service_discovered_cb(service, service_id); - }); - } + NEARBY_LOG(INFO, + "G3 OnWifiLanServiceStateChanged [service impl=%p]; context=%p, " + "notify=%d", + &info, &service, enable_notifications_.load()); + if (!enable_notifications_) return; + if (enabled) { + RunOnMediumEnvironmentThread([&info, &service, service_id]() { + info.discovery_callback.service_discovered_cb(service, service_id); + }); } else { - NEARBY_LOG(INFO, - "G3 OnWifiLanServiceStateChanged [service impl=%p]: exisitng " - "service", - &service); - if (enabled) { - RunOnMediumEnvironmentThread([&info, &service, service_id]() { - info.discovery_callback.service_discovered_cb(service, service_id); - }); - } else { - RunOnMediumEnvironmentThread([&info, &service, service_id]() { - info.discovery_callback.service_lost_cb(service, service_id); - }); - info.services.erase(item); - } + RunOnMediumEnvironmentThread([&info, &service, service_id]() { + info.discovery_callback.service_lost_cb(service, service_id); + }); } } @@ -284,14 +271,44 @@ void MediumEnvironment::SendWebRtcSignalingMessage(absl::string_view peer_id, }); } -void MediumEnvironment::RegisterWifiLanMedium(api::WifiLanMedium& medium) { +void MediumEnvironment::RegisterWifiLanMedium(api::WifiLanMedium& medium, + api::WifiLanService& service) { if (!enabled_) return; - RunOnMediumEnvironmentThread([this, &medium]() { - wifi_lan_mediums_.insert({&medium, WifiLanMediumContext{}}); + RunOnMediumEnvironmentThread([this, &medium, &service]() { + wifi_lan_mediums_.insert({&medium, WifiLanMediumContext{ + .service = &service, + }}); NEARBY_LOG(INFO, "Registered: medium=%p", &medium); }); } +void MediumEnvironment::UpdateWifiLanMediumForAdvertising( + api::WifiLanMedium& medium, api::WifiLanService& service, + const std::string& service_id, bool enabled) { + if (!enabled_) return; + RunOnMediumEnvironmentThread([this, &medium, &service, service_id, + enabled]() { + auto item = wifi_lan_mediums_.find(&medium); + if (item == wifi_lan_mediums_.end()) { + NEARBY_LOG( + INFO, "Update WifiLan medium failed. There is no medium registered."); + return; + } + auto& context = item->second; + context.advertising = enabled; + NEARBY_LOG( + INFO, + "Update WifiLan medium for advertising: this=%p; medium=%p; name=%s; " + "enabled=%d; advertising=%d", + this, &medium, service.GetName().c_str(), enabled, context.advertising); + for (auto& [local_medium, info] : wifi_lan_mediums_) { + // Do not send notification to the same medium. + if (local_medium == &medium) continue; + OnWifiLanServiceStateChanged(info, service, service_id, enabled); + } + }); +} + void MediumEnvironment::UpdateWifiLanMediumForDiscovery( api::WifiLanMedium& medium, api::WifiLanService& service, const std::string& service_id, WifiLanDiscoveredServiceCallback callback, @@ -307,16 +324,29 @@ void MediumEnvironment::UpdateWifiLanMediumForDiscovery( } auto& context = item->second; context.discovery_callback = std::move(callback); - NEARBY_LOG(INFO, "Updated: this=%p; medium=%p", this, &medium); - OnWifiLanServiceStateChanged(context, service, service_id, enabled); + NEARBY_LOG( + INFO, + "Update WifiLan medium for discovery: this=%p; medium=%p; name=%s; " + "enabled=%d; advertising=%d", + this, &medium, service.GetName().c_str(), enabled, context.advertising); + for (auto& [local_medium, info] : wifi_lan_mediums_) { + // Do not send notification to the same medium. + if (local_medium == &medium) continue; + // Search advertising mediums and send notification. + if (info.advertising && enabled) { + OnWifiLanServiceStateChanged(context, *(info.service), service_id, + enabled); + } + } }); } void MediumEnvironment::UpdateWifiLanMediumForAcceptedConnection( - api::WifiLanMedium& medium, const std::string& service_id, + api::WifiLanMedium& medium, api::WifiLanService& service, + const std::string& service_id, WifiLanAcceptedConnectionCallback accepted_connection_callback) { if (!enabled_) return; - RunOnMediumEnvironmentThread([this, &medium, + RunOnMediumEnvironmentThread([this, &medium, &service, service_id, accepted_connection_callback = std::move(accepted_connection_callback)]() { auto item = wifi_lan_mediums_.find(&medium); @@ -328,7 +358,10 @@ void MediumEnvironment::UpdateWifiLanMediumForAcceptedConnection( auto& context = item->second; context.accepted_connection_callback = std::move(accepted_connection_callback); - NEARBY_LOG(INFO, "Updated: this=%p; medium=%p", this, &medium); + NEARBY_LOG(INFO, + "Update WifiLan medium for accepted callback: this=%p; " + "medium=%p; name=%s; ", + this, &medium, service.GetName().c_str()); }); } @@ -341,5 +374,22 @@ void MediumEnvironment::UnregisterWifiLanMedium(api::WifiLanMedium& medium) { }); } +void MediumEnvironment::CallWifiLanAcceptedConnectionCallback( + api::WifiLanMedium& medium, api::WifiLanSocket& socket, + const std::string& service_id) { + if (!enabled_) return; + RunOnMediumEnvironmentThread([this, &medium, &socket, service_id]() { + auto item = wifi_lan_mediums_.find(&medium); + if (item == wifi_lan_mediums_.end()) { + NEARBY_LOG(INFO, + "Call AcceptedConnectionCallback failed.. There is no medium " + "registered."); + return; + } + auto& info = item->second; + info.accepted_connection_callback.accepted_cb(socket, service_id); + }); +} + } // namespace nearby } // namespace location diff --git a/cpp/platform_v2/base/medium_environment.h b/cpp/platform_v2/base/medium_environment.h index b34f8cf5..31fab859 100644 --- a/cpp/platform_v2/base/medium_environment.h +++ b/cpp/platform_v2/base/medium_environment.h @@ -104,17 +104,48 @@ class MediumEnvironment { // |peer_id|. void SendWebRtcSignalingMessage(absl::string_view peer_id, const ByteArray& message); - // Wifi-Lan medium registration/update calls. - void RegisterWifiLanMedium(api::WifiLanMedium& medium); + // Adds medium-related info to allow for discovery/advertising to work. + // This provides acccess to this medium from other mediums, when protocol + // expects they should communicate. + void RegisterWifiLanMedium(api::WifiLanMedium& medium, + api::WifiLanService& service); + + // Updates advertising info to indicate the current medium is exposing + // advertising event. + void UpdateWifiLanMediumForAdvertising( + api::WifiLanMedium& medium, api::WifiLanService& service, + const std::string& service_id, bool enabled); + + // Updates discovery callback info to allow for dispatch of discovery events. + // + // Invokes callback asynchronously when any changes happen to discoverable + // devices, or if the defice is turned off, whether or not it is discoverable, + // if it was ever reported as discoverable. + // + // This should be called when discoverable state changes. + // with user-specified callback when discovery is enabled, and with default + // (empty) callback otherwise. void UpdateWifiLanMediumForDiscovery( api::WifiLanMedium& medium, api::WifiLanService& service, const std::string& service_id, WifiLanDiscoveredServiceCallback discovery_callback, bool enabled); + + // Updates Accepted connection callback info to allow for dispatch of + // advertising events. void UpdateWifiLanMediumForAcceptedConnection( - api::WifiLanMedium& medium, const std::string& service_id, + api::WifiLanMedium& medium, api::WifiLanService& service, + const std::string& service_id, WifiLanAcceptedConnectionCallback accepted_connection_callback); + + // Removes medium-related info. This should correspond to device power off. void UnregisterWifiLanMedium(api::WifiLanMedium& medium); + // Call back when advertising has created the server socket and is ready for + // connect. + void CallWifiLanAcceptedConnectionCallback(api::WifiLanMedium& medium, + api::WifiLanSocket& socket, + const std::string& service_id); + private: struct BluetoothMediumContext { BluetoothDiscoveryCallback callback; @@ -126,8 +157,8 @@ class MediumEnvironment { struct WifiLanMediumContext { WifiLanDiscoveredServiceCallback discovery_callback; WifiLanAcceptedConnectionCallback accepted_connection_callback; - // discovered service vs service name map. - absl::flat_hash_map services; + api::WifiLanService* service = nullptr; + bool advertising = false; }; // This is a singleton object, for which destructor will never be called. diff --git a/cpp/platform_v2/impl/g3/wifi_lan.cc b/cpp/platform_v2/impl/g3/wifi_lan.cc index 2088c8e0..1b68f30c 100644 --- a/cpp/platform_v2/impl/g3/wifi_lan.cc +++ b/cpp/platform_v2/impl/g3/wifi_lan.cc @@ -1,5 +1,6 @@ #include "platform_v2/impl/g3/wifi_lan.h" +#include #include #include @@ -12,20 +13,45 @@ namespace location { namespace nearby { namespace g3 { -InputStream& WifiLanSocket::GetInputStream() { +WifiLanSocket::~WifiLanSocket() { absl::MutexLock lock(&mutex_); - return pipe_.GetInputStream(); + DoClose(); +} + +void WifiLanSocket::Connect(WifiLanSocket& other) { + absl::MutexLock lock(&mutex_); + remote_socket_ = &other; + input_ = other.output_; +} + +InputStream& WifiLanSocket::GetInputStream() { + auto* remote_socket = GetRemoteSocket(); + CHECK(remote_socket != nullptr); + return remote_socket->GetLocalInputStream(); } OutputStream& WifiLanSocket::GetOutputStream() { + return GetLocalOutputStream(); +} + +WifiLanSocket* WifiLanSocket::GetRemoteSocket() { absl::MutexLock lock(&mutex_); - return pipe_.GetOutputStream(); + return remote_socket_; +} + +bool WifiLanSocket::IsConnected() const { + absl::MutexLock lock(&mutex_); + return IsConnectedLocked(); +} + +bool WifiLanSocket::IsClosed() const { + absl::MutexLock lock(&mutex_); + return closed_; } Exception WifiLanSocket::Close() { absl::MutexLock lock(&mutex_); - pipe_.GetOutputStream().Close(); - pipe_.GetInputStream().Close(); + DoClose(); return {Exception::kSuccess}; } @@ -34,45 +60,215 @@ WifiLanService* WifiLanSocket::GetRemoteWifiLanService() { return service_; } +void WifiLanSocket::DoClose() { + if (!closed_) { + remote_socket_ = nullptr; + output_->GetOutputStream().Close(); + output_->GetInputStream().Close(); + if (IsConnectedLocked()) { + input_->GetOutputStream().Close(); + input_->GetInputStream().Close(); + } + closed_ = true; + } +} + +bool WifiLanSocket::IsConnectedLocked() const { return input_ != nullptr; } + +InputStream& WifiLanSocket::GetLocalInputStream() { + absl::MutexLock lock(&mutex_); + return output_->GetInputStream(); +} + +OutputStream& WifiLanSocket::GetLocalOutputStream() { + absl::MutexLock lock(&mutex_); + return output_->GetOutputStream(); +} + +std::unique_ptr WifiLanServerSocket::Accept() { + absl::MutexLock lock(&mutex_); + if (closed_) return {}; + while (pending_sockets_.empty()) { + cond_.Wait(&mutex_); + if (closed_) break; + } + if (closed_) return {}; + auto* remote_socket = + pending_sockets_.extract(pending_sockets_.begin()).value(); + CHECK(remote_socket); + auto local_socket = std::make_unique(); + local_socket->Connect(*remote_socket); + remote_socket->Connect(*local_socket); + cond_.SignalAll(); + return local_socket; +} + +bool WifiLanServerSocket::Connect(WifiLanSocket& socket) { + absl::MutexLock lock(&mutex_); + if (closed_) return false; + if (socket.IsConnected()) { + NEARBY_LOG(ERROR, + "Failed to connect to WifiLan server socket: already connected"); + return true; // already connected. + } + // add client socket to the pending list + pending_sockets_.emplace(&socket); + cond_.SignalAll(); + while (!socket.IsConnected()) { + cond_.Wait(&mutex_); + if (closed_) return false; + } + return true; +} + +void WifiLanServerSocket::SetCloseNotifier(std::function notifier) { + absl::MutexLock lock(&mutex_); + close_notifier_ = std::move(notifier); +} + +WifiLanServerSocket::~WifiLanServerSocket() { + absl::MutexLock lock(&mutex_); + DoClose(); +} + +Exception WifiLanServerSocket::Close() { + absl::MutexLock lock(&mutex_); + return DoClose(); +} + +Exception WifiLanServerSocket::DoClose() { + bool should_notify = !closed_; + closed_ = true; + if (should_notify) { + cond_.SignalAll(); + if (close_notifier_) { + auto notifier = std::move(close_notifier_); + mutex_.Unlock(); + // Notifier may contain calls to public API, and may cause deadlock, if + // mutex_ is held during the call. + notifier(); + mutex_.Lock(); + } + } + return {Exception::kSuccess}; +} + WifiLanMedium::WifiLanMedium() { + service_.SetMedium(this); auto& env = MediumEnvironment::Instance(); - env.RegisterWifiLanMedium(*this); + env.RegisterWifiLanMedium(*this, service_); } WifiLanMedium::~WifiLanMedium() { + service_.SetMedium(nullptr); auto& env = MediumEnvironment::Instance(); env.UnregisterWifiLanMedium(*this); + + StopAdvertising(advertising_info_.service_id); + StopDiscovery(discovering_info_.service_id); + + accept_loops_runner_.Shutdown(); + NEARBY_LOG(INFO, + "WifiLanMedium dtor advertising_accept_thread_running_ = %d", + acceptance_thread_running_.load()); + // If acceptance thread is still running, wait to finish. + if (acceptance_thread_running_) { + while (acceptance_thread_running_) { + CountDownLatch latch(1); + close_accept_loops_runner_.Execute([&latch]() { latch.CountDown(); }); + latch.Await(); + } + } } bool WifiLanMedium::StartAdvertising( const std::string& service_id, const std::string& wifi_lan_service_info_name) { - // TODO(edwinwu): Integrate medium_environment. - // steps: - // 1. create wifi_lan_service as the parameter to create wifi_lan_socket - // auto service = std::make_unique(); - // auto socket = std::make_unique(service); - // 2. callback for accepting connection; otherwise don't callback if not - // accepted connection. - // accepted_connection_callback_.accepted_cb(socket, service_id); + NEARBY_LOG(INFO, + "G3 WifiLan StartAdvertising: service_id=%s, service_name=%s", + service_id.c_str(), wifi_lan_service_info_name.c_str()); + auto& env = MediumEnvironment::Instance(); + env.UpdateWifiLanMediumForAdvertising(*this, service_, service_id, true); + + absl::MutexLock lock(&mutex_); + if (server_socket_ != nullptr) server_socket_.release(); + server_socket_ = std::make_unique(); + + acceptance_thread_running_.exchange(true); + accept_loops_runner_.Execute([&env, this, service_id]() mutable { + if (!accept_loops_runner_.InShutdown()) { + while (true) { + auto client_socket = server_socket_->Accept(); + if (client_socket == nullptr) break; + env.CallWifiLanAcceptedConnectionCallback(*this, *client_socket, + service_id); + } + } + acceptance_thread_running_.exchange(false); + }); + advertising_info_.service_id = service_id; return true; } bool WifiLanMedium::StopAdvertising(const std::string& service_id) { - // TODO(edwinwu): Integrate medium_environment. + NEARBY_LOG(INFO, "G3 WifiLan StopAdvertising: service_id=%s", + service_id.c_str()); + { + absl::MutexLock lock(&mutex_); + if (advertising_info_.Empty()) { + NEARBY_LOG( + INFO, "Can't stop advertising because we never started advertising."); + return false; + } + advertising_info_.Clear(); + } + + auto& env = MediumEnvironment::Instance(); + env.UpdateWifiLanMediumForAdvertising(*this, service_, service_id, false); + accept_loops_runner_.Shutdown(); + if (server_socket_ == nullptr) { + NEARBY_LOG(ERROR, "Failed to find WifiLan Server socket: service_id=%s", + service_id.c_str()); + // Fall through for server socket not found. + return true; + } + + if (!server_socket_->Close().Ok()) { + NEARBY_LOG(INFO, "Failed to close WifiLan server socket for %s.", + service_id.c_str()); + return false; + } + return true; } bool WifiLanMedium::StartDiscovery(const std::string& service_id, DiscoveredServiceCallback callback) { + NEARBY_LOG(INFO, "G3 WifiLan StartDiscovery: service_id=%s", + service_id.c_str()); auto& env = MediumEnvironment::Instance(); - NEARBY_LOG(INFO, "G3 StartDiscovery: service_id=%s", service_id.c_str()); env.UpdateWifiLanMediumForDiscovery(*this, service_, service_id, std::move(callback), true); + { + absl::MutexLock lock(&mutex_); + discovering_info_.service_id = service_id; + } return true; } bool WifiLanMedium::StopDiscovery(const std::string& service_id) { + NEARBY_LOG(INFO, "G3 WifiLan StopDiscovery: service_id=%s", + service_id.c_str()); + { + absl::MutexLock lock(&mutex_); + if (discovering_info_.Empty()) { + NEARBY_LOG( + INFO, "Can't stop discovering because we never started discovering."); + return false; + } + discovering_info_.Clear(); + } + auto& env = MediumEnvironment::Instance(); env.UpdateWifiLanMediumForDiscovery(*this, service_, service_id, {}, false); return true; @@ -80,33 +276,58 @@ bool WifiLanMedium::StopDiscovery(const std::string& service_id) { bool WifiLanMedium::StartAcceptingConnections( const std::string& service_id, AcceptedConnectionCallback callback) { - // TODO(edwinwu): Integrate medium_environment. - // steps: + NEARBY_LOG(INFO, "G3 WifiLan StartAcceptingConnections: service_id=%s", + service_id.c_str()); auto& env = MediumEnvironment::Instance(); - env.UpdateWifiLanMediumForAcceptedConnection(*this, service_id, callback); + env.UpdateWifiLanMediumForAcceptedConnection(*this, service_, service_id, + callback); return true; } bool WifiLanMedium::StopAcceptingConnections(const std::string& service_id) { - // TODO(edwinwu): Integrate medium_environment. + NEARBY_LOG(INFO, "G3 WifiLan StopAcceptingConnections: service_id=%s", + service_id.c_str()); auto& env = MediumEnvironment::Instance(); - env.UpdateWifiLanMediumForAcceptedConnection(*this, service_id, {}); + env.UpdateWifiLanMediumForAcceptedConnection(*this, service_, service_id, {}); return true; } std::unique_ptr WifiLanMedium::Connect( - api::WifiLanService& service, const std::string& service_id) { + api::WifiLanService& remote_service, const std::string& service_id) { + NEARBY_LOG(INFO, "G3 WifiLan Connect: medium=%p, service=%p, service_id=%s", + this, &service_, service_id.c_str()); + // First, find an instance of remote medium, that exposed this service. + auto* medium = static_cast(remote_service).GetMedium(); + + if (!medium) return {}; // Can't find medium. Bail out. + + WifiLanServerSocket* server_socket = nullptr; + NEARBY_LOG(INFO, + "G3 WifiLan Connect [peer]: medium=%p, service=%p, service_id=%s", + medium, &remote_service, service_id.c_str()); + // Then, find our server socket context in this medium. + { + absl::MutexLock medium_lock(&medium->mutex_); + server_socket = medium->server_socket_.get(); + if (server_socket == nullptr) { + NEARBY_LOG(ERROR, "Failed to find WifiLan Server socket: service_id=%s", + service_id.c_str()); + return {}; + } + } + auto socket = std::make_unique(); - NEARBY_LOG(INFO, "G3 Connect: medium=%p, service_id=%s", this, - service_id.c_str()); + // Finally, Request to connect to this socket. + if (!server_socket->Connect(*socket)) { + NEARBY_LOG( + ERROR, + "Failed to connect to existing WifiLan Server socket: service_id=%s", + service_id.c_str()); + return {}; + } + + NEARBY_LOG(INFO, "G3 WifiLan Connect: connected: socket=%p", socket.get()); return socket; - // TODO(edwinwu): Integrate medium_environment. - // steps: - // Request a connection, and block until the socket is provided via the - // callback. - // 1. connection = wifi_lan_service.requestConnection_(); - // 2. create wifi_lan_socket with wifi_lan_service and connection - // return wifi_lan_socket; } } // namespace g3 diff --git a/cpp/platform_v2/impl/g3/wifi_lan.h b/cpp/platform_v2/impl/g3/wifi_lan.h index c8995c02..45bdfbfd 100644 --- a/cpp/platform_v2/impl/g3/wifi_lan.h +++ b/cpp/platform_v2/impl/g3/wifi_lan.h @@ -1,20 +1,25 @@ #ifndef PLATFORM_V2_IMPL_G3_WIFI_LAN_H_ #define PLATFORM_V2_IMPL_G3_WIFI_LAN_H_ +#include #include #include "platform_v2/api/wifi_lan.h" #include "platform_v2/base/byte_array.h" #include "platform_v2/base/input_stream.h" #include "platform_v2/base/output_stream.h" +#include "platform_v2/impl/g3/multi_thread_executor.h" #include "platform_v2/impl/g3/pipe.h" #include "absl/container/flat_hash_map.h" +#include "absl/container/flat_hash_set.h" #include "absl/synchronization/mutex.h" namespace location { namespace nearby { namespace g3 { +class WifiLanMedium; + // Opaque wrapper over a WifiLan service which contains encoded WifiLan service // info name. class WifiLanService : public api::WifiLanService { @@ -25,19 +30,23 @@ class WifiLanService : public api::WifiLanService { void SetName(std::string name) { name_ = std::move(name); } std::string GetName() const override { return name_; } + void SetMedium(WifiLanMedium* medium) { medium_ = medium; } + WifiLanMedium* GetMedium() { return medium_; } + private: std::string name_; + WifiLanMedium* medium_ = nullptr; }; class WifiLanSocket : public api::WifiLanSocket { public: WifiLanSocket() = default; explicit WifiLanSocket(WifiLanService* service) : service_(service) {} - ~WifiLanSocket() override = default; + ~WifiLanSocket() override; // Connect to another WifiLanSocket, to form a functional low-level channel. // from this point on, and until Close is called, connection exists. - void ConnectTo(WifiLanSocket* other) ABSL_LOCKS_EXCLUDED(mutex_); + void Connect(WifiLanSocket& other) ABSL_LOCKS_EXCLUDED(mutex_); // Returns the InputStream of this connected WifiLanSocket. InputStream& GetInputStream() override ABSL_LOCKS_EXCLUDED(mutex_); @@ -46,6 +55,15 @@ class WifiLanSocket : public api::WifiLanSocket { // This stream is for local side to write. OutputStream& GetOutputStream() override ABSL_LOCKS_EXCLUDED(mutex_); + // Returns address of a remote WifiLanSocket or nullptr. + WifiLanSocket* GetRemoteSocket() ABSL_LOCKS_EXCLUDED(mutex_); + + // Returns true if connection exists to the (possibly closed) remote socket. + bool IsConnected() const ABSL_LOCKS_EXCLUDED(mutex_); + + // Returns true if socket is closed. + bool IsClosed() const ABSL_LOCKS_EXCLUDED(mutex_); + // Returns Exception::kIo on error, Exception::kSuccess otherwise. Exception Close() override ABSL_LOCKS_EXCLUDED(mutex_); @@ -55,9 +73,75 @@ class WifiLanSocket : public api::WifiLanSocket { ABSL_LOCKS_EXCLUDED(mutex_); private: - Pipe pipe_; - WifiLanService* service_; + void DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + // Returns true if connection exists to the (possibly closed) remote socket. + bool IsConnectedLocked() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + // Returns InputStream of our side of a connection. + // This is what the remote side is supposed to read from. + // This is a helper for GetInputStream() method. + InputStream& GetLocalInputStream() ABSL_LOCKS_EXCLUDED(mutex_); + + // Returns OutputStream of our side of a connection. + // This is what the local size is supposed to write to. + // This is a helper for GetOutputStream() method. + OutputStream& GetLocalOutputStream() ABSL_LOCKS_EXCLUDED(mutex_); + + // Output pipe is initialized by constructor, it remains always valid, until + // it is closed. it represents output part of a local socket. Input part of a + // local socket comes from the peer socket, after connection. + std::shared_ptr output_ {new Pipe}; + std::shared_ptr input_; mutable absl::Mutex mutex_; + WifiLanService* service_; + WifiLanSocket* remote_socket_ ABSL_GUARDED_BY(mutex_) = nullptr; + bool closed_ ABSL_GUARDED_BY(mutex_) = false; +}; + +class WifiLanServerSocket { + public: + ~WifiLanServerSocket(); + + // Blocks until either: + // - at least one incoming connection request is available, or + // - ServerSocket is closed. + // On success, returns connected socket, ready to exchange data. + // Returns nullptr on error. + // Once error is reported, it is permanent, and ServerSocket has to be closed. + // + // Called by the server side of a connection. + // Returns WifiLanSocket to the server side. + // If not null, returned socket is connected to its remote (client-side) peer. + std::unique_ptr Accept() ABSL_LOCKS_EXCLUDED(mutex_); + + // Blocks until either: + // - connection is available, or + // - server socket is closed, or + // - error happens. + // + // Called by the client side of a connection. + // Returns true, if socket is successfully connected. + bool Connect(WifiLanSocket& socket) ABSL_LOCKS_EXCLUDED(mutex_); + + // Called by the server side of a connection before passing ownership of + // WifiLanServerSocker to user, to track validity of a pointer to this + // server socket, + void SetCloseNotifier(std::function notifier) + ABSL_LOCKS_EXCLUDED(mutex_); + + // Returns Exception::kIo on error, Exception::kSuccess otherwise. + // Calls close_notifier if it was previously set, and marks socket as closed. + Exception Close() ABSL_LOCKS_EXCLUDED(mutex_); + + private: + Exception DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + absl::Mutex mutex_; + absl::CondVar cond_; + absl::flat_hash_set pending_sockets_ ABSL_GUARDED_BY(mutex_); + std::function close_notifier_ ABSL_GUARDED_BY(mutex_); + bool closed_ ABSL_GUARDED_BY(mutex_) = false; }; // Container of operations that can be performed over the WifiLan medium. @@ -91,15 +175,48 @@ class WifiLanMedium : public api::WifiLanMedium { bool StopAcceptingConnections(const std::string& service_id) override ABSL_LOCKS_EXCLUDED(mutex_); - // Returns a new WifiLanSocket. On Success, WifiLanSocket::IsValid() - // returns true. + // Connects to existing remote WifiLan service. + // + // On success, returns a new WifiLanSocket. + // On error, returns nullptr. std::unique_ptr Connect( - api::WifiLanService& service, const std::string& service_id) override - ABSL_LOCKS_EXCLUDED(mutex_); + api::WifiLanService& remote_service, + const std::string& service_id) override ABSL_LOCKS_EXCLUDED(mutex_); private: + static constexpr int kMaxConcurrentAcceptLoops = 5; + + struct AdvertisingInfo { + bool Empty() const { return service_id.empty(); } + void Clear() { service_id.clear(); } + + std::string service_id; + }; + + struct DiscoveringInfo { + bool Empty() const { return service_id.empty(); } + void Clear() { service_id.clear(); } + + std::string service_id; + }; + absl::Mutex mutex_; WifiLanService service_{"wifi_lan_service_info_name"}; + + // A thread pool dedicated to running all the accept loops from + // StartAdvertising(). + MultiThreadExecutor accept_loops_runner_{kMaxConcurrentAcceptLoops}; + std::atomic_bool acceptance_thread_running_ = false; + + // A thread pool dedicated to wait to complete the accept_loops_runner_. + MultiThreadExecutor close_accept_loops_runner_{kMaxConcurrentAcceptLoops}; + + // TODO(edwinwu): Extend it to hashmap to accept multiple sockets for multiple + // entrance. + // A server socket is established when start advertising. + std::unique_ptr server_socket_; + AdvertisingInfo advertising_info_ ABSL_GUARDED_BY(mutex_); + DiscoveringInfo discovering_info_ ABSL_GUARDED_BY(mutex_); }; } // namespace g3 diff --git a/cpp/platform_v2/public/BUILD b/cpp/platform_v2/public/BUILD index 66925ef6..59902e1e 100644 --- a/cpp/platform_v2/public/BUILD +++ b/cpp/platform_v2/public/BUILD @@ -93,6 +93,7 @@ cc_test( "atomic_reference_test.cc", "bluetooth_adapter_test.cc", "bluetooth_classic_test.cc", + "cancelable_alarm_test.cc", "condition_variable_test.cc", "count_down_latch_test.cc", "crypto_test.cc", diff --git a/cpp/platform_v2/public/cancelable.h b/cpp/platform_v2/public/cancelable.h index 3105648b..83f10291 100644 --- a/cpp/platform_v2/public/cancelable.h +++ b/cpp/platform_v2/public/cancelable.h @@ -24,7 +24,9 @@ class Cancelable final { explicit Cancelable(std::shared_ptr impl) : impl_(std::move(impl)) {} - bool Cancel() { return impl_->Cancel(); } + bool Cancel() { return impl_ ? impl_->Cancel() : false; } + + bool IsValid() { return impl_ != nullptr; } private: std::shared_ptr impl_; diff --git a/cpp/platform_v2/public/cancelable_alarm.h b/cpp/platform_v2/public/cancelable_alarm.h index 1fc26788..d00d6241 100644 --- a/cpp/platform_v2/public/cancelable_alarm.h +++ b/cpp/platform_v2/public/cancelable_alarm.h @@ -21,6 +21,7 @@ namespace nearby { */ class CancelableAlarm { public: + CancelableAlarm() = default; CancelableAlarm(absl::string_view name, std::function&& runnable, absl::Duration delay, ScheduledExecutor* scheduled_executor) : name_(name), @@ -44,6 +45,10 @@ class CancelableAlarm { return cancelable_.Cancel(); } + bool IsValid() { + return cancelable_.IsValid(); + } + private: Mutex mutex_; std::string name_; diff --git a/cpp/platform_v2/public/cancelable_alarm_test.cc b/cpp/platform_v2/public/cancelable_alarm_test.cc new file mode 100644 index 00000000..5bebf2cb --- /dev/null +++ b/cpp/platform_v2/public/cancelable_alarm_test.cc @@ -0,0 +1,54 @@ +#include "platform_v2/public/cancelable_alarm.h" + +#include "platform_v2/public/atomic_boolean.h" +#include "platform_v2/public/scheduled_executor.h" +#include "gmock/gmock.h" +#include "gtest/gtest.h" +#include "absl/time/time.h" + +namespace location { +namespace nearby { +namespace { + +TEST(CancelableAlarmTest, CanCreateDefault) { CancelableAlarm alarm; } + +TEST(CancelableAlarmTest, CancelDefaultFails) { + CancelableAlarm alarm; + EXPECT_FALSE(alarm.Cancel()); +} + +TEST(CancelableAlarmTest, CanCreateAndFireAlarm) { + ScheduledExecutor alarm_executor; + AtomicBoolean done{false}; + CancelableAlarm alarm( + "test_alarm", [&done]() { done.Set(true); }, absl::Milliseconds(100), + &alarm_executor); + SystemClock::Sleep(absl::Milliseconds(1000)); + EXPECT_TRUE(done.Get()); +} + +TEST(CancelableAlarmTest, CanCreateAndCancelAlarm) { + ScheduledExecutor alarm_executor; + AtomicBoolean done{false}; + CancelableAlarm alarm( + "test_alarm", [&done]() { done.Set(true); }, absl::Milliseconds(100), + &alarm_executor); + EXPECT_TRUE(alarm.Cancel()); + SystemClock::Sleep(absl::Milliseconds(1000)); + EXPECT_FALSE(done.Get()); +} + +TEST(CancelableAlarmTest, CancelExpiredAlarmFails) { + ScheduledExecutor alarm_executor; + AtomicBoolean done{false}; + CancelableAlarm alarm( + "test_alarm", [&done]() { done.Set(true); }, absl::Milliseconds(100), + &alarm_executor); + SystemClock::Sleep(absl::Milliseconds(1000)); + EXPECT_TRUE(done.Get()); + EXPECT_FALSE(alarm.Cancel()); +} + +} // namespace +} // namespace nearby +} // namespace location diff --git a/cpp/platform_v2/public/wifi_lan.cc b/cpp/platform_v2/public/wifi_lan.cc index 32eefa18..e1894a17 100644 --- a/cpp/platform_v2/public/wifi_lan.cc +++ b/cpp/platform_v2/public/wifi_lan.cc @@ -48,6 +48,7 @@ bool WifiLanMedium::StartDiscovery(const std::string& service_id, [this](api::WifiLanService& service, const std::string& service_id) { MutexLock lock(&mutex_); + if (services_.empty()) return; auto item = services_.extract(&service); auto& context = *item.mapped(); NEARBY_LOG(INFO, "Removing service=%p, impl=%p", @@ -87,9 +88,8 @@ bool WifiLanMedium::StartAcceptingConnections( if (!pair.second) { NEARBY_LOG(INFO, "Adding (again) socket=%p, impl=%p", &context.socket, &socket); - return; + context.socket = WifiLanSocket(&socket); } - context.socket = WifiLanSocket(&socket); NEARBY_LOG(INFO, "Adding socket=%p, impl=%p", &context.socket, &socket); accepted_connection_callback_.accepted_cb(context.socket, @@ -106,7 +106,7 @@ bool WifiLanMedium::StopAcceptingConnections(const std::string& service_id) { NEARBY_LOG(INFO, "WifiLan accepted connection disabled: impl=%p", &GetImpl()); } - return impl_->StopDiscovery(service_id); + return impl_->StopAcceptingConnections(service_id); } WifiLanSocket WifiLanMedium::Connect(WifiLanService& service, diff --git a/cpp/platform_v2/public/wifi_lan.h b/cpp/platform_v2/public/wifi_lan.h index 7274414f..f2403f04 100644 --- a/cpp/platform_v2/public/wifi_lan.h +++ b/cpp/platform_v2/public/wifi_lan.h @@ -102,8 +102,8 @@ class WifiLanMedium final { }; struct AcceptedConnectionCallback { - std::function - accepted_cb = DefaultCallback(); + std::function + accepted_cb = DefaultCallback(); }; struct AcceptedConnectionInfo { WifiLanSocket socket; diff --git a/cpp/platform_v2/public/wifi_lan_test.cc b/cpp/platform_v2/public/wifi_lan_test.cc index 398fa242..8a701efa 100644 --- a/cpp/platform_v2/public/wifi_lan_test.cc +++ b/cpp/platform_v2/public/wifi_lan_test.cc @@ -13,10 +13,12 @@ namespace nearby { namespace { constexpr absl::string_view kServiceID{"com.google.location.nearby.apps.test"}; +constexpr absl::string_view kServiceName{"service name"}; class WifiLanMediumTest : public ::testing::Test { protected: using DiscoveredServiceCallback = WifiLanMedium::DiscoveredServiceCallback; + using AcceptedConnectionCallback = WifiLanMedium::AcceptedConnectionCallback; WifiLanMediumTest() { env_.Stop(); } @@ -25,75 +27,149 @@ class WifiLanMediumTest : public ::testing::Test { TEST_F(WifiLanMediumTest, ConstructorDestructorWorks) { env_.Start(); - WifiLanMedium medium_a; - WifiLanMedium medium_b; + WifiLanMedium wifi_a; + WifiLanMedium wifi_b; // Make sure we can create functional mediums. - ASSERT_TRUE(medium_a.IsValid()); - ASSERT_TRUE(medium_b.IsValid()); + ASSERT_TRUE(wifi_a.IsValid()); + ASSERT_TRUE(wifi_b.IsValid()); // Make sure we can create 2 distinct mediums. - EXPECT_NE(&medium_a.GetImpl(), &medium_b.GetImpl()); + EXPECT_NE(&wifi_a.GetImpl(), &wifi_b.GetImpl()); env_.Stop(); } -TEST_F(WifiLanMediumTest, CanStartDiscoveryAndServiceIndeedDiscovered) { +TEST_F(WifiLanMediumTest, CanStartAdvertising) { env_.Start(); - WifiLanMedium medium; + WifiLanMedium wifi_a; + WifiLanMedium wifi_b; + std::string service_id(kServiceID); + std::string service_name{kServiceName}; + CountDownLatch found_latch(1); + + wifi_a.StartAdvertising(service_id, service_name); + + EXPECT_TRUE(wifi_b.StartDiscovery( + service_id, DiscoveredServiceCallback{ + .service_discovered_cb = + [&found_latch](WifiLanService& service, + const std::string& service_id) { + found_latch.CountDown(); + }, + })); + EXPECT_TRUE(found_latch.Await(absl::Milliseconds(1000)).result()); + EXPECT_TRUE(wifi_a.StopAdvertising(service_id)); + EXPECT_TRUE(wifi_b.StopDiscovery(service_id)); + env_.Stop(); +} + +TEST_F(WifiLanMediumTest, CanStartDiscovery) { + env_.Start(); + WifiLanMedium wifi_a; + WifiLanMedium wifi_b; + std::string service_id(kServiceID); + std::string service_name{kServiceName}; CountDownLatch found_latch(1); CountDownLatch lost_latch(1); - medium.StartDiscovery(std::string(kServiceID), + wifi_a.StartDiscovery(service_id, DiscoveredServiceCallback{ .service_discovered_cb = [&found_latch](WifiLanService& service, const std::string& service_id) { - NEARBY_LOG(INFO, "Service discovered: %s", - service.GetName().c_str()); - EXPECT_EQ(kServiceID, service_id); found_latch.CountDown(); }, .service_lost_cb = [&lost_latch](WifiLanService& service, const std::string& service_id) { - NEARBY_LOG(INFO, "Service lost: %s", - service.GetName().c_str()); - EXPECT_EQ(kServiceID, service_id); lost_latch.CountDown(); }, }); + EXPECT_TRUE(wifi_b.StartAdvertising(service_id, service_name)); EXPECT_TRUE(found_latch.Await(absl::Milliseconds(1000)).result()); + EXPECT_TRUE(wifi_b.StopAdvertising(service_id)); + EXPECT_TRUE(lost_latch.Await(absl::Milliseconds(1000)).result()); + EXPECT_TRUE(wifi_a.StopDiscovery(service_id)); env_.Stop(); } TEST_F(WifiLanMediumTest, CanStopDiscovery) { env_.Start(); - WifiLanMedium medium; + WifiLanMedium wifi_a; + WifiLanMedium wifi_b; + std::string service_id(kServiceID); + std::string service_name{kServiceName}; CountDownLatch found_latch(1); CountDownLatch lost_latch(1); - medium.StartDiscovery(std::string(kServiceID), + wifi_a.StartDiscovery(service_id, DiscoveredServiceCallback{ .service_discovered_cb = [&found_latch](WifiLanService& service, const std::string& service_id) { - NEARBY_LOG(INFO, "Service discovered: %s", - service.GetName().c_str()); - EXPECT_EQ(kServiceID, service_id); found_latch.CountDown(); }, .service_lost_cb = [&lost_latch](WifiLanService& service, const std::string& service_id) { - NEARBY_LOG(INFO, "Service lost: %s", - service.GetName().c_str()); - EXPECT_EQ(kServiceID, service_id); lost_latch.CountDown(); }, }); + EXPECT_TRUE(wifi_b.StartAdvertising(service_id, service_name)); EXPECT_TRUE(found_latch.Await(absl::Milliseconds(1000)).result()); - bool stop = medium.StopDiscovery(std::string(kServiceID)); - EXPECT_TRUE(stop); + EXPECT_TRUE(wifi_a.StopDiscovery(service_id)); + EXPECT_TRUE(wifi_b.StopAdvertising(service_id)); + EXPECT_FALSE(lost_latch.Await(absl::Milliseconds(1000)).result()); + env_.Stop(); +} + +TEST_F(WifiLanMediumTest, CanStartAcceptingConnectionsAndConnect) { + env_.Start(); + WifiLanMedium wifi_a; + WifiLanMedium wifi_b; + std::string service_id(kServiceID); + std::string service_name{kServiceName}; + CountDownLatch found_latch(1); + CountDownLatch accepted_latch(1); + + WifiLanService* discovered_service = nullptr; + wifi_a.StartDiscovery( + service_id, + DiscoveredServiceCallback{ + .service_discovered_cb = + [&found_latch, &discovered_service]( + WifiLanService& service, const std::string& service_id) { + NEARBY_LOG(INFO, "Service discovered: %s, %p", + service.GetName().c_str(), &service); + discovered_service = &service; + found_latch.CountDown(); + }, + }); + wifi_b.StartAdvertising(service_id, service_name); + wifi_b.StartAcceptingConnections( + service_id, + AcceptedConnectionCallback{ + .accepted_cb = [&accepted_latch](WifiLanSocket socket, + const std::string& service_id) { + NEARBY_LOG(INFO, "Connection accepted: socket=%p, service_id=%s", + &socket, service_id.c_str()); + accepted_latch.CountDown(); + }}); + EXPECT_TRUE(found_latch.Await(absl::Milliseconds(1000)).result()); + + WifiLanSocket socket_a; + EXPECT_FALSE(socket_a.IsValid()); + { + SingleThreadExecutor client_executor; + client_executor.Execute( + [&wifi_a, &socket_a, discovered_service, &service_id]() { + socket_a = wifi_a.Connect(*discovered_service, service_id); + }); + } + EXPECT_TRUE(accepted_latch.Await(absl::Milliseconds(1000)).result()); + EXPECT_TRUE(socket_a.IsValid()); + wifi_b.StopAdvertising(service_id); + wifi_a.StopDiscovery(service_id); env_.Stop(); } diff --git a/proto/error_code_enums.proto b/proto/error_code_enums.proto index 62232412..0f463a0b 100644 --- a/proto/error_code_enums.proto +++ b/proto/error_code_enums.proto @@ -103,9 +103,10 @@ enum StartAdvertisingError { // System error, all advertising slot ran out, can't available for new // regular advertisement. BLE_MAX_GATT_ADVERTISEMENT_SLOT_REACHED = 35; - // System error, failed to start advertising for legacy advertisements + // System error, failed to start advertising for legacy advertisements on BLE START_LEGACY_ADVERTISING_FAILED = 36; - // System error, failed to start advertising for extended advertisements + // System error, failed to start advertising for extended advertisements on + // BLE START_EXTENDED_ADVERTISING_FAILED = 38; // System error, there's already someone advertising on Bluetooth, not allow // to start another one. @@ -128,6 +129,22 @@ enum StartAdvertisingError { // Next ID :46 } +// The error for event START_DISCOVERING. The range between 31 and 99. +enum StartDiscoveringError { + // Developing error, this service ID already requested, should not request it + // again without stop discovering. + DUPLICATE_DISCOVERING_REQUESTED = 31; + // System error, failed to start discovering for legacy advertisements on BLE + START_LEGACY_DISCOVERING_FAILED = 32; + // System error, failed to start discovering for extended advertisements on + // BLE + START_EXTENDED_DISCOVERING_FAILED = 33; + // System error, failed to start discovering. + START_DISCOVERING_FAILED = 34; + + // Next ID :34 +} + enum Description { reserved 28; @@ -171,4 +188,15 @@ enum Description { NULL_WIFI_AWARE_MANAGER = 38; STALE_ANDROID_VERSION = 39; NULL_SERVICE_INFO = 40; + NULL_WORK_SOURCE = 41; + NULL_CALLBACK = 42; + NULL_BLUETOOTH_LE_SCANNER_COMPAT = 43; + EMPTY_WORK_SOURCE_CACHE = 44; + SCAN_FAILED_ALREADY_STARTED = 45; + SCAN_FAILED_APPLICATION_REGISTRATION_FAILED = 46; + SCAN_FAILED_INTERNAL_ERROR = 47; + SCAN_FAILED_FEATURE_UNSUPPORTED = 48; + SCAN_FAILED_BLUETOOTH_DISABLED = 49; + SCAN_FILTERS_NOT_ALLOWED_FOR_LOCATION = 50; + BLUETOOTH_SCAN_REJUVENATE_FAILED = 51; }