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

Roll forward to cl/338725629

Browse Source 
Signed-off-by: hai007 <hais@google.com>
This commit is contained in:
hai007
2020-10-23 12:43:22 -07:00
parent 2155b3ddeb
commit d3b3d07805
29 changed files with 478 additions and 1562 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/core/core.h
+3
View File
@@ -215,6 +215,9 @@ class Core {
void InitiateBandwidthUpgrade(absl::string_view endpoint_id,
ResultCallback callback);
// Gets the local endpoint generated by Nearby Connections.
std::string GetLocalEndpointId() { return client_.GetLocalEndpointId(); }
private:
static constexpr absl::Duration kWaitForDisconnect = absl::Milliseconds(5000);
cpp/core/internal/BUILD
+4
View File
@@ -5,6 +5,7 @@ cc_library(
"base_pcp_handler.cc",
"ble_advertisement.cc",
"ble_endpoint_channel.cc",
"bluetooth_bwu_handler.cc",
"bluetooth_device_name.cc",
"bluetooth_endpoint_channel.cc",
"bwu_manager.cc",
@@ -24,6 +25,7 @@ cc_library(
"service_controller_router.cc",
"webrtc_bwu_handler.cc",
"webrtc_endpoint_channel.cc",
"wifi_lan_bwu_handler.cc",
"wifi_lan_endpoint_channel.cc",
"wifi_lan_service_info.cc",
],
@@ -33,6 +35,7 @@ cc_library(
"base_pcp_handler.h",
"ble_advertisement.h",
"ble_endpoint_channel.h",
"bluetooth_bwu_handler.h",
"bluetooth_device_name.h",
"bluetooth_endpoint_channel.h",
"bwu_handler.h",
@@ -57,6 +60,7 @@ cc_library(
"service_controller_router.h",
"webrtc_bwu_handler.h",
"webrtc_endpoint_channel.h",
"wifi_lan_bwu_handler.h",
"wifi_lan_endpoint_channel.h",
"wifi_lan_service_info.h",
],
cpp/core/internal/base_bwu_handler.h
-2
View File
@@ -30,8 +30,6 @@ class BaseBwuHandler : public BwuHandler {
: channel_manager_(&channel_manager),
bwu_notifications_(std::move(bwu_notifications)) {}
~BaseBwuHandler() override = default;
void OnIncomingConnection(ClientProxy* client,
IncomingSocketConnection* connection);
protected:
// Represents the incoming Socket the Initiator has gotten after initializing
cpp/core/internal/base_pcp_handler.cc
+4 -4
View File
@@ -673,12 +673,12 @@ void BasePcpHandler::OnIncomingFrame(OfflineFrame& frame,
void BasePcpHandler::OnEndpointDisconnect(ClientProxy* client,
const std::string& endpoint_id,
CountDownLatch* barrier) {
CountDownLatch barrier) {
if (stop_.Get()) {
if (barrier) barrier->CountDown();
barrier.CountDown();
return;
}
RunOnPcpHandlerThread([this, client, endpoint_id, barrier]() {
RunOnPcpHandlerThread([this, client, endpoint_id, barrier]() mutable {
auto item = pending_alarms_.find(endpoint_id);
if (item != pending_alarms_.end()) {
auto& alarm = item->second;
@@ -686,7 +686,7 @@ void BasePcpHandler::OnEndpointDisconnect(ClientProxy* client,
pending_alarms_.erase(item);
}
ProcessPreConnectionResultFailure(client, endpoint_id);
barrier->CountDown();
barrier.CountDown();
});
}
cpp/core/internal/base_pcp_handler.h
+1 -1
View File
@@ -149,7 +149,7 @@ class BasePcpHandler : public PcpHandler,
// approve/reject the connection.
// @EndpointManagerThread
void OnEndpointDisconnect(ClientProxy* client, const std::string& endpoint_id,
CountDownLatch* barrier) override;
CountDownLatch barrier) override;
Pcp GetPcp() const override { return pcp_; }
Strategy GetStrategy() const override { return strategy_; }
cpp/core/internal/base_pcp_handler_test.cc
+12 -1
View File
@@ -379,6 +379,7 @@ TEST_P(BasePcpHandlerTest, ConstructorDestructorWorks) {
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
SUCCEED();
bwu.Shutdown();
}
TEST_P(BasePcpHandlerTest, StartAdvertisingChangesState) {
@@ -389,6 +390,7 @@ TEST_P(BasePcpHandlerTest, StartAdvertisingChangesState) {
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartAdvertising(&client, &pcp_handler);
bwu.Shutdown();
}
TEST_P(BasePcpHandlerTest, StopAdvertisingChangesState) {
@@ -403,6 +405,7 @@ TEST_P(BasePcpHandlerTest, StopAdvertisingChangesState) {
EXPECT_TRUE(client.IsAdvertising());
pcp_handler.StopAdvertising(&client);
EXPECT_FALSE(client.IsAdvertising());
bwu.Shutdown();
}
TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) {
@@ -413,6 +416,7 @@ TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) {
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
bwu.Shutdown();
}
TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) {
@@ -427,6 +431,7 @@ TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) {
EXPECT_TRUE(client.IsDiscovering());
pcp_handler.StopDiscovery(&client);
EXPECT_FALSE(client.IsDiscovering());
bwu.Shutdown();
}
TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) {
@@ -450,6 +455,7 @@ TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) {
&pcp_handler, connect_medium);
NEARBY_LOG(INFO, "RequestConnection complete");
channel_b->Close();
bwu.Shutdown();
pcp_handler.DisconnectFromEndpointManager();
}
@@ -478,6 +484,7 @@ TEST_P(BasePcpHandlerTest, AcceptConnectionChangesState) {
EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(AtLeast(0));
NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id;
channel_b->Close();
bwu.Shutdown();
pcp_handler.DisconnectFromEndpointManager();
}
@@ -502,6 +509,7 @@ TEST_P(BasePcpHandlerTest, RejectConnectionChangesState) {
Status{Status::kSuccess});
NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id;
channel_b->Close();
bwu.Shutdown();
pcp_handler.DisconnectFromEndpointManager();
}
@@ -536,6 +544,7 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) {
connect_medium);
NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id;
channel_b->Close();
bwu.Shutdown();
pcp_handler.DisconnectFromEndpointManager();
}
@@ -568,8 +577,8 @@ TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) {
EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(AtLeast(0));
NEARBY_LOG(INFO, "Closing connection: id=%s", endpoint_id.c_str());
channel_b->Close();
pcp_handler.DisconnectFromEndpointManager();
bwu.Shutdown();
pcp_handler.DisconnectFromEndpointManager();
}
EXPECT_EQ(destroyed_flag.load(), mediums_count);
}
@@ -615,6 +624,7 @@ TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) {
}
NEARBY_LOG(INFO, "Closing connection: id=%s", endpoint_id.c_str());
channel_b->Close();
bwu.Shutdown();
pcp_handler.DisconnectFromEndpointManager();
}
EXPECT_EQ(destroyed_flag.load(), mediums_count);
@@ -672,6 +682,7 @@ TEST_F(BasePcpHandlerTest, InjectEndpoint) {
.medium = Medium::BLUETOOTH,
.remote_bluetooth_mac_address = ByteArray(kFakeMacAddress),
});
bwu.Shutdown();
}
} // namespace
cpp/core/internal/bluetooth_bwu_handler.cc
+116
View File
@@ -0,0 +1,116 @@
#include "core/internal/bluetooth_bwu_handler.h"
#include "core/internal/bluetooth_endpoint_channel.h"
#include "core/internal/client_proxy.h"
#include "core/internal/offline_frames.h"
#include "absl/functional/bind_front.h"
// Manages the Bluetooth-specific methods needed to upgrade an {@link
// EndpointChannel}.
namespace location {
namespace nearby {
namespace connections {
BluetoothBwuHandler::BluetoothBwuHandler(
Mediums& mediums, EndpointChannelManager& channel_manager,
BwuNotifications notifications)
: BaseBwuHandler(channel_manager, std::move(notifications)),
mediums_(mediums) {}
void BluetoothBwuHandler::Revert() {
for (const std::string& service_id : active_service_ids_) {
bluetooth_medium_.StopAcceptingConnections(service_id);
}
active_service_ids_.clear();
NEARBY_LOG(INFO,
"BluetoothBwuHandler successfully reverted all Bluetooth state.");
}
// Accept Connection Callback.
// Notifies that the remote party called BluetoothClassic::Connect()
// for this socket.
void BluetoothBwuHandler::OnIncomingBluetoothConnection(
ClientProxy* client, const std::string& service_id,
BluetoothSocket socket) {
auto channel =
absl::make_unique<BluetoothEndpointChannel>(service_id, socket);
std::unique_ptr<IncomingSocketConnection> connection{
new IncomingSocketConnection{
.socket =
std::make_unique<BluetoothIncomingSocket>(service_id, socket),
.channel = std::move(channel),
}};
bwu_notifications_.incoming_connection_cb(client, std::move(connection));
}
// Called by BWU initiator. BT Medium is set up, and BWU request is prepared,
// with necessary info (service_id, MAC address) for remote party to perform
// discovery.
ByteArray BluetoothBwuHandler::InitializeUpgradedMediumForEndpoint(
ClientProxy* client, const std::string& service_id,
const std::string& endpoint_id) {
std::string upgrade_service_id = Utils::WrapUpgradeServiceId(service_id);
std::string mac_address = bluetooth_medium_.GetMacAddress();
if (mac_address.empty()) {
return {};
}
if (!bluetooth_medium_.IsAcceptingConnections(upgrade_service_id)) {
if (!bluetooth_medium_.StartAcceptingConnections(
upgrade_service_id,
{
.accepted_cb = absl::bind_front(
&BluetoothBwuHandler::OnIncomingBluetoothConnection, this,
client, service_id),
})) {
return {};
}
}
// cache service ID to revert
active_service_ids_.emplace(upgrade_service_id);
return parser::ForBwuBluetoothPathAvailable(upgrade_service_id, mac_address);
}
// Called by BWU target. Retrieves a new medium info from incoming message,
// and establishes connection over BT using this info.
// Returns a channel ready to exchange data or nullptr on error.
std::unique_ptr<EndpointChannel>
BluetoothBwuHandler::CreateUpgradedEndpointChannel(
ClientProxy* client, const std::string& service_id,
const std::string& endpoint_id, const UpgradePathInfo& upgrade_path_info) {
const UpgradePathInfo::BluetoothCredentials& bluetooth_credentials =
upgrade_path_info.bluetooth_credentials();
if (!bluetooth_credentials.has_service_name() ||
!bluetooth_credentials.has_mac_address()) {
return nullptr;
}
const std::string& service_name = bluetooth_credentials.service_name();
const std::string& mac_address = bluetooth_credentials.mac_address();
BluetoothDevice device = bluetooth_medium_.GetRemoteDevice(mac_address);
if (!device.IsValid()) {
return nullptr;
}
BluetoothSocket socket = bluetooth_medium_.Connect(device, service_name);
if (!socket.IsValid()) {
return nullptr;
}
auto channel =
std::make_unique<BluetoothEndpointChannel>(service_name, socket);
if (channel == nullptr) {
socket.Close();
return nullptr;
}
return channel;
}
} // namespace connections
} // namespace nearby
} // namespace location
cpp/core/internal/bluetooth_bwu_handler.h
+83
View File
@@ -0,0 +1,83 @@
#ifndef CORE_INTERNAL_BLUETOOTH_BWU_HANDLER_H_
#define CORE_INTERNAL_BLUETOOTH_BWU_HANDLER_H_
#include <string>
#include "core/internal/base_bwu_handler.h"
#include "core/internal/client_proxy.h"
#include "core/internal/mediums/mediums.h"
#include "core/internal/mediums/utils.h"
#include "proto/connections/offline_wire_formats.pb.h"
#include "platform/public/bluetooth_classic.h"
#include "platform/public/count_down_latch.h"
#include "proto/connections_enums.pb.h"
namespace location {
namespace nearby {
namespace connections {
// Defines the set of methods that need to be implemented to handle the
// per-Medium-specific operations needed to upgrade an EndpointChannel.
class BluetoothBwuHandler : public BaseBwuHandler {
public:
BluetoothBwuHandler(Mediums& mediums, EndpointChannelManager& channel_manager,
BwuNotifications notifications);
~BluetoothBwuHandler() override = default;
private:
constexpr static const int kServiceIdLength = 10;
// Implements BaseBwuHandler:
// Reverts any changes made to the device in the process of upgrading
// endpoints.
void Revert() override;
// Cleans up in-progress upgrades after endpoint disconnection.
void OnEndpointDisconnect(ClientProxy* client,
const std::string& endpoint_id) override {}
void OnIncomingBluetoothConnection(ClientProxy* client,
const std::string& service_id,
BluetoothSocket socket);
class BluetoothIncomingSocket : public IncomingSocket {
public:
explicit BluetoothIncomingSocket(const std::string& name,
BluetoothSocket socket)
: name_(name), socket_(socket) {}
~BluetoothIncomingSocket() override = default;
std::string ToString() override { return name_; }
void Close() override { socket_.Close(); }
private:
std::string name_;
BluetoothSocket socket_;
};
// First part of InitiateBwuForEndpoint implementation;
// returns a BWU request to remote party as byte array.
ByteArray InitializeUpgradedMediumForEndpoint(
ClientProxy* client, const std::string& service_id,
const std::string& endpoint_id) override;
// Invoked from OnBwuNegotiationFrame.
std::unique_ptr<EndpointChannel> CreateUpgradedEndpointChannel(
ClientProxy* client, const std::string& service_id,
const std::string& endpoint_id,
const UpgradePathInfo& upgrade_path_info) override;
// Returns the upgrade medium of the BwuHandler.
// @BwuHandlerThread
Medium GetUpgradeMedium() const override { return Medium::BLUETOOTH; }
Mediums& mediums_;
absl::flat_hash_set<std::string> active_service_ids_;
BluetoothRadio& bluetooth_radio_{mediums_.GetBluetoothRadio()};
BluetoothClassic& bluetooth_medium_{mediums_.GetBluetoothClassic()};
};
} // namespace connections
} // namespace nearby
} // namespace location
#endif // CORE_INTERNAL_BLUETOOTH_BWU_HANDLER_H_
cpp/core/internal/bwu_handler.h
+2
View File
@@ -28,6 +28,7 @@ class BwuHandler {
virtual ByteArray InitializeUpgradedMediumForEndpoint(
ClientProxy* client, const std::string& service_id,
const std::string& endpoint_id) = 0;
// Called to revert any state changed by the Initiator to setup the upgraded
// medium for an endpoint.
// @BwuHandlerThread
@@ -41,6 +42,7 @@ class BwuHandler {
ClientProxy* client, const std::string& service_id,
const std::string& endpoint_id,
const UpgradePathInfo& upgrade_path_info) = 0;
// Returns the upgrade medium of the BwuHandler.
// @BwuHandlerThread
virtual Medium GetUpgradeMedium() const = 0;
cpp/core/internal/bwu_manager.cc
+34 -25
View File
@@ -3,9 +3,11 @@
#include <algorithm>
#include <memory>
#include "core/internal/bluetooth_bwu_handler.h"
#include "core/internal/bwu_handler.h"
#include "core/internal/offline_frames.h"
#include "core/internal/webrtc_bwu_handler.h"
#include "core/internal/wifi_lan_bwu_handler.h"
#include "platform/base/byte_array.h"
#include "platform/public/count_down_latch.h"
#include "proto/connections_enums.pb.h"
@@ -54,11 +56,21 @@ void BwuManager::InitBwuHandlers() {
.incoming_connection_cb =
absl::bind_front(&BwuManager::OnIncomingConnection, this),
};
if (config_.allow_upgrade_to.wifi_lan) {
handlers_.emplace(Medium::WIFI_LAN,
std::make_unique<WifiLanBwuHandler>(
*mediums_, *channel_manager_, notifications));
}
if (config_.allow_upgrade_to.web_rtc) {
handlers_.emplace(Medium::WEB_RTC,
std::make_unique<WebrtcBwuHandler>(
*mediums_, *channel_manager_, notifications));
}
if (config_.allow_upgrade_to.bluetooth) {
handlers_.emplace(Medium::BLUETOOTH,
std::make_unique<BluetoothBwuHandler>(
*mediums_, *channel_manager_, notifications));
}
}
void BwuManager::Shutdown() {
@@ -67,31 +79,26 @@ void BwuManager::Shutdown() {
endpoint_manager_->UnregisterFrameProcessor(
V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION, this);
CountDownLatch latch(1);
RunOnBwuManagerThread([this, &latch]() {
for (auto& item : previous_endpoint_channels_) {
EndpointChannel* channel = item.second.get();
if (!channel) continue;
channel->Close(DisconnectionReason::SHUTDOWN);
}
CancelAllRetryUpgradeAlarms();
medium_ = Medium::UNKNOWN_MEDIUM;
for (auto& item : handlers_) {
BwuHandler& handler = *item.second;
handler.Revert();
}
handlers_.clear();
latch.CountDown();
});
latch.Await();
// Stop all the ongoing Runnables (as gracefully as possible).
alarm_executor_.Shutdown();
serial_executor_.Shutdown();
// After worker threads are down we became exclusive owners of data and
// may access it from current thread.
for (auto& item : previous_endpoint_channels_) {
EndpointChannel* channel = item.second.get();
if (!channel) continue;
channel->Close(DisconnectionReason::SHUTDOWN);
}
CancelAllRetryUpgradeAlarms();
medium_ = Medium::UNKNOWN_MEDIUM;
for (auto& item : handlers_) {
BwuHandler& handler = *item.second;
handler.Revert();
}
handlers_.clear();
NEARBY_LOG(INFO, "BwuHandler has shut down.");
}
@@ -182,10 +189,10 @@ void BwuManager::OnIncomingFrame(OfflineFrame& frame,
void BwuManager::OnEndpointDisconnect(ClientProxy* client,
const std::string& endpoint_id,
CountDownLatch* barrier) {
RunOnBwuManagerThread([this, client, endpoint_id, barrier]() {
CountDownLatch barrier) {
RunOnBwuManagerThread([this, client, endpoint_id, barrier]() mutable {
if (medium_ == Medium::UNKNOWN_MEDIUM) {
barrier->CountDown();
barrier.CountDown();
return;
}
@@ -213,7 +220,7 @@ void BwuManager::OnEndpointDisconnect(ClientProxy* client,
if (channel_manager_->GetConnectedEndpointsCount() <= 1) {
Revert();
}
barrier->CountDown();
barrier.CountDown();
});
}
@@ -541,6 +548,8 @@ void BwuManager::ProcessSafeToClosePriorChannelEvent(
"trying to upgrade endpoint %s.",
endpoint_id.c_str());
previous_endpoint_channel->Write(parser::ForDisconnection());
// Wait for in-flight messages to reach their peers.
SystemClock::Sleep(absl::Seconds(1));
previous_endpoint_channel->Close(DisconnectionReason::UPGRADED);
cpp/core/internal/bwu_manager.cc.orig
-780
View File
@@ -1,780 +0,0 @@
#include "core/internal/bwu_manager.h"
#include <algorithm>
#include <memory>
#include "core/internal/bluetooth_bwu_handler.h"
#include "core/internal/bwu_handler.h"
#include "core/internal/offline_frames.h"
#include "core/internal/webrtc_bwu_handler.h"
#include "platform/base/byte_array.h"
#include "platform/public/count_down_latch.h"
#include "proto/connections_enums.pb.h"
#include "absl/functional/bind_front.h"
#include "absl/time/time.h"
namespace location {
namespace nearby {
namespace connections {
using ::location::nearby::proto::connections::ConnectionAttemptResult;
using ::location::nearby::proto::connections::DisconnectionReason;
BwuManager::BwuManager(
Mediums& mediums, EndpointManager& endpoint_manager,
EndpointChannelManager& channel_manager,
absl::flat_hash_map<Medium, std::unique_ptr<BwuHandler>> handlers,
Config config)
: config_(config),
mediums_(&mediums),
endpoint_manager_(&endpoint_manager),
channel_manager_(&channel_manager) {
if (config_.bandwidth_upgrade_retry_delay == absl::ZeroDuration()) {
config_.bandwidth_upgrade_retry_delay = absl::Seconds(5);
}
if (config_.bandwidth_upgrade_retry_max_delay == absl::ZeroDuration()) {
config_.bandwidth_upgrade_retry_max_delay = absl::Seconds(10);
}
if (config_.allow_upgrade_to.All(false)) {
config_.allow_upgrade_to.web_rtc = true;
}
if (!handlers.empty()) {
handlers_ = std::move(handlers);
} else {
InitBwuHandlers();
}
// Register the offline frame processor.
endpoint_manager_->RegisterFrameProcessor(
V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION, this);
}
void BwuManager::InitBwuHandlers() {
// Register the supported concrete BwuMedium implementations.
BwuHandler::BwuNotifications notifications{
.incoming_connection_cb =
absl::bind_front(&BwuManager::OnIncomingConnection, this),
};
if (config_.allow_upgrade_to.web_rtc) {
handlers_.emplace(Medium::WEB_RTC,
std::make_unique<WebrtcBwuHandler>(
*mediums_, *channel_manager_, notifications));
}
if (config_.allow_upgrade_to.bluetooth) {
handlers_.emplace(Medium::BLUETOOTH,
std::make_unique<BluetoothBwuHandler>(
*mediums_, *channel_manager_, notifications));
}
}
void BwuManager::Shutdown() {
NEARBY_LOG(INFO, "Initiating shutdown of BwuManager.");
endpoint_manager_->UnregisterFrameProcessor(
V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION, this);
CountDownLatch latch(1);
RunOnBwuManagerThread([this, &latch]() {
for (auto& item : previous_endpoint_channels_) {
EndpointChannel* channel = item.second.get();
if (!channel) continue;
channel->Close(DisconnectionReason::SHUTDOWN);
}
CancelAllRetryUpgradeAlarms();
medium_ = Medium::UNKNOWN_MEDIUM;
for (auto& item : handlers_) {
BwuHandler& handler = *item.second;
handler.Revert();
}
handlers_.clear();
latch.CountDown();
});
latch.Await();
// Stop all the ongoing Runnables (as gracefully as possible).
alarm_executor_.Shutdown();
serial_executor_.Shutdown();
NEARBY_LOG(INFO, "BwuHandler has shut down.");
}
// This is the point on the Initiator side where the
// medium_ is set.
void BwuManager::InitiateBwuForEndpoint(ClientProxy* client,
const std::string& endpoint_id,
Medium new_medium) {
RunOnBwuManagerThread([this, client, endpoint_id, new_medium]() {
Medium proposed_medium = ChooseBestUpgradeMedium(
client->GetUpgradeMediums(endpoint_id).GetMediums(true));
if (new_medium != Medium::UNKNOWN_MEDIUM) {
proposed_medium = new_medium;
}
auto* handler = SetCurrentBwuHandler(proposed_medium);
if (!handler) return;
if (in_progress_upgrades_.contains(endpoint_id)) {
return;
}
auto channel = channel_manager_->GetChannelForEndpoint(endpoint_id);
if (channel == nullptr) {
return;
}
// Ignore requests where the medium we're upgrading to is the medium we're
// already connected over. This can happen now that Bluetooth is both an
// advertising medium and a potential bandwidth upgrade, and will continue
// to be possible as we add other new advertising mediums like mDNS (WiFi
// LAN). Very specifically, this happens now when a device uses P2P_CLUSTER,
// connects over Bluetooth, and is not connected to LAN. Bluetooth is the
// best medium, and we attempt to upgrade from Bluetooth to Bluetooth.
if (medium_ == channel->GetMedium()) {
return;
}
std::string service_id = client->GetServiceId();
ByteArray bytes = handler->InitializeUpgradedMediumForEndpoint(
client, service_id, endpoint_id);
// Because we grab the endpointChannel first thing, it is possible the
// endpointChannel is stale by the time we attempt to write over it.
if (bytes.Empty()) {
NEARBY_LOG(ERROR,
"Couldn't complete the upgrade for endpoint "
"%s to %d because it failed to initialize the "
"BWU_NEGOTIATION.UPGRADE_PATH_AVAILABLE OfflineFrame.",
endpoint_id.c_str(), medium_);
UpgradePathInfo info;
info.set_medium(parser::MediumToUpgradePathInfoMedium(medium_));
ProcessUpgradeFailureEvent(client, endpoint_id, info);
return;
}
if (!channel->Write(bytes).Ok()) {
NEARBY_LOG(ERROR,
"Couldn't complete the upgrade for endpoint %s to %d because "
"it failed to write the "
"BWU_NEGOTIATION.UPGRADE_PATH_AVAILABLE OfflineFrame.",
endpoint_id.c_str(), medium_);
return;
}
NEARBY_LOG(INFO,
"Successfully wrote the BWU_NEGOTIATION.UPGRADE_PATH_AVAILABLE "
"OfflineFrame while upgrading endpoint %s to %d.",
endpoint_id.c_str(), medium_);
in_progress_upgrades_.emplace(endpoint_id, client);
});
}
void BwuManager::OnIncomingFrame(OfflineFrame& frame,
const std::string& endpoint_id,
ClientProxy* client, Medium medium) {
if (parser::GetFrameType(frame) != V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION)
return;
auto bwu_frame = frame.v1().bandwidth_upgrade_negotiation();
CountDownLatch latch(1);
RunOnBwuManagerThread([this, client, endpoint_id, &bwu_frame, &latch]() {
OnBwuNegotiationFrame(client, bwu_frame, endpoint_id);
latch.CountDown();
});
latch.Await();
}
void BwuManager::OnEndpointDisconnect(ClientProxy* client,
const std::string& endpoint_id,
CountDownLatch* barrier) {
RunOnBwuManagerThread([this, client, endpoint_id, barrier]() {
if (medium_ == Medium::UNKNOWN_MEDIUM) {
barrier->CountDown();
return;
}
if (handler_) {
handler_->OnEndpointDisconnect(client, endpoint_id);
}
auto item = previous_endpoint_channels_.extract(endpoint_id);
if (!item.empty()) {
auto old_channel = item.mapped();
if (old_channel != nullptr) {
old_channel->Close(DisconnectionReason::SHUTDOWN);
}
}
in_progress_upgrades_.erase(endpoint_id);
CancelRetryUpgradeAlarm(endpoint_id);
successfully_upgraded_endpoints_.erase(endpoint_id);
// If this was our very last endpoint:
//
// a) revert all the changes for currentBwuMedium.
// b) reset currentBwuMedium.
if (channel_manager_->GetConnectedEndpointsCount() <= 1) {
Revert();
}
barrier->CountDown();
});
}
BwuHandler* BwuManager::SetCurrentBwuHandler(Medium medium) {
handler_ = nullptr;
medium_ = medium;
if (medium != Medium::UNKNOWN_MEDIUM) {
auto item = handlers_.find(medium);
if (item != handlers_.end()) {
handler_ = item->second.get();
}
}
return handler_;
}
void BwuManager::Revert() {
if (handler_) {
handler_->Revert();
medium_ = Medium::UNKNOWN_MEDIUM;
handler_ = nullptr;
}
}
void BwuManager::OnBwuNegotiationFrame(ClientProxy* client,
const BwuNegotiationFrame& frame,
const string& endpoint_id) {
switch (frame.event_type()) {
case BwuNegotiationFrame::UPGRADE_PATH_AVAILABLE:
ProcessBwuPathAvailableEvent(client, endpoint_id,
frame.upgrade_path_info());
break;
case BwuNegotiationFrame::UPGRADE_FAILURE:
ProcessUpgradeFailureEvent(client, endpoint_id,
frame.upgrade_path_info());
break;
case BwuNegotiationFrame::LAST_WRITE_TO_PRIOR_CHANNEL:
ProcessLastWriteToPriorChannelEvent(client, endpoint_id);
break;
case BwuNegotiationFrame::SAFE_TO_CLOSE_PRIOR_CHANNEL:
ProcessSafeToClosePriorChannelEvent(client, endpoint_id);
break;
default:
break;
}
}
void BwuManager::OnIncomingConnection(
ClientProxy* client,
std::unique_ptr<BwuHandler::IncomingSocketConnection> mutable_connection) {
std::shared_ptr<BwuHandler::IncomingSocketConnection> connection(
mutable_connection.release());
RunOnBwuManagerThread([this, client, connection]() {
EndpointChannel* channel = connection->channel.get();
if (channel == nullptr) {
connection->socket->Close();
return;
}
ClientIntroduction introduction;
if (!ReadClientIntroductionFrame(channel, introduction)) {
// This was never a fully EstablishedConnection, no need to provide a
// closure reason.
channel->Close();
return;
}
const std::string& endpoint_id = introduction.endpoint_id();
auto item = in_progress_upgrades_.extract(endpoint_id);
if (item.empty()) return;
ClientProxy* mapped_client = item.mapped();
CancelRetryUpgradeAlarm(endpoint_id);
if (mapped_client == nullptr) {
// This was never a fully EstablishedConnection, no need to provide a
// closure reason.
channel->Close();
return;
}
CHECK(client == mapped_client);
// Use the introductory client information sent over to run the upgrade
// protocol.
RunUpgradeProtocol(mapped_client, endpoint_id,
std::move(connection->channel));
});
}
void BwuManager::RunOnBwuManagerThread(Runnable runnable) {
serial_executor_.Execute(std::move(runnable));
}
void BwuManager::RunUpgradeProtocol(
ClientProxy* client, const std::string& endpoint_id,
std::unique_ptr<EndpointChannel> new_channel) {
// First, register this new EndpointChannel as *the* EndpointChannel to use
// for this endpoint here onwards. NOTE: We pause this new EndpointChannel
// until we've completely drained the old EndpointChannel to avoid out of
// order reads on the other side. This is a consequence of using the same
// UKEY2 context for both the previous and new EndpointChannels. UKEY2 uses
// sequence numbers for writes and reads, and simultaneously sending Payloads
// on the new channel and control messages on the old channel cause the other
// side to read messages out of sequence
new_channel->Pause();
auto old_channel = channel_manager_->GetChannelForEndpoint(endpoint_id);
if (!old_channel) return;
channel_manager_->ReplaceChannelForEndpoint(client, endpoint_id,
std::move(new_channel));
// Next, initiate a clean shutdown for the previous EndpointChannel used for
// this endpoint by telling the remote device that it will not receive any
// more writes over that EndpointChannel.
if (!old_channel->Write(parser::ForBwuLastWrite()).Ok()) {
return;
}
// The remainder of this clean shutdown for the previous EndpointChannel will
// continue when we receive a corresponding
// BANDWIDTH_UPGRADE_NEGOTIATION.LAST_WRITE_TO_PRIOR_CHANNEL OfflineFrame from
// the remote device, so for now, just store that previous EndpointChannel.
previous_endpoint_channels_.emplace(endpoint_id, old_channel);
// If we already read LAST_WRITE on the old endpoint channel, then we can
// safely close it now.
auto item = successfully_upgraded_endpoints_.extract(endpoint_id);
if (!item.empty()) {
ProcessLastWriteToPriorChannelEvent(client, endpoint_id);
}
}
// Outgoing BWU session.
void BwuManager::ProcessBwuPathAvailableEvent(
ClientProxy* client, const string& endpoint_id,
const UpgradePathInfo& upgrade_path_info) {
Medium medium =
parser::UpgradePathInfoMediumToMedium(upgrade_path_info.medium());
if (medium_ == Medium::UNKNOWN_MEDIUM) {
SetCurrentBwuHandler(medium);
}
// Check for the correct medium so we don't process an incorrect OfflineFrame.
if (medium != medium_) {
RunUpgradeFailedProtocol(client, endpoint_id, upgrade_path_info);
return;
}
auto channel = ProcessBwuPathAvailableEventInternal(client, endpoint_id,
upgrade_path_info);
ConnectionAttemptResult connectionAttemptResult;
if (channel != nullptr) {
connectionAttemptResult = ConnectionAttemptResult::RESULT_SUCCESS;
} else {
connectionAttemptResult = ConnectionAttemptResult::RESULT_ERROR;
}
if (channel == nullptr) {
RunUpgradeFailedProtocol(client, endpoint_id, upgrade_path_info);
return;
}
RunUpgradeProtocol(client, endpoint_id, std::move(channel));
}
std::unique_ptr<EndpointChannel>
BwuManager::ProcessBwuPathAvailableEventInternal(
ClientProxy* client, const string& endpoint_id,
const UpgradePathInfo& upgrade_path_info) {
std::unique_ptr<EndpointChannel> channel =
handler_->CreateUpgradedEndpointChannel(client, client->GetServiceId(),
endpoint_id, upgrade_path_info);
if (!channel) {
return nullptr;
}
// Write the requisite BANDWIDTH_UPGRADE_NEGOTIATION.CLIENT_INTRODUCTION as
// the first OfflineFrame on this new EndpointChannel.
if (!channel->Write(parser::ForBwuIntroduction(client->GetLocalEndpointId()))
.Ok()) {
// This was never a fully EstablishedConnection, no need to provide a
// closure reason.
channel->Close();
NEARBY_LOG(
ERROR,
"Failed to write BWU_NEGOTIATION.CLIENT_INTRODUCTION OfflineFrame to "
"newly-created EndpointChannel %s, aborting upgrade.",
channel->GetName().c_str());
return {};
}
NEARBY_LOG(
INFO,
"Successfully wrote BWU_NEGOTIATION.CLIENT_INTRODUCTION OfflineFrame to "
"newly-created EndpointChannel %s while upgrading endpoint %s.",
channel->GetName().c_str(), endpoint_id.c_str());
// Set the AnalyticsRecorder so that the future closure of this
// EndpointChannel will be recorded.
return channel;
}
void BwuManager::RunUpgradeFailedProtocol(
ClientProxy* client, const std::string& endpoint_id,
const UpgradePathInfo& upgrade_path_info) {
// We attempted to connect to the new medium that the remote device has set up
// for us but we failed. We need to let the remote device know so that they
// can pick another medium for us to try.
std::shared_ptr<EndpointChannel> channel =
channel_manager_->GetChannelForEndpoint(endpoint_id);
if (!channel) {
NEARBY_LOG(ERROR,
"Couldn't find a previous EndpointChannel for %s "
"when sending an upgrade failure frame, short-circuiting the "
"upgrade protocol.",
endpoint_id.c_str());
return;
}
// Report UPGRADE_FAILURE to the remote device.
if (!channel->Write(parser::ForBwuFailure(upgrade_path_info)).Ok()) {
channel->Close(DisconnectionReason::IO_ERROR);
NEARBY_LOG(
ERROR,
"Failed to write BANDWIDTH_UPGRADE_NEGOTIATION.UPGRADE_FAILURE "
"OfflineFrame to endpoint %s, short-circuiting the upgrade protocol.",
endpoint_id.c_str());
return;
}
// And lastly, clean up our currentBwuMedium since we failed to
// utilize it anyways.
if (medium_ != Medium::UNKNOWN_MEDIUM) {
Revert();
}
}
bool BwuManager::ReadClientIntroductionFrame(EndpointChannel* channel,
ClientIntroduction& introduction) {
auto data = channel->Read();
if (!data.ok()) return false;
auto transfer(parser::FromBytes(data.result()));
if (!transfer.ok()) return false;
OfflineFrame frame = transfer.result();
if (!frame.has_v1() || !frame.v1().has_bandwidth_upgrade_negotiation())
return false;
const auto& frame_intro =
frame.v1().bandwidth_upgrade_negotiation().client_introduction();
introduction = frame_intro;
return true;
}
void BwuManager::ProcessLastWriteToPriorChannelEvent(
ClientProxy* client, const std::string& endpoint_id) {
// By this point in the upgrade protocol, there is the guarantee that both
// involved endpoints have registered a new EndpointChannel with the
// EndpointChannelManager as the official channel for communication; given
// the way communication is structured in the EndpointManager, this means
// that all new writes are happening over that new EndpointChannel, but
// reads are still happening over this prior EndpointChannel (to avoid data
// loss). But now that we've received this definitive final write over that
// prior EndpointChannel, we can let the remote device that they can safely
// close their end of this now-dormant EndpointChannel.
EndpointChannel* previous_endpoint_channel =
previous_endpoint_channels_[endpoint_id].get();
if (!previous_endpoint_channel) {
NEARBY_LOG(
ERROR,
"Received a BWU_NEGOTIATION.LAST_WRITE_TO_PRIOR_CHANNEL OfflineFrame "
"for unknown endpoint %s, can't complete the upgrade protocol.",
endpoint_id.c_str());
successfully_upgraded_endpoints_.emplace(endpoint_id);
return;
}
if (!previous_endpoint_channel->Write(parser::ForBwuSafeToClose()).Ok()) {
previous_endpoint_channel->Close(DisconnectionReason::IO_ERROR);
// Remove this prior EndpointChannel from previous_endpoint_channels to
// avoid leaks.
previous_endpoint_channels_.erase(endpoint_id);
NEARBY_LOG(
ERROR,
"Failed to write BWU_NEGOTIATION.SAFE_TO_CLOSE_PRIOR_CHANNEL "
"OfflineFrame to endpoint %s, short-circuiting the upgrade protocol.",
endpoint_id.c_str());
return;
}
// The upgrade protocol's clean shutdown of the prior EndpointChannel will
// conclude when we receive a corresponding
// BANDWIDTH_UPGRADE_NEGOTIATION.SAFE_TO_CLOSE_PRIOR_CHANNEL OfflineFrame
// from the remote device.
}
void BwuManager::ProcessSafeToClosePriorChannelEvent(
ClientProxy* client, const std::string& endpoint_id) {
// By this point in the upgrade protocol, there's no more writes happening
// over the prior EndpointChannel, and the remote device has given us the
// go-ahead to close this EndpointChannel [1], so we can safely close it
// (and depend on the EndpointManager querying the EndpointChannelManager to
// start reading from the new EndpointChannel).
//
// [1] Which also implies that they've received our
// BANDWIDTH_UPGRADE_NEGOTIATION.LAST_WRITE_TO_PRIOR_CHANNEL OfflineFrame),
// so there can be no data loss, regardless of whether the EndpointChannel
// allows reads of queued, unread data after the EndpointChannel has been
// closed from the other end (as is the case with conventional TCP sockets)
// or not (as is the case with Android's Bluetooth sockets, where closing
// instantly throws an IOException on the remote device).
auto item = previous_endpoint_channels_.extract(endpoint_id);
auto& previous_endpoint_channel = item.mapped();
if (previous_endpoint_channel == nullptr) {
NEARBY_LOG(
ERROR,
"Received a BWU_NEGOTIATION.SAFE_TO_CLOSE_PRIOR_CHANNEL OfflineFrame "
"for unknown endpoint %s, can't complete the upgrade protocol.",
endpoint_id.c_str());
return;
}
NEARBY_LOG(INFO,
"BwuManager successfully received a "
"BWU_NEGOTIATION.SAFE_TO_CLOSE_PRIOR_CHANNEL OfflineFrame while "
"trying to upgrade endpoint %s.",
endpoint_id.c_str());
// Wait for in-flight messages to reach their peers.
SystemClock::Sleep(absl::Seconds(1));
previous_endpoint_channel->Close(DisconnectionReason::UPGRADED);
// Now that the old channel has been drained, we can unpause the new channel
std::shared_ptr<EndpointChannel> channel =
channel_manager_->GetChannelForEndpoint(endpoint_id);
if (!channel) {
NEARBY_LOG(ERROR,
"Attempted to resume the current EndpointChannel with endpoint "
"%s, but none was found",
endpoint_id.c_str());
return;
}
channel->Resume();
// Report the success to the client
client->OnBandwidthChanged(endpoint_id, channel->GetMedium());
}
void BwuManager::ProcessUpgradeFailureEvent(
ClientProxy* client, const std::string& endpoint_id,
const UpgradePathInfo& upgrade_info) {
// The remote device failed to upgrade to the new medium we set up for them.
// That's alright! We'll just try the next available medium (if there is
// one).
in_progress_upgrades_.erase(endpoint_id);
// The first thing we have to do is to replace our
// currentBwuMedium with the next best upgrade medium we share
// with the remote device. The catch is that we can only do this if we only
// have one connected endpoint. Otherwise, we'll end up disrupting our other
// connected peers.
if (channel_manager_->GetConnectedEndpointsCount() > 1) {
// We can't change the currentBwuMedium, so there are no more
// upgrade attempts for this endpoint. Sorry.
NEARBY_LOG(
ERROR,
"Failed to attempt a new bandwidth upgrade for endpoint %s because we "
"have other connected endpoints and can't try a new upgrade medium.",
endpoint_id.c_str());
return;
}
// Revert the existing upgrade medium for now.
if (medium_ != Medium::UNKNOWN_MEDIUM) {
Revert();
}
// Loop through the ordered list of upgrade mediums. One by one, remove the
// top element until we get to the medium we last attempted to upgrade to.
// The remainder of the list will contain the mediums we haven't attempted
// yet.
Medium last = parser::UpgradePathInfoMediumToMedium(upgrade_info.medium());
std::vector<Medium> all_possible_mediums =
client->GetUpgradeMediums(endpoint_id).GetMediums(true);
std::vector<Medium> untried_mediums(all_possible_mediums);
for (Medium medium : all_possible_mediums) {
untried_mediums.erase(untried_mediums.begin());
if (medium == last) {
break;
}
}
RetryUpgradeMediums(client, endpoint_id, untried_mediums);
}
void BwuManager::RetryUpgradeMediums(ClientProxy* client,
const std::string& endpoint_id,
std::vector<Medium> upgrade_mediums) {
Medium next_medium = ChooseBestUpgradeMedium(upgrade_mediums);
// If current medium is not WiFi and we have not succeeded with upgrading
// yet, retry upgrade.
Medium current_medium = GetEndpointMedium(endpoint_id);
if (current_medium != Medium::WIFI_LAN &&
(next_medium == current_medium || next_medium == Medium::UNKNOWN_MEDIUM ||
upgrade_mediums.empty())) {
RetryUpgradesAfterDelay(client, endpoint_id);
return;
}
// Attempt to set the new upgrade medium.
if (!SetCurrentBwuHandler(next_medium)) {
NEARBY_LOG(
INFO,
"BwuManager failed to attempt a new bandwidth upgrade for endpoint %s "
"because we couldn't set a new bandwidth upgrade medium.",
endpoint_id.c_str());
return;
}
// Now that we've successfully picked a new upgrade medium to try,
// re-initiate the bandwidth upgrade.
NEARBY_LOG(INFO,
"BwuManager is attempting to upgrade endpoint %s again with a new "
" bandwidth upgrade medium.",
endpoint_id.c_str());
InitiateBwuForEndpoint(client, endpoint_id);
}
std::vector<Medium> BwuManager::StripOutUnavailableMediums(
const std::vector<Medium>& mediums) {
std::vector<Medium> available_mediums;
for (Medium m : mediums) {
bool available = false;
switch (m) {
case Medium::WIFI_LAN:
available = mediums_->GetWifiLan().IsAvailable();
break;
case Medium::WEB_RTC:
available = mediums_->GetWebRtc().IsAvailable();
break;
case Medium::BLUETOOTH:
available = mediums_->GetBluetoothClassic().IsAvailable();
break;
default:
break;
}
if (available) {
available_mediums.push_back(m);
}
}
return available_mediums;
}
// Returns the optimal medium supported by both devices.
// Each medium in the passed in list is checked for its availability with the
// medium_manager_ to ensure that the chosen upgrade medium is supported and
// available locally before continuing the upgrade. Once we pick a medium, all
// future connections will use it too. eg. If we chose Wifi LAN, we'll attempt
// to upgrade the 2nd, 3rd, etc remote endpoints with Wifi LAN even if they're
// on a different network (or had a better medium). This is a quick and easy
// way to prevent mediums, like Wifi Hotspot, from interfering with active
// connections (although it's suboptimal for bandwidth throughput). When all
// endpoints disconnect, we reset the bandwidth upgrade medium.
Medium BwuManager::ChooseBestUpgradeMedium(const std::vector<Medium>& mediums) {
auto available_mediums = StripOutUnavailableMediums(mediums);
if (medium_ == Medium::UNKNOWN_MEDIUM) {
if (!available_mediums.empty()) {
// Case 1: This is our first time upgrading, and we have at least one
// supported medium to choose from. Return the first medium in the list,
// since they are ordered by preference.
return available_mediums[0];
}
// Case 2: This is our first time upgrading, but there are no available
// upgrade mediums. Fall through to returning UNKNOWN_MEDIUM at the
// bottom.
NEARBY_LOG(
INFO,
"Current upgrade medium is unset, but there are no common supported "
"upgrade mediums.");
} else {
// Case 3: We have already upgraded, and there is a list of supported
// mediums to check against. Return the current upgrade medium if it's in
// the supported list.
if (std::find(available_mediums.begin(), available_mediums.end(),
medium_) != available_mediums.end()) {
return medium_;
}
// Case 4: We have already upgraded, but the current medium is not
// supported by the remote endpoint (it's not in the list, or the list is
// empty). Fall through and return Medium.UNKNOWN_MEDIUM because we cannot
// continue with the current upgrade medium, and we are not allowed to
// switch.
NEARBY_LOG(
INFO,
"Current upgrade medium %d is not supported by the remote endpoint",
medium_);
}
return Medium::UNKNOWN_MEDIUM;
}
void BwuManager::RetryUpgradesAfterDelay(ClientProxy* client,
const std::string& endpoint_id) {
absl::Duration delay = CalculateNextRetryDelay(endpoint_id);
CancelRetryUpgradeAlarm(endpoint_id);
CancelableAlarm alarm(
"BWU alarm",
[this, client, endpoint_id]() {
RunOnBwuManagerThread([this, client, endpoint_id]() {
if (!client->IsConnectedToEndpoint(endpoint_id)) {
return;
}
RetryUpgradeMediums(
client, endpoint_id,
client->GetUpgradeMediums(endpoint_id).GetMediums(true));
});
},
delay, &alarm_executor_);
retry_upgrade_alarms_.emplace(endpoint_id,
std::make_pair(std::move(alarm), delay));
NEARBY_LOGS(INFO) << "Retry bandwidth upgrade after " << delay;
}
absl::Duration BwuManager::CalculateNextRetryDelay(
const std::string& endpoint_id) {
auto item = retry_upgrade_alarms_.find(endpoint_id);
auto initial_delay = config_.bandwidth_upgrade_retry_delay;
auto delay = item == retry_upgrade_alarms_.end()
? initial_delay
: item->second.second + initial_delay;
return std::min(delay, config_.bandwidth_upgrade_retry_max_delay);
}
void BwuManager::CancelRetryUpgradeAlarm(const std::string& endpoint_id) {
auto item = retry_upgrade_alarms_.extract(endpoint_id);
if (item.empty()) return;
auto& pair = item.mapped();
pair.first.Cancel();
}
void BwuManager::CancelAllRetryUpgradeAlarms() {
for (const auto& item : retry_upgrade_alarms_) {
const std::string& endpoint_id = item.first;
CancelRetryUpgradeAlarm(endpoint_id);
}
}
Medium BwuManager::GetEndpointMedium(const std::string& endpoint_id) {
auto channel = channel_manager_->GetChannelForEndpoint(endpoint_id);
return channel == nullptr ? Medium::UNKNOWN_MEDIUM : channel->GetMedium();
}
} // namespace connections
} // namespace nearby
} // namespace location
cpp/core/internal/bwu_manager.h
+1 -1
View File
@@ -81,7 +81,7 @@ class BwuManager : public EndpointManager::FrameProcessor {
// @EndpointManagerReaderThread
void OnEndpointDisconnect(ClientProxy* client_proxy,
const std::string& endpoint_id,
CountDownLatch* barrier) override;
CountDownLatch barrier) override;
void Shutdown();
private:
cpp/core/internal/bwu_manager_test.cc
+7
View File
@@ -6,8 +6,10 @@
#include "core/internal/endpoint_channel_manager.h"
#include "core/internal/endpoint_manager.h"
#include "core/internal/mediums/mediums.h"
#include "platform/public/system_clock.h"
#include "gmock/gmock.h"
#include "gtest/gtest.h"
#include "absl/time/time.h"
namespace location {
namespace nearby {
@@ -19,6 +21,10 @@ TEST(BwuManagerTest, CanCreateInstance) {
EndpointChannelManager ecm;
EndpointManager em{&ecm};
BwuManager bwu_manager{mediums, em, ecm, {}, {}};
SystemClock::Sleep(absl::Seconds(3));
bwu_manager.Shutdown();
}
TEST(BwuManagerTest, CanInitiateBwu) {
@@ -31,6 +37,7 @@ TEST(BwuManagerTest, CanInitiateBwu) {
// Method returns void, so we just verify we did not SEGFAULT while calling.
bwu_manager.InitiateBwuForEndpoint(&client, endpoint_id);
SystemClock::Sleep(absl::Seconds(3));
bwu_manager.Shutdown();
}
cpp/core/internal/endpoint_channel_manager.cc
+13
View File
@@ -2,14 +2,22 @@
#include <memory>
#include "core/internal/offline_frames.h"
#include "proto/connections/offline_wire_formats.pb.h"
#include "platform/public/logging.h"
#include "platform/public/mutex.h"
#include "platform/public/mutex_lock.h"
#include "platform/public/system_clock.h"
#include "absl/time/time.h"
namespace location {
namespace nearby {
namespace connections {
namespace {
const absl::Duration kDataTransferDelay = absl::Milliseconds(500);
}
EndpointChannelManager::~EndpointChannelManager() {
MutexLock lock(&mutex_);
channel_state_.DestroyAll();
@@ -118,6 +126,11 @@ bool EndpointChannelManager::ChannelState::RemoveEndpoint(
auto item = endpoints_.find(endpoint_id);
if (item == endpoints_.end()) return false;
item->second.disconnect_reason = reason;
auto channel = item->second.channel;
if (channel) {
channel->Write(parser::ForDisconnection());
SystemClock::Sleep(kDataTransferDelay);
}
endpoints_.erase(item);
return true;
}
cpp/core/internal/endpoint_manager.cc
+1 -1
View File
@@ -473,7 +473,7 @@ void EndpointManager::WaitForEndpointDisconnectionProcessing(
NEARBY_LOGS(INFO) << "processor=" << processor << "; type=" << item.first;
if (processor) {
valid++;
processor->OnEndpointDisconnect(client, endpoint_id, &barrier);
processor->OnEndpointDisconnect(client, endpoint_id, barrier);
} else {
barrier.CountDown();
}
cpp/core/internal/endpoint_manager.h
+1 -1
View File
@@ -70,7 +70,7 @@ class EndpointManager {
// @EndpointManagerThread
virtual void OnEndpointDisconnect(ClientProxy* client,
const std::string& endpoint_id,
CountDownLatch* barrier) = 0;
CountDownLatch barrier) = 0;
};
explicit EndpointManager(EndpointChannelManager* manager);
cpp/core/internal/endpoint_manager_test.cc
+1 -1
View File
@@ -71,7 +71,7 @@ class MockFrameProcessor : public EndpointManager::FrameProcessor {
MOCK_METHOD(void, OnEndpointDisconnect,
(ClientProxy * client, const std::string& endpoint_id,
CountDownLatch* barrier),
CountDownLatch barrier),
(override));
};
cpp/core/internal/mediums/webrtc.h.orig
-174
View File
@@ -1,174 +0,0 @@
#ifndef CORE_V2_INTERNAL_MEDIUMS_WEBRTC_H_
#define CORE_V2_INTERNAL_MEDIUMS_WEBRTC_H_
#include <memory>
#include <string>
#include "core_v2/internal/mediums/webrtc/connection_flow.h"
#include "core_v2/internal/mediums/webrtc/data_channel_listener.h"
#include "core_v2/internal/mediums/webrtc/local_ice_candidate_listener.h"
#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 "proto/connections/offline_wire_formats.pb.h"
#include "proto/connections/offline_wire_formats.pb.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/cancelable_alarm.h"
#include "platform_v2/public/future.h"
#include "platform_v2/public/mutex.h"
#include "platform_v2/public/scheduled_executor.h"
#include "platform_v2/public/single_thread_executor.h"
#include "platform_v2/public/webrtc.h"
#include "location/nearby/mediums/proto/web_rtc_signaling_frames.pb.h"
#include "webrtc/api/data_channel_interface.h"
#include "webrtc/api/jsep.h"
#include "webrtc/api/scoped_refptr.h"
namespace location {
namespace nearby {
namespace connections {
namespace mediums {
// Callback that is invoked when a new connection is accepted.
struct AcceptedConnectionCallback {
std::function<void(WebRtcSocketWrapper socket)> accepted_cb =
DefaultCallback<WebRtcSocketWrapper>();
};
// Entry point for connecting a data channel between two devices via WebRtc.
class WebRtc {
public:
WebRtc();
~WebRtc();
// Returns if WebRtc is available as a medium for nearby to transport data.
// Runs on @MainThread.
bool IsAvailable();
// Returns if the device is ready to accept connections from remote devices.
// Runs on @MainThread.
bool IsAcceptingConnections() ABSL_LOCKS_EXCLUDED(mutex_);
// Prepares the device to accept incoming WebRtc connections. Returns a
// boolean value indicating if the device has started accepting connections.
// Runs on @MainThread.
bool StartAcceptingConnections(const PeerId& self_id,
const LocationHint& location_hint,
AcceptedConnectionCallback callback)
ABSL_LOCKS_EXCLUDED(mutex_);
// Prevents device from accepting future connections until
// StartAcceptingConnections() is called.
// Runs on @MainThread.
void StopAcceptingConnections() ABSL_LOCKS_EXCLUDED(mutex_);
// Initiates a WebRtc connection with peer device identified by |peer_id|.
// Runs on @MainThread.
WebRtcSocketWrapper Connect(const PeerId& peer_id,
const LocationHint& location_hint)
ABSL_LOCKS_EXCLUDED(mutex_);
private:
enum class Role {
kNone = 0,
kOfferer = 1,
kAnswerer = 2,
};
bool InitWebRtcFlow(Role role, const PeerId& self_id,
const LocationHint& location_hint)
ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
Future<WebRtcSocketWrapper> ListenForWebRtcSocketFuture(
Future<rtc::scoped_refptr<webrtc::DataChannelInterface>>
data_channel_future,
AcceptedConnectionCallback callback);
WebRtcSocketWrapper CreateWebRtcSocketWrapper(
rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel);
LocalIceCandidateListener GetLocalIceCandidateListener();
void OnLocalIceCandidate(
const webrtc::IceCandidateInterface* local_ice_candidate);
DataChannelListener GetDataChannelListener();
void OnDataChannelClosed();
void OnDataChannelMessageReceived(const ByteArray& message);
void OnDataChannelBufferedAmountChanged();
// Runs on @MainThread and |single_thread_executor_|.
bool SetLocalSessionDescription(SessionDescriptionWrapper sdp)
ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
// Runs on |single_thread_executor_|.
bool IsSignaling() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
// Runs on |single_thread_executor_|.
void ProcessSignalingMessage(const ByteArray& message)
ABSL_LOCKS_EXCLUDED(mutex_);
// Runs on |single_thread_executor_|.
void SendOfferAndIceCandidatesToPeer() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
// Runs on |single_thread_executor_|.
void SendAnswerToPeer() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
// Runs on @MainThread and |single_thread_executor_|.
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 DisconnectLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
void LogAndShutdownSignaling(const std::string& error_message)
ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
// Runs on @MainThread and |single_thread_executor_|.
void ShutdownSignaling() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
// Runs on @MainThread and |single_thread_executor_|.
void ShutdownWebRtcSocket() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
// Runs on @MainThread and |single_thread_executor_|.
void ShutdownIceCandidateCollection();
void OffloadFromSignalingThread(Runnable runnable);
// Runs on |restart_receive_messages_executor_|.
void RestartReceiveMessages(const LocationHint& location_hint)
ABSL_LOCKS_EXCLUDED(mutex_);
Mutex mutex_;
Role role_ ABSL_GUARDED_BY(mutex_) = Role::kNone;
PeerId self_id_ ABSL_GUARDED_BY(mutex_);
PeerId peer_id_ ABSL_GUARDED_BY(mutex_);
ByteArray pending_local_offer_ ABSL_GUARDED_BY(mutex_);
std::vector<::location::nearby::mediums::IceCandidate>
pending_local_ice_candidates_ ABSL_GUARDED_BY(mutex_);
WebRtcMedium medium_;
std::unique_ptr<ConnectionFlow> connection_flow_;
std::unique_ptr<WebRtcSignalingMessenger> signaling_messenger_
ABSL_GUARDED_BY(mutex_);
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
} // namespace connections
} // namespace nearby
} // namespace location
#endif // CORE_V2_INTERNAL_MEDIUMS_WEBRTC_H_
cpp/core/internal/offline_frames.h
+1
View File
@@ -59,6 +59,7 @@ ByteArray ForBwuLastWrite();
ByteArray ForBwuSafeToClose();
ByteArray ForKeepAlive();
ByteArray ForDisconnection();
UpgradePathInfo::Medium MediumToUpgradePathInfoMedium(Medium medium);
Medium UpgradePathInfoMediumToMedium(UpgradePathInfo::Medium medium);
cpp/core/internal/offline_service_controller_test.cc.orig
-374
View File
@@ -1,374 +0,0 @@
#include "core_v2/internal/offline_service_controller.h"
#include <array>
#include "core_v2/internal/offline_simulation_user.h"
#include "platform_v2/base/medium_environment.h"
#include "platform_v2/base/output_stream.h"
#include "platform_v2/public/count_down_latch.h"
#include "platform_v2/public/logging.h"
#include "platform_v2/public/pipe.h"
#include "platform_v2/public/system_clock.h"
#include "gmock/gmock.h"
#include "gtest/gtest.h"
namespace location {
namespace nearby {
namespace connections {
namespace {
using ::testing::Eq;
constexpr std::array<char, 6> kFakeMacAddress = {'a', 'b', 'c', 'd', 'e', 'f'};
constexpr absl::string_view kServiceId = "service-id";
constexpr absl::string_view kDeviceA = "device-a";
constexpr absl::string_view kDeviceB = "device-b";
constexpr absl::string_view kMessage = "message";
constexpr absl::Duration kProgressTimeout = absl::Milliseconds(1000);
constexpr absl::Duration kDefaultTimeout = absl::Milliseconds(1000);
constexpr absl::Duration kDisconnectTimeout = absl::Milliseconds(15000);
constexpr BooleanMediumSelector kTestCases[] = {
BooleanMediumSelector{
.bluetooth = true,
},
BooleanMediumSelector{
.wifi_lan = true,
},
BooleanMediumSelector{
.bluetooth = true,
.wifi_lan = true,
},
};
class OfflineServiceControllerTest
: public ::testing::TestWithParam<BooleanMediumSelector> {
protected:
OfflineServiceControllerTest() { env_.Stop(); }
bool SetupConnection(OfflineSimulationUser& user_a,
OfflineSimulationUser& user_b) {
user_a.StartAdvertising(std::string(kServiceId), &connect_latch_);
user_b.StartDiscovery(std::string(kServiceId), &discover_latch_);
EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result());
EXPECT_EQ(user_b.GetDiscovered().service_id, kServiceId);
EXPECT_EQ(user_b.GetDiscovered().endpoint_info, user_a.GetInfo());
EXPECT_FALSE(user_b.GetDiscovered().endpoint_id.empty());
NEARBY_LOG(INFO, "EP-B: [discovered] %s",
user_b.GetDiscovered().endpoint_id.c_str());
user_b.RequestConnection(&connect_latch_);
EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result());
EXPECT_FALSE(user_a.GetDiscovered().endpoint_id.empty());
NEARBY_LOG(INFO, "EP-A: [discovered] %s",
user_a.GetDiscovered().endpoint_id.c_str());
NEARBY_LOG(INFO, "Both users discovered their peers.");
user_a.AcceptConnection(&accept_latch_);
user_b.AcceptConnection(&accept_latch_);
EXPECT_TRUE(accept_latch_.Await(kDefaultTimeout).result());
NEARBY_LOG(INFO, "Both users reached connected state.");
return user_a.IsConnected() && user_b.IsConnected();
}
CountDownLatch discover_latch_{1};
CountDownLatch lost_latch_{1};
CountDownLatch connect_latch_{2};
CountDownLatch accept_latch_{2};
CountDownLatch payload_latch_{1};
MediumEnvironment& env_ = MediumEnvironment::Instance();
};
TEST_P(OfflineServiceControllerTest, CanCreateOne) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA, GetParam());
env_.Stop();
}
TEST_P(OfflineServiceControllerTest, CanCreateMany) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA, GetParam());
OfflineSimulationUser user_b(kDeviceB, GetParam());
env_.Stop();
}
TEST_P(OfflineServiceControllerTest, CanStartAdvertising) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA, GetParam());
OfflineSimulationUser user_b(kDeviceB, GetParam());
EXPECT_FALSE(user_a.IsAdvertising());
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(user_a.IsAdvertising());
env_.Stop();
}
TEST_P(OfflineServiceControllerTest, CanStartDiscoveryBeforeAdvertising) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA, GetParam());
OfflineSimulationUser user_b(kDeviceB, GetParam());
EXPECT_FALSE(user_b.IsDiscovering());
EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(user_b.IsDiscovering());
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_P(OfflineServiceControllerTest, CanStartDiscoveryAfterAdvertising) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA, GetParam());
OfflineSimulationUser user_b(kDeviceB, GetParam());
EXPECT_FALSE(user_b.IsDiscovering());
EXPECT_FALSE(user_b.IsAdvertising());
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr),
Eq(Status{Status::kSuccess}));
EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(user_a.IsAdvertising());
EXPECT_TRUE(user_b.IsDiscovering());
EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_P(OfflineServiceControllerTest, CanStopAdvertising) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA, GetParam());
OfflineSimulationUser user_b(kDeviceB, GetParam());
EXPECT_FALSE(user_a.IsAdvertising());
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(user_a.IsAdvertising());
user_a.StopAdvertising();
EXPECT_FALSE(user_a.IsAdvertising());
EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_,
&lost_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(user_b.IsDiscovering());
auto discover_none = discover_latch_.Await(kDefaultTimeout).GetResult();
if (!discover_none) {
EXPECT_TRUE(true);
} else {
// There are rare cases (1/1000) that advertisment data has been captured by
// discovery device before advertising is stopped. So we need to check if
// lost_cb has grabbed the event in the end to prove the advertising service
// is stopped.
EXPECT_TRUE(lost_latch_.Await(kDefaultTimeout).result());
}
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_P(OfflineServiceControllerTest, CanStopDiscovery) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA, GetParam());
OfflineSimulationUser user_b(kDeviceB, GetParam());
EXPECT_FALSE(user_b.IsDiscovering());
EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(user_b.IsDiscovering());
user_b.StopDiscovery();
EXPECT_FALSE(user_b.IsDiscovering());
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr),
Eq(Status{Status::kSuccess}));
EXPECT_FALSE(discover_latch_.Await(kDefaultTimeout).result());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_P(OfflineServiceControllerTest, CanConnect) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA, GetParam());
OfflineSimulationUser user_b(kDeviceB, GetParam());
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), &connect_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result());
EXPECT_THAT(user_b.RequestConnection(&connect_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_P(OfflineServiceControllerTest, CanAcceptConnection) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA, GetParam());
OfflineSimulationUser user_b(kDeviceB, GetParam());
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), &connect_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result());
EXPECT_THAT(user_b.RequestConnection(&connect_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result());
EXPECT_THAT(user_a.AcceptConnection(&accept_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_THAT(user_b.AcceptConnection(&accept_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(accept_latch_.Await(kDefaultTimeout).result());
EXPECT_TRUE(user_a.IsConnected());
EXPECT_TRUE(user_b.IsConnected());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_P(OfflineServiceControllerTest, CanRejectConnection) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA, GetParam());
OfflineSimulationUser user_b(kDeviceB, GetParam());
CountDownLatch reject_latch(1);
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), &connect_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result());
EXPECT_THAT(user_b.RequestConnection(&connect_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result());
user_a.ExpectRejectedConnection(reject_latch);
EXPECT_THAT(user_b.RejectConnection(nullptr), Eq(Status{Status::kSuccess}));
EXPECT_TRUE(reject_latch.Await(kDefaultTimeout).result());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_P(OfflineServiceControllerTest, CanSendBytePayload) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA, GetParam());
OfflineSimulationUser user_b(kDeviceB, GetParam());
ASSERT_TRUE(SetupConnection(user_a, user_b));
ByteArray message(std::string{kMessage});
user_a.SendPayload(Payload(message));
user_b.ExpectPayload(payload_latch_);
EXPECT_TRUE(payload_latch_.Await(kDefaultTimeout).result());
EXPECT_EQ(user_b.GetPayload().AsBytes(), message);
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_P(OfflineServiceControllerTest, CanSendStreamPayload) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA, GetParam());
OfflineSimulationUser user_b(kDeviceB, GetParam());
ASSERT_TRUE(SetupConnection(user_a, user_b));
ByteArray message(std::string{kMessage});
auto pipe = std::make_shared<Pipe>();
OutputStream& tx = pipe->GetOutputStream();
user_a.SendPayload(Payload([pipe]() -> InputStream& {
return pipe->GetInputStream(); // NOLINT
}));
user_b.ExpectPayload(payload_latch_);
tx.Write(message);
EXPECT_TRUE(payload_latch_.Await(kDefaultTimeout).result());
EXPECT_NE(user_b.GetPayload().AsStream(), nullptr);
InputStream& rx = *user_b.GetPayload().AsStream();
ASSERT_TRUE(user_b.WaitForProgress(
[size = message.size()](const PayloadProgressInfo& info) -> bool {
return info.bytes_transferred >= size;
},
kProgressTimeout));
EXPECT_EQ(rx.Read(Pipe::kChunkSize).result(), message);
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_P(OfflineServiceControllerTest, CanCancelStreamPayload) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA, GetParam());
OfflineSimulationUser user_b(kDeviceB, GetParam());
ASSERT_TRUE(SetupConnection(user_a, user_b));
ByteArray message(std::string{kMessage});
auto pipe = std::make_shared<Pipe>();
OutputStream& tx = pipe->GetOutputStream();
user_a.SendPayload(Payload([pipe]() -> InputStream& {
return pipe->GetInputStream(); // NOLINT
}));
user_b.ExpectPayload(payload_latch_);
tx.Write(message);
EXPECT_TRUE(payload_latch_.Await(kDefaultTimeout).result());
EXPECT_NE(user_b.GetPayload().AsStream(), nullptr);
InputStream& rx = *user_b.GetPayload().AsStream();
ASSERT_TRUE(user_b.WaitForProgress(
[size = message.size()](const PayloadProgressInfo& info) -> bool {
return info.bytes_transferred >= size;
},
kProgressTimeout));
EXPECT_EQ(rx.Read(Pipe::kChunkSize).result(), message);
user_b.CancelPayload();
int count = 0;
while (true) {
count++;
if (!tx.Write(message).Ok()) break;
SystemClock::Sleep(kDefaultTimeout);
}
EXPECT_TRUE(user_a.WaitForProgress(
[](const PayloadProgressInfo& info) -> bool {
return info.status == PayloadProgressInfo::Status::kCanceled;
},
kProgressTimeout));
EXPECT_LT(count, 10);
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_P(OfflineServiceControllerTest, CanDisconnect) {
env_.Start();
CountDownLatch disconnect_latch(1);
OfflineSimulationUser user_a(kDeviceA, GetParam());
OfflineSimulationUser user_b(kDeviceB, GetParam());
ASSERT_TRUE(SetupConnection(user_a, user_b));
NEARBY_LOGS(INFO) << "Disconnecting";
user_b.ExpectDisconnect(disconnect_latch);
user_b.Disconnect();
EXPECT_TRUE(disconnect_latch.Await(kDisconnectTimeout).result());
NEARBY_LOGS(INFO) << "Disconnected";
EXPECT_FALSE(user_b.IsConnected());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
INSTANTIATE_TEST_SUITE_P(ParametrisedOfflineServiceControllerTest,
OfflineServiceControllerTest,
::testing::ValuesIn(kTestCases));
// Verifies that InjectEndpoint() can be run successfully; does not test the
// full connection flow given that normal discovery/advertisement is skipped.
// Note: Not parameterized because InjectEndpoint only works over Bluetooth.
TEST_F(OfflineServiceControllerTest, InjectEndpoint) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA,
BooleanMediumSelector{.bluetooth = true});
EXPECT_THAT(user_a.StartDiscovery(std::string(kServiceId),
/*found_latch=*/nullptr),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(user_a.IsDiscovering());
user_a.InjectEndpoint(
std::string(kServiceId),
OutOfBandConnectionMetadata{
.medium = Medium::BLUETOOTH,
.remote_bluetooth_mac_address = ByteArray(kFakeMacAddress),
});
user_a.Stop();
env_.Stop();
}
} // namespace
} // namespace connections
} // namespace nearby
} // namespace location
cpp/core/internal/payload_manager.cc
+4 -4
View File
@@ -388,12 +388,12 @@ void PayloadManager::OnIncomingFrame(
void PayloadManager::OnEndpointDisconnect(ClientProxy* client,
const std::string& endpoint_id,
CountDownLatch* barrier) {
CountDownLatch barrier) {
if (shutdown_.Get()) {
if (barrier) barrier->CountDown();
barrier.CountDown();
return;
}
RunOnStatusUpdateThread([this, client, endpoint_id, barrier]() {
RunOnStatusUpdateThread([this, client, endpoint_id, barrier]() mutable {
// Iterate through all our payloads and look for payloads associated
// with this endpoint.
MutexLock lock(&mutex_);
@@ -423,7 +423,7 @@ void PayloadManager::OnEndpointDisconnect(ClientProxy* client,
client->OnPayloadProgress(endpoint_id, update);
}
barrier->CountDown();
barrier.CountDown();
});
}
cpp/core/internal/payload_manager.h
+1 -1
View File
@@ -46,7 +46,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
// @EndpointManagerThread
void OnEndpointDisconnect(ClientProxy* client, const std::string& endpoint_id,
CountDownLatch* barrier) override;
CountDownLatch barrier) override;
void DisconnectFromEndpointManager();
cpp/core/internal/simulation_user.h
+1
View File
@@ -48,6 +48,7 @@ class SimulationUser {
void Stop() {
pm_.DisconnectFromEndpointManager();
mgr_.DisconnectFromEndpointManager();
bwu_.Shutdown();
}
// Calls PcpManager::StartAdvertising.
cpp/core/internal/wifi_lan_bwu_handler.cc
+115
View File
@@ -0,0 +1,115 @@
#include "core/internal/wifi_lan_bwu_handler.h"
#include <locale>
#include <string>
#include "core/internal/client_proxy.h"
#include "core/internal/mediums/utils.h"
#include "core/internal/offline_frames.h"
#include "core/internal/wifi_lan_endpoint_channel.h"
#include "platform/public/wifi_lan.h"
#include "absl/functional/bind_front.h"
namespace location {
namespace nearby {
namespace connections {
WifiLanBwuHandler::WifiLanBwuHandler(Mediums& mediums,
EndpointChannelManager& channel_manager,
BwuNotifications notifications)
: BaseBwuHandler(channel_manager, std::move(notifications)),
mediums_(mediums) {}
// Called by BWU initiator. Set up WifiLan upgraded medium for this endpoint,
// and returns a upgrade path info (ip address, port) for remote party to
// perform discovery.
ByteArray WifiLanBwuHandler::InitializeUpgradedMediumForEndpoint(
ClientProxy* client, const std::string& service_id,
const std::string& endpoint_id) {
// Use wrapped service ID to avoid have the same ID with the one for
// startAdvertising. Otherwise, the listening request would be ignored because
// the medium already start accepting the connection because the client not
// stop the advertising yet.
std::string upgrade_service_id = Utils::WrapUpgradeServiceId(service_id);
if (!wifi_lan_medium_.IsAcceptingConnections(upgrade_service_id)) {
if (!wifi_lan_medium_.StartAcceptingConnections(
upgrade_service_id,
{
.accepted_cb = absl::bind_front(
&WifiLanBwuHandler::OnIncomingWifiLanConnection, this,
client),
})) {
NEARBY_LOG(ERROR,
"WifiLanBwuHandler couldn't initiate the WifiLan upgrade for "
"endpoint %s because it failed to start listening for "
"incoming WifiLan connections.",
endpoint_id.c_str());
return {};
}
NEARBY_LOG(INFO,
"WifiLanBwuHandler successfully started listening for incoming "
"WifiLan connections while upgrading endpoint %s",
endpoint_id.c_str());
}
// cache service ID to revert
active_service_ids_.emplace(upgrade_service_id);
// TODO(b/169303360): Implements wifiLanCredntials for wif_lan_medium to
// get ip_address and port.
std::string ip_addresss;
std::int32_t port = 0;
return parser::ForBwuWifiLanPathAvailable(ip_addresss, port);
}
void WifiLanBwuHandler::Revert() {
for (const std::string& service_id : active_service_ids_) {
wifi_lan_medium_.StopAcceptingConnections(service_id);
}
active_service_ids_.clear();
NEARBY_LOG(INFO, "WifiLanBwuHandler successfully reverted all states.");
}
// Called by BWU target. Retrieves a new medium info from incoming message,
// and establishes connection over WifiLan using this info.
std::unique_ptr<EndpointChannel>
WifiLanBwuHandler::CreateUpgradedEndpointChannel(
ClientProxy* client, const std::string& service_id,
const std::string& endpoint_id, const UpgradePathInfo& upgrade_path_info) {
// TODO(b/169303360): Implements connect WifiLan over ip address and port.
WifiLanSocket socket;
// Create a new WifiLanEndpointChannel.
auto channel = std::make_unique<WifiLanEndpointChannel>(service_id, socket);
if (channel == nullptr) {
socket.Close();
NEARBY_LOG(ERROR,
"WifiLanBwuHandler failed to create new EndpointChannel for "
"outgoing socket %p, aborting upgrade.",
&socket.GetImpl());
}
return channel;
}
// Accept Connection Callback.
void WifiLanBwuHandler::OnIncomingWifiLanConnection(
ClientProxy* client, WifiLanSocket socket,
const std::string& upgrade_service_id) {
std::string service_id = Utils::UnwrapUpgradeServiceId(upgrade_service_id);
auto channel = std::make_unique<WifiLanEndpointChannel>(service_id, socket);
auto wifi_lan_socket =
std::make_unique<WifiLanIncomingSocket>(service_id, socket);
std::unique_ptr<IncomingSocketConnection> connection(
new IncomingSocketConnection{std::move(wifi_lan_socket),
std::move(channel)});
bwu_notifications_.incoming_connection_cb(client, std::move(connection));
}
} // namespace connections
} // namespace nearby
} // namespace location
cpp/core/internal/wifi_lan_bwu_handler.h
+65
View File
@@ -0,0 +1,65 @@
#ifndef CORE_INTERNAL_WIFI_LAN_BWU_HANDLER_H_
#define CORE_INTERNAL_WIFI_LAN_BWU_HANDLER_H_
#include "core/internal/base_bwu_handler.h"
#include "core/internal/client_proxy.h"
#include "core/internal/endpoint_channel_manager.h"
#include "core/internal/mediums/mediums.h"
namespace location {
namespace nearby {
namespace connections {
// Defines the set of methods that need to be implemented to handle the
// per-Medium-specific operations needed to upgrade an EndpointChannel.
class WifiLanBwuHandler : public BaseBwuHandler {
public:
WifiLanBwuHandler(Mediums& mediums, EndpointChannelManager& channel_manager,
BwuNotifications notifications);
~WifiLanBwuHandler() override = default;
private:
ByteArray InitializeUpgradedMediumForEndpoint(
ClientProxy* client, const std::string& service_id,
const std::string& endpoint_id) override;
void Revert() override;
std::unique_ptr<EndpointChannel> CreateUpgradedEndpointChannel(
ClientProxy* client, const std::string& service_id,
const std::string& endpoint_id,
const UpgradePathInfo& upgrade_path_info) override;
Medium GetUpgradeMedium() const override { return Medium::WIFI_LAN; }
void OnEndpointDisconnect(ClientProxy* client,
const std::string& endpoint_id) override {}
void OnIncomingWifiLanConnection(ClientProxy* client, WifiLanSocket socket,
const std::string& upgrade_service_id);
class WifiLanIncomingSocket : public BwuHandler::IncomingSocket {
public:
explicit WifiLanIncomingSocket(const std::string& name,
WifiLanSocket socket)
: name_(name), socket_(socket) {}
~WifiLanIncomingSocket() override = default;
std::string ToString() override { return name_; }
void Close() override { socket_.Close(); }
private:
std::string name_;
WifiLanSocket socket_;
};
Mediums& mediums_;
WifiLan& wifi_lan_medium_{mediums_.GetWifiLan()};
absl::flat_hash_set<std::string> active_service_ids_;
};
} // namespace connections
} // namespace nearby
} // namespace location
#endif // CORE_INTERNAL_WIFI_LAN_BWU_HANDLER_H_
cpp/core/internal/wifi_lan_service_info.cc
+5
View File
@@ -14,6 +14,11 @@ namespace location {
namespace nearby {
namespace connections {
// These definitions are necessary before C++17.
constexpr absl::string_view WifiLanServiceInfo::kKeyEndpointInfo;
constexpr std::uint32_t WifiLanServiceInfo::kServiceIdHashLength;
constexpr int WifiLanServiceInfo::kMaxEndpointInfoLength;
WifiLanServiceInfo::WifiLanServiceInfo(Version version, Pcp pcp,
absl::string_view endpoint_id,
const ByteArray& service_id_hash,
cpp/platform/impl/shared/BUILD.orig
-71
View File
@@ -1,71 +0,0 @@
cc_library(
name = "posix_lock",
srcs = [
"posix_lock.cc",
],
hdrs = [
"posix_lock.h",
],
visibility = [
"//googlemac/iPhone/Shared/Nearby/Connections:__subpackages__",
"//platform/impl:__subpackages__",
],
deps = [
"//platform/api",
],
)
cc_library(
name = "posix_condition_variable",
srcs = [
"posix_condition_variable.cc",
],
hdrs = [
"posix_condition_variable.h",
],
visibility = [
"//googlemac/iPhone/Shared/Nearby/Connections:__subpackages__",
"//platform/impl:__subpackages__",
],
deps = [
":posix_lock",
"//platform:types",
"//platform/api:condition_variable",
],
)
cc_library(
name = "atomic_boolean",
hdrs = ["atomic_boolean_impl.h"],
visibility = ["//platform/impl:__subpackages__"],
deps = ["//platform/api"],
)
cc_library(
name = "file",
srcs = ["file_impl.cc"],
hdrs = ["file_impl.h"],
visibility = [
"//googlemac/iPhone/Shared/Nearby/Connections:__subpackages__",
"//core:__subpackages__",
"//platform/impl:__subpackages__",
],
deps = [
"//platform:types",
"//platform/api",
],
)
cc_test(
name = "file_test",
timeout = "short",
srcs = [
"file_impl_test.cc",
],
deps = [
":file",
"//file/util:temp_path",
"//testing/base/public:gunit_main",
"//absl/strings",
],
)
cpp/platform/public/BUILD.orig
-118
View File
@@ -1,118 +0,0 @@
cc_library(
name = "types",
srcs = [
"pipe.cc",
],
hdrs = [
"atomic_boolean.h",
"atomic_reference.h",
"cancelable.h",
"cancelable_alarm.h",
"condition_variable.h",
"count_down_latch.h",
"crypto.h",
"file.h",
"future.h",
"logging.h",
"multi_thread_executor.h",
"mutex.h",
"mutex_lock.h",
"pipe.h",
"scheduled_executor.h",
"settable_future.h",
"single_thread_executor.h",
"submittable_executor.h",
"system_clock.h",
],
visibility = [
"//core_v2:__subpackages__",
"//platform_v2/base:__pkg__",
],
deps = [
":logging",
"//platform_v2/api:platform",
"//platform_v2/api:types",
"//platform_v2/base",
"//platform_v2/base:logging",
"//platform_v2/base:util",
"//absl/base:core_headers",
"//absl/container:flat_hash_map",
"//absl/time",
],
)
cc_library(
name = "comm",
srcs = [
"ble.cc",
"bluetooth_classic.cc",
"wifi_lan.cc",
],
hdrs = [
"ble.h",
"bluetooth_adapter.h",
"bluetooth_classic.h",
"webrtc.h",
"wifi_lan.h",
],
visibility = ["//core_v2:__subpackages__"],
deps = [
":logging",
":types",
"//proto/connections:offline_wire_formats_portable_proto",
"//platform_v2/api:comm",
"//platform_v2/api:platform",
"//platform_v2/base",
"//absl/container:flat_hash_map",
"//absl/strings",
"//webrtc/api:libjingle_peerconnection_api",
],
)
cc_library(
name = "logging",
hdrs = [
"logging.h",
],
visibility = ["//core_v2:__subpackages__"],
deps = [
"//platform_v2/base:logging",
],
)
cc_test(
name = "public_test",
size = "small",
srcs = [
"atomic_boolean_test.cc",
"atomic_reference_test.cc",
"ble_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",
"future_test.cc",
"logging_test.cc",
"multi_thread_executor_test.cc",
"mutex_test.cc",
"pipe_test.cc",
"scheduled_executor_test.cc",
"single_thread_executor_test.cc",
"wifi_lan_test.cc",
],
shard_count = 16,
deps = [
":comm",
":logging",
":types",
"//platform_v2/base",
"//platform_v2/base:test_util",
"//platform_v2/impl/g3", # build_cleaner: keep
"//testing/base/public:gunit_main",
"//absl/strings",
"//absl/synchronization",
"//absl/time",
],
)
cpp/platform/public/count_down_latch.h
+3 -3
View File
@@ -20,8 +20,8 @@ class CountDownLatch final {
using Platform = api::ImplementationPlatform;
explicit CountDownLatch(int count)
: impl_(Platform::CreateCountDownLatch(count)) {}
CountDownLatch(CountDownLatch&&) = default;
CountDownLatch& operator=(CountDownLatch&&) = default;
CountDownLatch(const CountDownLatch&) = default;
CountDownLatch& operator=(const CountDownLatch&) = default;
~CountDownLatch() = default;
Exception Await() { return impl_->Await(); }
@@ -31,7 +31,7 @@ class CountDownLatch final {
void CountDown() { impl_->CountDown(); }
private:
std::unique_ptr<api::CountDownLatch> impl_;
std::shared_ptr<api::CountDownLatch> impl_;
};
} // namespace nearby