TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-14 14:46:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

Implement multiplex for WIFI_LAN

Browse Source 
PiperOrigin-RevId: 694239460
This commit is contained in:
hai007
2024-11-07 14:02:22 -08:00
committed by Copybara-Service
parent 0a0c915a79
commit cb07eead88
28 changed files with 516 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connections/implementation/base_endpoint_channel.cc
+5
View File
@@ -331,6 +331,11 @@ void BaseEndpointChannel::Close(
}
}
bool BaseEndpointChannel::IsClosed() const {
MutexLock lock(&is_paused_mutex_);
return is_closed_;
}
std::string BaseEndpointChannel::GetType() const {
MutexLock crypto_lock(&crypto_mutex_);
std::string subtype = IsEncryptionEnabledLocked() ? "ENCRYPTED_" : "";
connections/implementation/base_endpoint_channel.h
+1 -1
View File
@@ -28,7 +28,6 @@
#include "internal/platform/input_stream.h"
#include "internal/platform/mutex.h"
#include "internal/platform/output_stream.h"
#include "internal/platform/socket.h"
namespace nearby {
namespace connections {
@@ -63,6 +62,7 @@ class BaseEndpointChannel : public EndpointChannel {
location::nearby::proto::connections::DisconnectionReason reason,
location::nearby::analytics::proto::ConnectionsLog::
EstablishedConnection::SafeDisconnectionResult result) override;
bool IsClosed() const ABSL_LOCKS_EXCLUDED(is_paused_mutex_) override;
std::string GetType() const override;
std::string GetServiceId() const override;
std::string GetName() const override;
connections/implementation/client_proxy.cc
+4 -2
View File
@@ -1193,9 +1193,11 @@ std::int32_t ClientProxy::GetLocalMultiplexSocketBitmask() const {
if (NearbyFlags::GetInstance().GetBoolFlag(
config_package_nearby::nearby_connections_feature::
kEnableMultiplex)) {
std::int32_t multiplex_bitmask =
kBtMultiplexEnabled | kWifiLanMultiplexEnabled;
NEARBY_LOGS(INFO) << "ClientProxy [GetLocalMultiplexSocketBitmask]: "
<< kBtMultiplexEnabled;
return kBtMultiplexEnabled;
<< multiplex_bitmask;
return multiplex_bitmask;
}
return 0;
}
connections/implementation/client_proxy_test.cc
+8 -4
View File
@@ -1554,19 +1554,23 @@ TEST_F(ClientProxyTest, TestAutoBwuWhenListeningWithAutoBwu) {
}
TEST_F(ClientProxyTest, TestMultiplexSocketBitmask) {
EXPECT_EQ(client1()->GetLocalMultiplexSocketBitmask(), 0);
if (!NearbyFlags::GetInstance().GetBoolFlag(
config_package_nearby::nearby_connections_feature::
kEnableMultiplex)) {
EXPECT_EQ(client1()->GetLocalMultiplexSocketBitmask(), 0);
}
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::kEnableMultiplex,
true);
EXPECT_EQ(client1()->GetLocalMultiplexSocketBitmask(),
ClientProxy::kBtMultiplexEnabled);
ClientProxy::kBtMultiplexEnabled |
ClientProxy::kWifiLanMultiplexEnabled);
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::kEnableMultiplex,
false);
}
TEST_F(ClientProxyTest, TestRemoteMultiplexSocketBitmask) {
EXPECT_EQ(client1()->GetLocalMultiplexSocketBitmask(), 0);
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::kEnableMultiplex,
true);
@@ -1586,7 +1590,7 @@ TEST_F(ClientProxyTest, TestRemoteMultiplexSocketBitmask) {
ClientProxy::kBtMultiplexEnabled | ClientProxy::kWifiLanMultiplexEnabled);
EXPECT_TRUE(client1()->IsMultiplexSocketSupported(advertising_endpoint.id,
Medium::BLUETOOTH));
EXPECT_FALSE(client1()->IsMultiplexSocketSupported(advertising_endpoint.id,
EXPECT_TRUE(client1()->IsMultiplexSocketSupported(advertising_endpoint.id,
Medium::WIFI_LAN));
EXPECT_FALSE(client1()->IsMultiplexSocketSupported(advertising_endpoint.id,
Medium::WIFI_AWARE));
connections/implementation/connections_authentication_transport_test.cc
+1
View File
@@ -51,6 +51,7 @@ class MockEndpointChannel : public EndpointChannel {
location::nearby::analytics::proto::ConnectionsLog::
EstablishedConnection::SafeDisconnectionResult result),
(override));
MOCK_METHOD(bool, IsClosed, (), (const, override));
MOCK_METHOD(std::string, GetType, (), (const, override));
MOCK_METHOD(std::string, GetServiceId, (), (const, override));
MOCK_METHOD(std::string, GetName, (), (const, override));
connections/implementation/encryption_runner_test.cc
+1 -1
View File
@@ -28,7 +28,6 @@
#include "internal/platform/input_stream.h"
#include "internal/platform/output_stream.h"
#include "internal/platform/pipe.h"
#include "internal/platform/system_clock.h"
#include "proto/connections_enums.pb.h"
#include "third_party/ukey2/src/main/cpp/include/securegcm/ukey2_handshake.h"
@@ -74,6 +73,7 @@ class FakeEndpointChannel : public EndpointChannel {
EstablishedConnection::SafeDisconnectionResult result) override {
Close();
}
bool IsClosed() const override { return false; }
location::nearby::proto::connections::ConnectionTechnology GetTechnology()
const override {
return location::nearby::proto::connections::ConnectionTechnology::
connections/implementation/endpoint_channel.h
+3
View File
@@ -59,6 +59,9 @@ class EndpointChannel {
location::nearby::analytics::proto::ConnectionsLog::
EstablishedConnection::SafeDisconnectionResult result) = 0;
// True if the EndpointChannel is currently closed.
virtual bool IsClosed() const = 0;
// Returns a one-word type descriptor for the concrete EndpointChannel
// implementation that can be used in log messages; eg: BLUETOOTH, BLE, WIFI.
virtual std::string GetType() const = 0;
connections/implementation/endpoint_channel_manager.cc
+5 -6
View File
@@ -23,12 +23,10 @@
#include "connections/implementation/endpoint_channel.h"
#include "connections/implementation/offline_frames.h"
#include "internal/platform/condition_variable.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/implementation/system_clock.h"
#include "internal/platform/logging.h"
#include "internal/platform/mutex.h"
#include "internal/platform/mutex_lock.h"
#include "proto/connections_enums.pb.h"
namespace nearby {
namespace connections {
@@ -229,13 +227,14 @@ bool EndpointChannelManager::ChannelState::RemoveEndpoint(
auto item = endpoints_.find(endpoint_id);
if (item == endpoints_.end()) return false;
MarkEndpointStopWaitToDisconnect(endpoint_id,
/* is_safe_to_disconnect */ true,
/* notify_stop_waiting */ true);
MarkEndpointStopWaitToDisconnect(
endpoint_id,
/* is_safe_to_disconnect */ true,
/* notify_stop_waiting */ true);
item->second.disconnect_reason = reason;
auto channel = item->second.channel;
if (channel && !safe_to_disconnect_enabled) {
if (channel && !channel->IsClosed() && !safe_to_disconnect_enabled) {
// If the channel was paused (i.e. during a bandwidth upgrade negotiation)
// we resume to ensure the thread won't hang when trying to write to it.
channel->Resume();
connections/implementation/endpoint_manager_test.cc
+1 -1
View File
@@ -101,7 +101,7 @@ class MockEndpointChannel : public EndpointChannel {
MOCK_METHOD(void, SetAnalyticsRecorder,
(analytics::AnalyticsRecorder*, const std::string&), (override));
bool IsClosed() const {
bool IsClosed() const override {
absl::MutexLock lock(&mutex_);
return closed_;
}
connections/implementation/fake_endpoint_channel.h
+1
View File
@@ -65,6 +65,7 @@ class FakeEndpointChannel : public EndpointChannel {
EstablishedConnection::SafeDisconnectionResult result) override {
Close(reason);
}
bool IsClosed() const override { return is_closed_; }
location::nearby::proto::connections::ConnectionTechnology GetTechnology()
const override {
return location::nearby::proto::connections::ConnectionTechnology::
connections/implementation/mediums/BUILD
+1
View File
@@ -59,6 +59,7 @@ cc_library(
"//internal/platform:types",
"//internal/platform:uuid",
"//internal/platform/implementation:comm",
"//internal/platform/implementation:wifi_utils",
"//proto/mediums:web_rtc_signaling_frames_cc_proto",
# TODO: Support WebRTC
"@com_google_absl//absl/base:core_headers",
connections/implementation/mediums/bluetooth_classic.cc
+1 -1
View File
@@ -401,7 +401,7 @@ bool BluetoothClassic::StartAcceptingConnections(
std::make_shared<BluetoothSocket>(client_socket_bak);
MultiplexSocket* multiplex_socket =
MultiplexSocket::CreateIncomingSocket(physical_socket_ptr,
service_id);
service_id, 0);
if (multiplex_socket != nullptr &&
multiplex_socket->GetVirtualSocket(service_id)) {
connections/implementation/mediums/multiplex/multiplex_output_stream.h
+1 -1
View File
@@ -65,7 +65,7 @@ class MultiplexOutputStream {
MultiplexOutputStream(OutputStream* physical_writer,
AtomicBoolean& is_enabled);
~MultiplexOutputStream() { Shutdown(); }
~MultiplexOutputStream() = default;
// Writes the connection request frame to the physical output stream.
bool WriteConnectionRequestFrame(const std::string& service_id,
connections/implementation/mediums/multiplex/multiplex_socket.cc
+35 -11
View File
@@ -16,6 +16,7 @@
#include <cstdint>
#include <memory>
#include <new>
#include <string>
#include <type_traits>
#include <utility>
@@ -23,10 +24,12 @@
#include "absl/container/flat_hash_map.h"
#include "absl/functional/any_invocable.h"
#include "absl/strings/string_view.h"
#include "absl/time/clock.h"
#include "absl/time/time.h"
#include "connections/implementation/mediums/multiplex/multiplex_frames.h"
#include "connections/implementation/mediums/multiplex/multiplex_output_stream.h"
#include "connections/implementation/mediums/utils.h"
#include "internal/platform/atomic_boolean.h"
#include "internal/platform/base64_utils.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/count_down_latch.h"
@@ -64,6 +67,10 @@ using ::location::nearby::proto::connections::Medium;
using ::location::nearby::proto::connections::Medium_Name;
using ConnectionResponseCode = ConnectionResponseFrame::ConnectionResponseCode;
// AtomicBoolean is trivial destructible, so it is safe to use it as a static
// variable.
AtomicBoolean MultiplexSocket::is_shutting_down_{false}; // NOLINT
void MultiplexSocket::ListenForIncomingConnection(
const std::string& service_id, Medium type,
MultiplexIncomingConnectionCb incoming_connection_cb) {
@@ -105,7 +112,12 @@ MultiplexSocket::GetIncomingConnectionCallbacks() {
MultiplexSocket* MultiplexSocket::CreateIncomingSocket(
std::shared_ptr<MediumSocket> physical_socket,
const std::string& service_id) {
const std::string& service_id, std::int32_t first_frame_len) {
while (is_shutting_down_.Get()) {
NEARBY_LOGS(WARNING)
<< "Shutting down is going on, wait for 2ms to create incoming socket";
absl::SleepFor(absl::Milliseconds(2));
}
static MultiplexSocket* multiplex_incoming_socket = nullptr;
switch (physical_socket->GetMedium()) {
case Medium::BLUETOOTH:
@@ -114,7 +126,6 @@ MultiplexSocket* MultiplexSocket::CreateIncomingSocket(
storage_bt;
multiplex_incoming_socket =
new (&storage_bt) MultiplexSocket(physical_socket);
break;
case Medium::BLE:
static std::aligned_storage_t<sizeof(MultiplexSocket),
@@ -141,7 +152,7 @@ MultiplexSocket* MultiplexSocket::CreateIncomingSocket(
multiplex_incoming_socket->CreateFirstVirtualSocket(service_id,
(std::string)kFakeSalt);
multiplex_incoming_socket->StartReaderThread();
multiplex_incoming_socket->StartReaderThread(first_frame_len);
return multiplex_incoming_socket;
}
@@ -149,6 +160,11 @@ MultiplexSocket* MultiplexSocket::CreateIncomingSocket(
MultiplexSocket* MultiplexSocket::CreateOutgoingSocket(
std::shared_ptr<MediumSocket> physical_socket,
const std::string& service_id, const std::string& service_id_hash_salt) {
while (is_shutting_down_.Get()) {
NEARBY_LOGS(WARNING)
<< "Shutting down is going on, wait for 2ms to create outgoing socket";
absl::SleepFor(absl::Milliseconds(2));
}
static MultiplexSocket* multiplex_outgoing_socket = nullptr;
switch (physical_socket->GetMedium()) {
case Medium::BLUETOOTH:
@@ -182,7 +198,7 @@ MultiplexSocket* MultiplexSocket::CreateOutgoingSocket(
multiplex_outgoing_socket->CreateFirstVirtualSocket(service_id,
service_id_hash_salt);
multiplex_outgoing_socket->StartReaderThread();
multiplex_outgoing_socket->StartReaderThread(0);
return multiplex_outgoing_socket;
}
@@ -334,20 +350,26 @@ MediumSocket* MultiplexSocket::EstablishVirtualSocket(
return nullptr;
}
void MultiplexSocket::StartReaderThread() {
void MultiplexSocket::StartReaderThread(std::int32_t first_frame_len) {
if (is_shutdown_) {
NEARBY_LOGS(WARNING) << "Stop to start reader thread since socket is "
"shutdown.";
return;
}
reader_thread_shutdown_barrier_ = std::make_unique<CountDownLatch>(1);
physical_reader_thread_.Execute([this]() {
physical_reader_thread_.Execute([this, first_frame_len]() {
NEARBY_LOGS(INFO) << __func__ << " Reader thread starts.";
auto first_frame_len_copy = first_frame_len;
while (!is_shutdown_) {
bool fail = false;
ExceptionOr<ByteArray> bytes;
ExceptionOr<std::int32_t> read_int =
Base64Utils::ReadInt(physical_reader_);
ExceptionOr<std::int32_t> read_int;
if (first_frame_len_copy > 0) {
read_int = ExceptionOr<std::int32_t>(first_frame_len);
first_frame_len_copy = 0;
} else {
read_int = Base64Utils::ReadInt(physical_reader_);
}
if (!read_int.ok()) {
NEARBY_LOGS(WARNING)
<< __func__ << "Failed to read. Exception:" << read_int.exception();
@@ -650,6 +672,7 @@ void MultiplexSocket::OnVirtualSocketClosed(const std::string& service_id) {
if (virtual_sockets_.empty()) {
NEARBY_LOGS(INFO) << "Close the physical socket because all virtual "
"sockets disconnected.";
is_shutting_down_.Set(true);
Shutdown();
shutdown = true;
}
@@ -669,6 +692,7 @@ void MultiplexSocket::OnVirtualSocketClosed(const std::string& service_id) {
<< "Shutdown single_thread_offloader_ and physical_reader_thread_";
single_thread_offloader_.Shutdown();
physical_reader_thread_.Shutdown();
is_shutting_down_.Set(false);
}
}
@@ -761,9 +785,9 @@ void MultiplexSocket::ShutdownAll() {
if (!latch
.Await(FeatureFlags::GetInstance()
.GetFlags()
.mediums_frame_write_timeout_millis)
.result() +
200) {
.mediums_frame_write_timeout_millis +
absl::Milliseconds(100))
.result()) {
NEARBY_LOGS(ERROR) << "Timeout to close virtual socket";
}
connections/implementation/mediums/multiplex/multiplex_socket.h
+6 -3
View File
@@ -15,6 +15,7 @@
#ifndef CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_SOCKET_H_
#define CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_SOCKET_H_
#include <cstdint>
#include <memory>
#include <string>
#include <utility>
@@ -48,11 +49,12 @@ class MultiplexSocket {
public:
MultiplexSocket(const MultiplexSocket&) = delete;
MultiplexSocket& operator=(const MultiplexSocket&) = delete;
~MultiplexSocket() { ShutdownAll(); };
// Creates a new incoming MultiplexSocket.
static MultiplexSocket* CreateIncomingSocket(
std::shared_ptr<MediumSocket> physical_socket,
const std::string& service_id);
const std::string& service_id, std::int32_t first_frame_len);
// Creates a new outgoing MultiplexSocket.
static MultiplexSocket* CreateOutgoingSocket(
std::shared_ptr<MediumSocket> physical_socket,
@@ -111,7 +113,6 @@ class MultiplexSocket {
private:
explicit MultiplexSocket(std::shared_ptr<MediumSocket> physical_socket);
~MultiplexSocket() { ShutdownAll(); };
// Creates the first virtual socket for the service id. The first virtual
// socket is created by the sender.
@@ -128,7 +129,7 @@ class MultiplexSocket {
void UnRegisterConnectionResponse(const std::string& service_id);
// Starts the reader thread to read the incoming MultiplexFrame from the
// physical socket.
void StartReaderThread();
void StartReaderThread(std::int32_t first_frame_len);
// Handles the offline frame from the physical socket.
void HandleOfflineFrame(const ByteArray& bytes);
// Handles the control frame from the physical socket.
@@ -206,8 +207,10 @@ class MultiplexSocket {
// enable it once two devices negotiated finished.
AtomicBoolean enabled_{false};
std::int32_t first_frame_len_ = 0;
// If the socket is already shutdown and no longer in use.
bool is_shutdown_ = false;
static AtomicBoolean is_shutting_down_;
std::unique_ptr<CountDownLatch> reader_thread_shutdown_barrier_;
};
connections/implementation/mediums/multiplex/multiplex_socket_test.cc
+4 -4
View File
@@ -160,8 +160,8 @@ TEST(MultiplexSocketTest, CreateSuccessAndReaderThreadStarted) {
MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_1),
Medium::BLUETOOTH);
MultiplexSocket* multiplex_socket_incoming =
MultiplexSocket::CreateIncomingSocket(fake_socket_ptr,
std::string(SERVICE_ID_1));
MultiplexSocket::CreateIncomingSocket(
fake_socket_ptr, std::string(SERVICE_ID_1), /*first_frame_len*/ 0);
ASSERT_NE(multiplex_socket_incoming, nullptr);
FakeSocket* virtual_socket =
(FakeSocket*)multiplex_socket_incoming->GetVirtualSocket(
@@ -209,8 +209,8 @@ TEST(MultiplexSocketTest, CreateFail_MediumNotSupport) {
MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_1),
Medium::WEB_RTC);
MultiplexSocket* multiplex_socket_incoming =
MultiplexSocket::CreateIncomingSocket(fake_socket_ptr,
std::string(SERVICE_ID_1));
MultiplexSocket::CreateIncomingSocket(
fake_socket_ptr, std::string(SERVICE_ID_1), /*first_frame_len*/ 0);
ASSERT_EQ(multiplex_socket_incoming, nullptr);
}
connections/implementation/mediums/wifi_lan.cc
+193 -5
View File
@@ -15,22 +15,34 @@
#include "connections/implementation/mediums/wifi_lan.h"
#include <cstdint>
#include <memory>
#include <string>
#include <utility>
#include "absl/strings/str_cat.h"
#include "absl/strings/str_format.h"
#include "connections/implementation/mediums/multiplex/multiplex_socket.h"
#include "connections/implementation/mediums/utils.h"
#include "connections/medium_selector.h"
#include "internal/platform/base64_utils.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/cancellation_flag.h"
#include "internal/platform/exception.h"
#include "internal/platform/implementation/wifi_utils.h"
#include "internal/platform/logging.h"
#include "internal/platform/mutex_lock.h"
#include "internal/platform/nsd_service_info.h"
#include "internal/platform/socket.h"
#include "internal/platform/types.h"
#include "internal/platform/wifi_lan.h"
namespace nearby {
namespace connections {
namespace {
using MultiplexSocket = mediums::multiplex::MultiplexSocket;
} // namespace
WifiLan::~WifiLan() {
// Destructor is not taking locks, but methods it is calling are.
while (!discovering_info_.service_ids.empty()) {
@@ -42,7 +54,18 @@ WifiLan::~WifiLan() {
while (!advertising_info_.nsd_service_infos.empty()) {
StopAdvertising(advertising_info_.nsd_service_infos.begin()->first);
}
{
MutexLock lock(&mutex_);
if (is_multiplex_enabled_) {
NEARBY_LOGS(INFO) << "Closing multiplex sockets for "
<< multiplex_sockets_.size() << " IPs";
for (auto& [ip_addr, multiplex_socket] : multiplex_sockets_) {
NEARBY_LOGS(INFO) << "Closing multiplex sockets for: " << ip_addr;
multiplex_socket->~MultiplexSocket();
}
multiplex_sockets_.clear();
}
}
// All the AcceptLoopRunnable objects in here should already have gotten an
// opportunity to shut themselves down cleanly in the calls to
// StopAcceptingConnections() above.
@@ -256,20 +279,82 @@ bool WifiLan::StartAcceptingConnections(const std::string& service_id,
server_sockets_.insert({service_id, std::move(server_socket)})
.first->second;
// Register the callback to listen for incoming multiplex virtual socket.
if (is_multiplex_enabled_) {
MultiplexSocket::ListenForIncomingConnection(
service_id, Medium::WIFI_LAN,
[&callback](const std::string& listening_service_id,
MediumSocket* virtual_socket) mutable {
if (callback) {
callback(listening_service_id,
*(down_cast<WifiLanSocket*>(virtual_socket)));
}
});
}
// Start the accept loop on a dedicated thread - this stays alive and
// listening for new incoming connections until StopAcceptingConnections() is
// invoked.
accept_loops_runner_.Execute(
"wifi-lan-accept",
[callback = std::move(callback),
server_socket = std::move(owned_server_socket), service_id]() mutable {
"wifi-lan-accept", [callback = std::move(callback),
server_socket = std::move(owned_server_socket),
service_id, this]() mutable {
while (true) {
WifiLanSocket client_socket = server_socket.Accept();
if (!client_socket.IsValid()) {
server_socket.Close();
break;
}
if (callback) {
NEARBY_LOGS(INFO) << "Accepted connection for " << service_id;
bool callback_called = false;
{
MutexLock lock(&mutex_);
if (is_multiplex_enabled_) {
// Observed from the log that when the sender tries to connect to
// the receiver's server socket, the server side will somehow
// receive 3 connection request events(don’t know what’s happening
// in Windows’s lower layer code). The 2nd normally is the real
// one. The other two will result in a failed data receiving in
// Windows platform layer. To avoid creating multiplex
// IncomingSocket, we will check if the first read is successful
// or not. If not, discard it. If yes, save that packet
// content(the first frame length), then create the multiplex
// socket, then feed that content to that multiplex socket.
ExceptionOr<std::int32_t> read_int =
Base64Utils::ReadInt(&client_socket.GetInputStream());
if (!read_int.ok()) {
NEARBY_LOGS(WARNING)
<< __func__
<< "Failed to read. Exception:" << read_int.exception()
<< "Discard the connection.";
continue;
}
WifiLanSocket client_socket_bak = client_socket;
auto physical_socket_ptr =
std::make_shared<WifiLanSocket>(client_socket_bak);
MultiplexSocket* multiplex_socket =
MultiplexSocket::CreateIncomingSocket(
physical_socket_ptr, service_id, read_int.result());
if (multiplex_socket != nullptr &&
multiplex_socket->GetVirtualSocket(service_id)) {
multiplex_sockets_.emplace(server_socket.GetIPAddress(),
multiplex_socket);
MultiplexSocket::StopListeningForIncomingConnection(
service_id, Medium::WIFI_LAN);
NEARBY_LOGS(INFO) << "Multiplex virtaul socket created for "
<< server_socket.GetIPAddress();
if (callback) {
callback(
service_id,
*(down_cast<WifiLanSocket*>(
multiplex_socket->GetVirtualSocket(service_id))));
callback_called = true;
}
}
}
}
if (callback && !callback_called) {
NEARBY_LOGS(INFO) << "Call back triggered for physical socket.";
callback(service_id, std::move(client_socket));
}
}
@@ -293,6 +378,10 @@ bool WifiLan::StopAcceptingConnections(const std::string& service_id) {
<< service_id << " because it was never started.";
return false;
}
if (is_multiplex_enabled_) {
MultiplexSocket::StopListeningForIncomingConnection(service_id,
Medium::WIFI_LAN);
}
// Closing the WifiLanServerSocket will kick off the suicide of the thread
// in accept_loops_thread_pool_ that blocks on WifiLanServerSocket.accept().
@@ -352,12 +441,31 @@ WifiLanSocket WifiLan::Connect(const std::string& service_id,
return socket;
}
ExceptionOr<WifiLanSocket> virtual_socket =
ConnectWithMultiplexSocketLocked(service_id, service_info.GetIPAddress());
if (virtual_socket.ok()) {
return virtual_socket.result();
}
socket = medium_.ConnectToService(service_info, cancellation_flag);
if (!socket.IsValid()) {
NEARBY_LOGS(INFO) << "Failed to Connect via WifiLan [service_id="
<< service_id << "]";
return socket;
} else {
ExceptionOr<WifiLanSocket> virtual_socket =
CreateOutgoingMultiplexSocketLocked(socket, service_id,
service_info.GetIPAddress());
if (virtual_socket.ok()) {
NEARBY_LOGS(INFO)
<< "Successfully connected via Multiplex WifiLan [service_id="
<< service_id << "]";
return virtual_socket.result();
}
}
NEARBY_LOGS(INFO) << "Successfully connected via WifiLan [service_id="
<< service_id << "]";
return socket;
}
@@ -385,15 +493,95 @@ WifiLanSocket WifiLan::Connect(const std::string& service_id,
return socket;
}
ExceptionOr<WifiLanSocket> virtual_socket =
ConnectWithMultiplexSocketLocked(service_id, ip_address);
if (virtual_socket.ok()) {
return virtual_socket.result();
}
socket = medium_.ConnectToService(ip_address, port, cancellation_flag);
if (!socket.IsValid()) {
NEARBY_LOGS(INFO) << "Failed to Connect via WifiLan [service_id="
<< service_id << "]";
return socket;
} else {
ExceptionOr<WifiLanSocket> virtual_socket =
CreateOutgoingMultiplexSocketLocked(socket, service_id, ip_address);
if (virtual_socket.ok()) {
NEARBY_LOGS(INFO)
<< "Successfully connected via Multiplex WifiLan [service_id="
<< service_id << "]";
return virtual_socket.result();
}
}
NEARBY_LOGS(INFO) << "Successfully connected via WifiLan [service_id="
<< service_id << "]";
return socket;
}
ExceptionOr<WifiLanSocket> WifiLan::ConnectWithMultiplexSocketLocked(
const std::string& service_id, const std::string& ip_address) {
if (is_multiplex_enabled_) {
NEARBY_LOGS(INFO) << "multiplex_sockets_ size:"
<< multiplex_sockets_.size();
auto it = multiplex_sockets_.find(ip_address);
if (it != multiplex_sockets_.end()) {
MultiplexSocket* multiplex_socket = it->second;
if (multiplex_socket->IsShutdown()) {
NEARBY_LOGS(INFO)
<< "Erase multiplex_socket(already shutdown) for ip_address: "
<< WifiUtils::GetHumanReadableIpAddress(ip_address);
multiplex_socket->~MultiplexSocket();
multiplex_sockets_.erase(it);
return ExceptionOr<WifiLanSocket>(Exception::kFailed);
}
if (multiplex_socket->IsEnabled()) {
auto* virtual_socket =
multiplex_socket->EstablishVirtualSocket(service_id);
// Should not happen.
auto* wlan_socket = down_cast<WifiLanSocket*>(virtual_socket);
if (wlan_socket == nullptr) {
NEARBY_LOGS(INFO) << "Failed to cast to WifiLanSocket for "
<< service_id << " with ip_address: "
<< WifiUtils::GetHumanReadableIpAddress(ip_address);
return ExceptionOr<WifiLanSocket>(Exception::kFailed);
}
return ExceptionOr<WifiLanSocket>(*wlan_socket);
}
}
}
return ExceptionOr<WifiLanSocket>(Exception::kFailed);
}
ExceptionOr<WifiLanSocket> WifiLan::CreateOutgoingMultiplexSocketLocked(
WifiLanSocket& socket, const std::string& service_id,
const std::string& ip_address) {
if (is_multiplex_enabled_) {
// Create MultiplexSocket, but set it to be disabled as default. It will be
// enabled if both side support multiplex for WIFI_LAN
auto physical_socket_ptr = std::make_shared<WifiLanSocket>(socket);
MultiplexSocket* multiplex_socket =
MultiplexSocket::CreateOutgoingSocket(physical_socket_ptr, service_id);
auto* virtual_socket = multiplex_socket->GetVirtualSocket(service_id);
// Should not happen.
auto* wlan_socket = down_cast<WifiLanSocket*>(virtual_socket);
if (wlan_socket == nullptr) {
NEARBY_LOGS(INFO) << "Failed to cast to WifiLanSocket for " << service_id
<< " with ip_address: "
<< WifiUtils::GetHumanReadableIpAddress(ip_address);
return ExceptionOr<WifiLanSocket>(Exception::kFailed);
}
NEARBY_LOGS(INFO) << "Multiplex socket created for ip_address: "
<< WifiUtils::GetHumanReadableIpAddress(ip_address);
multiplex_sockets_.emplace(ip_address,
multiplex_socket);
return ExceptionOr<WifiLanSocket>(*wlan_socket);
}
return ExceptionOr<WifiLanSocket>(Exception::kFailed);
}
std::pair<std::string, int> WifiLan::GetCredentials(
const std::string& service_id) {
MutexLock lock(&mutex_);
connections/implementation/mediums/wifi_lan.h
+29 -2
View File
@@ -16,13 +16,18 @@
#define CORE_INTERNAL_MEDIUMS_WIFI_LAN_H_
#include <cstdint>
#include <functional>
#include <string>
#include <utility>
#include "absl/base/thread_annotations.h"
#include "absl/container/flat_hash_map.h"
#include "absl/container/flat_hash_set.h"
#include "internal/platform/byte_array.h"
#include "absl/functional/any_invocable.h"
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "connections/implementation/mediums/multiplex/multiplex_socket.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/cancellation_flag.h"
#include "internal/platform/exception.h"
#include "internal/platform/multi_thread_executor.h"
#include "internal/platform/mutex.h"
#include "internal/platform/nsd_service_info.h"
@@ -150,6 +155,18 @@ class WifiLan {
static constexpr int kMaxConcurrentAcceptLoops = 5;
// Establishes connection to WifiLan service by ip address through
// MultiplexSocket.
ExceptionOr<WifiLanSocket> ConnectWithMultiplexSocketLocked(
const std::string& service_id, const std::string& ip_address)
ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
// Creates a MultiplexSocket for outgoing connection based on connected
// WifiLanSocket physical socket for specific service_id and ip address.
ExceptionOr<WifiLanSocket> CreateOutgoingMultiplexSocketLocked(
WifiLanSocket& socket, const std::string& service_id,
const std::string& ip_address) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
// Same as IsAvailable(), but must be called with mutex_ held.
bool IsAvailableLocked() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
@@ -187,6 +204,16 @@ class WifiLan {
// and thus require pointer stability.
absl::flat_hash_map<std::string, WifiLanServerSocket> server_sockets_
ABSL_GUARDED_BY(mutex_);
// Whether the multiplex feature is enabled.
bool is_multiplex_enabled_ = NearbyFlags::GetInstance().GetBoolFlag(
config_package_nearby::nearby_connections_feature::
kEnableMultiplex);
// A map of IpAddress -> MultiplexSocket.
absl::flat_hash_map<std::string,
mediums::multiplex::MultiplexSocket*>
multiplex_sockets_ ABSL_GUARDED_BY(mutex_);
};
} // namespace connections
connections/implementation/mediums/wifi_lan_test.cc
+71
View File
@@ -15,16 +15,22 @@
#include "connections/implementation/mediums/wifi_lan.h"
#include <string>
#include <utility>
#include "gmock/gmock.h"
#include "protobuf-matchers/protocol-buffer-matchers.h"
#include "gtest/gtest.h"
#include "absl/strings/string_view.h"
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/cancellation_flag.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/logging.h"
#include "internal/platform/medium_environment.h"
#include "internal/platform/nsd_service_info.h"
#include "internal/platform/single_thread_executor.h"
#include "internal/platform/wifi_lan.h"
#include "internal/platform/base64_utils.h"
namespace nearby {
namespace connections {
@@ -106,6 +112,71 @@ TEST_P(WifiLanTest, CanConnect) {
env_.Stop();
}
TEST_P(WifiLanTest, CanConnectWithMultiplex) {
bool is_multiplex_enabled = NearbyFlags::GetInstance().GetBoolFlag(
config_package_nearby::nearby_connections_feature::kEnableMultiplex);
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::kEnableMultiplex,
true);
FeatureFlags feature_flags = GetParam();
env_.SetFeatureFlags(feature_flags);
env_.Start();
WifiLan wifi_lan_client;
WifiLan wifi_lan_server;
std::string service_id(kServiceID);
std::string service_info_name(kServiceInfoName);
std::string endpoint_info_name(kEndpointName);
CountDownLatch discovered_latch(1);
CountDownLatch accept_latch(1);
WifiLanSocket socket_for_server;
EXPECT_TRUE(wifi_lan_server.StartAcceptingConnections(
service_id, [&](const std::string& service_id, WifiLanSocket socket) {
socket_for_server = std::move(socket);
accept_latch.CountDown();
}));
NsdServiceInfo nsd_service_info;
nsd_service_info.SetServiceName(service_info_name);
nsd_service_info.SetTxtRecord(std::string(kEndpointInfoKey),
endpoint_info_name);
wifi_lan_server.StartAdvertising(service_id, nsd_service_info);
WifiLanSocket socket_for_client;
SingleThreadExecutor client_executor;
client_executor.Execute([&]() {
NsdServiceInfo discovered_service_info;
wifi_lan_client.StartDiscovery(
service_id,
{
.service_discovered_cb =
[&discovered_latch, &discovered_service_info](
NsdServiceInfo service_info, const std::string& service_id) {
NEARBY_LOGS(INFO)
<< "Discovered service_info=" << &service_info;
discovered_service_info = service_info;
discovered_latch.CountDown();
},
});
discovered_latch.Await(kWaitDuration).result();
ASSERT_TRUE(discovered_service_info.IsValid());
CancellationFlag flag;
socket_for_client =
wifi_lan_client.Connect(service_id, discovered_service_info, &flag);
Base64Utils::WriteInt(&socket_for_client.GetOutputStream(), 4);
});
EXPECT_TRUE(accept_latch.Await(kWaitDuration).result());
EXPECT_TRUE(wifi_lan_server.StopAcceptingConnections(service_id));
EXPECT_TRUE(wifi_lan_server.StopAdvertising(service_id));
EXPECT_TRUE(socket_for_server.IsValid());
EXPECT_TRUE(socket_for_client.IsValid());
env_.Stop();
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::kEnableMultiplex,
is_multiplex_enabled);
}
TEST_P(WifiLanTest, CanCancelConnect) {
FeatureFlags feature_flags = GetParam();
env_.SetFeatureFlags(feature_flags);
connections/implementation/p2p_cluster_pcp_handler_test.cc
+15 -8
View File
@@ -1057,14 +1057,21 @@ TEST_P(P2pClusterPcpHandlerTestWithParam, CanConnect) {
EXPECT_EQ(client_b_.GetApFrequency(discovered.endpoint_id), kFreq);
EXPECT_EQ(client_b_.GetIPAddress(discovered.endpoint_id),
std::string(kIp4Bytes));
EXPECT_EQ(client_a_.Is5GHzSupported(client_b_local_endpoint),
mediums_b.GetWifi().GetCapability().supports_5_ghz);
EXPECT_EQ(client_a_.GetBssid(client_b_local_endpoint),
mediums_b.GetWifi().GetInformation().bssid);
EXPECT_EQ(client_a_.GetApFrequency(client_b_local_endpoint),
mediums_b.GetWifi().GetInformation().ap_frequency);
EXPECT_EQ(client_a_.GetIPAddress(client_b_local_endpoint),
mediums_b.GetWifi().GetInformation().ip_address_4_bytes);
// When connection is established, EndpointManager will setup KeepAliveManager
// loop. When it fails, the connection will be dismantled. Since this a unit
// test, KeepAliveManager won't be really up. The disconnection may happen
// before the following check, which cause the check fail. So we check the
// connection status first.
if (client_b_.IsConnectedToEndpoint(discovered.endpoint_id)) {
EXPECT_EQ(client_a_.Is5GHzSupported(client_b_local_endpoint),
mediums_b.GetWifi().GetCapability().supports_5_ghz);
EXPECT_EQ(client_a_.GetBssid(client_b_local_endpoint),
mediums_b.GetWifi().GetInformation().bssid);
EXPECT_EQ(client_a_.GetApFrequency(client_b_local_endpoint),
mediums_b.GetWifi().GetInformation().ap_frequency);
EXPECT_EQ(client_a_.GetIPAddress(client_b_local_endpoint),
mediums_b.GetWifi().GetInformation().ip_address_4_bytes);
}
handler_b.StopDiscovery(&client_b_);
bwu_a.Shutdown();
connections/implementation/p2p_point_to_point_pcp_handler_test.cc
+15 -8
View File
@@ -237,14 +237,21 @@ TEST_P(P2pPointToPointPcpHandlerTest, CanConnect) {
EXPECT_EQ(client_b_.GetApFrequency(discovered.endpoint_id), kFreq);
EXPECT_EQ(client_b_.GetIPAddress(discovered.endpoint_id),
std::string(kIp4Bytes));
EXPECT_EQ(client_a_.Is5GHzSupported(client_b_local_endpoint),
mediums_b.GetWifi().GetCapability().supports_5_ghz);
EXPECT_EQ(client_a_.GetBssid(client_b_local_endpoint),
mediums_b.GetWifi().GetInformation().bssid);
EXPECT_EQ(client_a_.GetApFrequency(client_b_local_endpoint),
mediums_b.GetWifi().GetInformation().ap_frequency);
EXPECT_EQ(client_a_.GetIPAddress(client_b_local_endpoint),
mediums_b.GetWifi().GetInformation().ip_address_4_bytes);
// When connection is established, EndpointManager will setup KeepAliveManager
// loop. When it fails, the connection will be dismantled. Since this a unit
// test, KeepAliveManager won't be really up. The disconnection may happen
// before the following check, which cause the check fail. So we check the
// connection status first.
if (client_b_.IsConnectedToEndpoint(discovered.endpoint_id)) {
EXPECT_EQ(client_a_.Is5GHzSupported(client_b_local_endpoint),
mediums_b.GetWifi().GetCapability().supports_5_ghz);
EXPECT_EQ(client_a_.GetBssid(client_b_local_endpoint),
mediums_b.GetWifi().GetInformation().bssid);
EXPECT_EQ(client_a_.GetApFrequency(client_b_local_endpoint),
mediums_b.GetWifi().GetInformation().ap_frequency);
EXPECT_EQ(client_a_.GetIPAddress(client_b_local_endpoint),
mediums_b.GetWifi().GetInformation().ip_address_4_bytes);
}
handler_b.StopDiscovery(&client_b_);
bwu_a.Shutdown();
connections/implementation/wifi_lan_endpoint_channel.cc
+8
View File
@@ -16,6 +16,7 @@
#include <string>
#include "connections/implementation/base_endpoint_channel.h"
#include "internal/platform/logging.h"
#include "internal/platform/wifi_lan.h"
@@ -42,5 +43,12 @@ void WifiLanEndpointChannel::CloseImpl() {
}
}
bool WifiLanEndpointChannel::EnableMultiplexSocket() {
NEARBY_LOGS(INFO) << "WifiLanEndpointChannel MultiplexSocket will be "
"enabled if the WifiLan MultiplexSocket is valid";
socket_.EnableMultiplexSocket();
return true;
}
} // namespace connections
} // namespace nearby
connections/implementation/wifi_lan_endpoint_channel.h
+1
View File
@@ -30,6 +30,7 @@ class WifiLanEndpointChannel final : public BaseEndpointChannel {
const std::string& channel_name, WifiLanSocket socket);
location::nearby::proto::connections::Medium GetMedium() const override;
bool EnableMultiplexSocket() override;
private:
void CloseImpl() override;
internal/platform/bluetooth_classic.cc
-10
View File
@@ -31,16 +31,6 @@
namespace nearby {
using location::nearby::proto::connections::Medium;
MediumSocket* BluetoothSocket::CreateVirtualSocket(OutputStream* outputstream) {
if (IsVirtualSocket()) {
LOG(WARNING)
<< "Creating the virtual socket on a virtual socket is not allowed.";
return nullptr;
}
auto virtual_socket = std::make_shared<BluetoothSocket>(outputstream);
return virtual_socket.get();
}
MediumSocket* BluetoothSocket::CreateVirtualSocket(
const std::string& salted_service_id_hash_key, OutputStream* outputstream,
Medium medium,
internal/platform/bluetooth_classic.h
+1 -2
View File
@@ -96,8 +96,7 @@ class BluetoothSocket : public MediumSocket {
// Returns true if this is a virtual socket.
bool IsVirtualSocket() override { return is_virtual_socket_; }
// Creates a virtual socket only with outputstream.
MediumSocket* CreateVirtualSocket(OutputStream* outputstream) override;
// Creates a virtual socket.
MediumSocket* CreateVirtualSocket(
const std::string& salted_service_id_hash_key, OutputStream* outputstream,
location::nearby::proto::connections::Medium medium,
internal/platform/socket.h
-5
View File
@@ -53,11 +53,6 @@ class MediumSocket : public Socket {
return medium_;
}
/** Creates a virtual socket only with outputstream. */
virtual MediumSocket* CreateVirtualSocket(OutputStream* outputstream) {
return this;
}
/** Creates a virtual socket. */
virtual MediumSocket* CreateVirtualSocket(
const std::string& salted_service_id_hash_key, OutputStream* outputstream,
internal/platform/wifi_lan.cc
+36 -2
View File
@@ -14,13 +14,47 @@
#include "internal/platform/wifi_lan.h"
#include <memory>
#include <string>
#include <utility>
#include "internal/platform/implementation/wifi_utils.h"
#include "absl/container/flat_hash_map.h"
#include "internal/platform/cancellation_flag.h"
#include "internal/platform/logging.h"
#include "internal/platform/mutex_lock.h"
#include "internal/platform/nsd_service_info.h"
#include "internal/platform/output_stream.h"
#include "internal/platform/socket.h"
#include "internal/platform/implementation/wifi_utils.h"
namespace nearby {
using location::nearby::proto::connections::Medium;
MediumSocket* WifiLanSocket::CreateVirtualSocket(
const std::string& salted_service_id_hash_key, OutputStream* outputstream,
Medium medium,
absl::flat_hash_map<std::string, std::shared_ptr<MediumSocket>>*
virtual_sockets_ptr) {
if (IsVirtualSocket()) {
NEARBY_LOGS(WARNING)
<< "Creating the virtual socket on a virtual socket is not allowed.";
return nullptr;
}
auto virtual_socket = std::make_shared<WifiLanSocket>(outputstream);
virtual_socket->impl_ = this->impl_;
NEARBY_LOGS(WARNING) << "Created the virtual socket for Medium: "
<< Medium_Name(virtual_socket->GetMedium());
if (virtual_sockets_ptr_ == nullptr) {
virtual_sockets_ptr_ = virtual_sockets_ptr;
}
(*virtual_sockets_ptr_)[salted_service_id_hash_key] = virtual_socket;
NEARBY_LOGS(INFO) << "virtual_sockets_ size: "
<< virtual_sockets_ptr_->size();
return virtual_socket.get();
}
bool WifiLanMedium::StartAdvertising(const NsdServiceInfo& nsd_service_info) {
return impl_->StartAdvertising(nsd_service_info);
@@ -117,7 +151,7 @@ bool WifiLanMedium::StartDiscovery(const std::string& service_id,
// Insert callback to the map first no matter it succeeds or not.
MutexLock lock(&mutex_);
auto pair = service_type_to_callback_map_.insert(
{service_type, absl::make_unique<DiscoveryCallbackInfo>()});
{service_type, std::make_unique<DiscoveryCallbackInfo>()});
auto& context = *pair.first->second;
context.medium_callback = std::move(callback);
context.service_id = service_id;
internal/platform/wifi_lan.h
+69 -8
View File
@@ -15,47 +15,100 @@
#ifndef PLATFORM_PUBLIC_WIFI_LAN_H_
#define PLATFORM_PUBLIC_WIFI_LAN_H_
#include <cstdint>
#include <memory>
#include <string>
#include <utility>
#include "absl/container/flat_hash_map.h"
#include "absl/functional/any_invocable.h"
#include "absl/types/optional.h"
#include "internal/platform/blocking_queue_stream.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/cancellation_flag.h"
#include "internal/platform/exception.h"
#include "internal/platform/implementation/platform.h"
#include "internal/platform/implementation/wifi_lan.h"
#include "internal/platform/input_stream.h"
#include "internal/platform/listeners.h"
#include "internal/platform/logging.h"
#include "internal/platform/mutex.h"
#include "internal/platform/nsd_service_info.h"
#include "internal/platform/output_stream.h"
#include "internal/platform/socket.h"
namespace nearby {
class WifiLanSocket final {
class WifiLanSocket : public MediumSocket {
public:
WifiLanSocket() = default;
WifiLanSocket()
: MediumSocket(location::nearby::proto::connections::Medium::WIFI_LAN) {}
WifiLanSocket(const WifiLanSocket&) = default;
WifiLanSocket& operator=(const WifiLanSocket&) = default;
~WifiLanSocket() = default;
~WifiLanSocket() override = default;
// Creates a physical WifiLanSocket from a platform implementation.
explicit WifiLanSocket(std::unique_ptr<api::WifiLanSocket> socket)
: impl_(std::move(socket)) {}
: MediumSocket(location::nearby::proto::connections::Medium::WIFI_LAN),
impl_(socket.release()) {}
// Creates a virtual WifiLanSocket from a virtual output stream.
explicit WifiLanSocket(OutputStream* virtual_output_stream)
: MediumSocket(location::nearby::proto::connections::Medium::WIFI_LAN),
blocking_queue_input_stream_(std::make_shared<BlockingQueueStream>()),
virtual_output_stream_(virtual_output_stream),
is_virtual_socket_(true) {}
// Returns the InputStream of the WifiLanSocket.
// On error, returned stream will report Exception::kIo on any operation.
//
// The returned object is not owned by the caller, and can be invalidated once
// the WifiLanSocket object is destroyed.
InputStream& GetInputStream() { return impl_->GetInputStream(); }
InputStream& GetInputStream() override {
return IsVirtualSocket() ? *blocking_queue_input_stream_
: impl_->GetInputStream();
}
// Returns the OutputStream of the WifiLanSocket.
// On error, returned stream will report Exception::kIo on any operation.
//
// The returned object is not owned by the caller, and can be invalidated once
// the WifiLanSocket object is destroyed.
OutputStream& GetOutputStream() { return impl_->GetOutputStream(); }
OutputStream& GetOutputStream() override {
return IsVirtualSocket() ? *virtual_output_stream_
: impl_->GetOutputStream();
}
// Returns Exception::kIo on error, Exception::kSuccess otherwise.
Exception Close() { return impl_->Close(); }
Exception Close() override {
if (IsVirtualSocket()) {
NEARBY_LOGS(INFO) << "Multiplex: Closing virtual socket: " << this;
blocking_queue_input_stream_->Close();
virtual_output_stream_->Close();
CloseLocal(); // This will trigger MultiplexSocket::OnVirtualSocketClosed
return {Exception::kSuccess};
}
return impl_->Close();
}
// Returns true if this is a virtual socket.
bool IsVirtualSocket() override { return is_virtual_socket_; }
// Creates a virtual socket only with outputstream.
MediumSocket* CreateVirtualSocket(
const std::string& salted_service_id_hash_key, OutputStream* outputstream,
location::nearby::proto::connections::Medium medium,
absl::flat_hash_map<std::string, std::shared_ptr<MediumSocket>>*
virtual_sockets_ptr) override;
/** Feeds the received incoming data to the client. */
void FeedIncomingData(ByteArray data) override {
if (!IsVirtualSocket()) {
NEARBY_LOGS(INFO) << "Feeding data on a physical socket is not allowed.";
return;
}
blocking_queue_input_stream_->Write(data);
}
// Returns true if a socket is usable. If this method returns false,
// it is not safe to call any other method.
@@ -66,7 +119,10 @@ class WifiLanSocket final {
// an object returned by WifiLanMedium::Connect
// These methods may also return an invalid socket if connection failed for
// any reason.
bool IsValid() const { return impl_ != nullptr; }
bool IsValid() const {
if (is_virtual_socket_) return true;
return impl_ != nullptr;
}
// Returns reference to platform implementation.
// This is used to communicate with platform code, and for debugging purposes.
@@ -77,6 +133,11 @@ class WifiLanSocket final {
private:
std::shared_ptr<api::WifiLanSocket> impl_;
absl::flat_hash_map<std::string, std::shared_ptr<MediumSocket>>*
virtual_sockets_ptr_ = nullptr;
std::shared_ptr<BlockingQueueStream> blocking_queue_input_stream_ = nullptr;
OutputStream* virtual_output_stream_ = nullptr;
bool is_virtual_socket_ = false;
};
class WifiLanServerSocket final {