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

Merge branch 'master' into release

Browse Source
This commit is contained in:
Josh Nohle
2020-09-24 13:23:20 -07:00
parent b9ce9ac740 d4a3110624
commit a4adec6655
56 changed files with 1075 additions and 807 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/core/internal/mediums/webrtc/signaling_frames_test.cc
+3 -3
View File
@@ -110,7 +110,7 @@ TEST(SignalingFramesTest, EncodeValidOffer) {
TEST(SignalingFramesTest, DecodeValidOffer) {
location::nearby::mediums::WebRtcSignalingFrame frame;
proto2::TextFormat::ParseFromString(kOfferProto, &frame);
proto2::TextFormat::ParseFromStringPiece(kOfferProto, &frame);
Ptr<webrtc::SessionDescriptionInterface> decoded_offer = DecodeOffer(frame);
EXPECT_EQ(webrtc::SdpType::kOffer, decoded_offer->GetType());
@@ -134,7 +134,7 @@ TEST(SignalingFramesTest, EncodeValidAnswer) {
TEST(SignalingFramesTest, DecodeValidAnswer) {
location::nearby::mediums::WebRtcSignalingFrame frame;
proto2::TextFormat::ParseFromString(kAnswerProto, &frame);
proto2::TextFormat::ParseFromStringPiece(kAnswerProto, &frame);
Ptr<webrtc::SessionDescriptionInterface> decoded_answer = DecodeAnswer(frame);
EXPECT_EQ(webrtc::SdpType::kAnswer, decoded_answer->GetType());
@@ -177,7 +177,7 @@ TEST(SignalingFramesTest, DecodeValidIceCandidates) {
std::vector<location::nearby::mediums::IceCandidate> encoded_candidates_vec;
location::nearby::mediums::WebRtcSignalingFrame frame;
proto2::TextFormat::ParseFromString(kIceCandidatesProto, &frame);
proto2::TextFormat::ParseFromStringPiece(kIceCandidatesProto, &frame);
std::vector<ConstPtr<webrtc::IceCandidateInterface>> decoded_candidates =
DecodeIceCandidates(frame);
cpp/core_v2/internal/base_pcp_handler.cc
+89 -37
View File
@@ -23,6 +23,7 @@
#include "core_v2/internal/offline_frames.h"
#include "core_v2/internal/pcp_handler.h"
#include "core_v2/options.h"
#include "platform_v2/base/bluetooth_utils.h"
#include "platform_v2/public/logging.h"
#include "platform_v2/public/system_clock.h"
#include "securegcm/d2d_connection_context_v1.h"
@@ -43,11 +44,13 @@ constexpr absl::Duration BasePcpHandler::kRejectedConnectionCloseDelay;
BasePcpHandler::BasePcpHandler(Mediums* mediums,
EndpointManager* endpoint_manager,
EndpointChannelManager* channel_manager, Pcp pcp)
EndpointChannelManager* channel_manager,
BwuManager* bwu_manager, Pcp pcp)
: mediums_(mediums),
endpoint_manager_(endpoint_manager),
channel_manager_(channel_manager),
pcp_(pcp) {}
pcp_(pcp),
bwu_manager_(bwu_manager) {}
BasePcpHandler::~BasePcpHandler() {
NEARBY_LOGS(INFO) << "BasePcpHandler: going down; strategy="
@@ -77,23 +80,23 @@ Status BasePcpHandler::StartAdvertising(ClientProxy* client,
const ConnectionRequestInfo& info) {
Future<Status> response;
ConnectionOptions advertising_options = options.CompatibleOptions();
RunOnPcpHandlerThread(
[this, client, &service_id, &info, &advertising_options, &response]() {
auto result = StartAdvertisingImpl(
client, service_id, client->GetLocalEndpointId(),
info.endpoint_info, advertising_options);
if (!result.status.Ok()) {
response.Set(result.status);
return;
}
RunOnPcpHandlerThread([this, client, &service_id, &info, &advertising_options,
&response]() {
auto result =
StartAdvertisingImpl(client, service_id, client->GetLocalEndpointId(),
info.endpoint_info, advertising_options);
if (!result.status.Ok()) {
response.Set(result.status);
return;
}
// Now that we've succeeded, mark the client as advertising.
advertising_options_ = advertising_options;
advertising_listener_ = info.listener;
client->StartedAdvertising(service_id, GetStrategy(), info.listener,
absl::MakeSpan(result.mediums));
response.Set({Status::kSuccess});
});
// Now that we've succeeded, mark the client as advertising.
advertising_options_ = advertising_options;
advertising_listener_ = info.listener;
client->StartedAdvertising(service_id, GetStrategy(), info.listener,
absl::MakeSpan(result.mediums));
response.Set({Status::kSuccess});
});
return WaitForResult(
absl::StrCat("StartAdvertising(", std::string(info.endpoint_info), ")"),
client->GetClientId(), &response);
@@ -246,8 +249,8 @@ void BasePcpHandler::OnEncryptionSuccessRunnable(
.raw_authentication_token = raw_auth_token,
.is_incoming_connection = connection_info.is_incoming,
},
connection_info.options,
std::move(connection_info.channel), connection_info.listener);
connection_info.options, std::move(connection_info.channel),
connection_info.listener);
if (connection_info.result != nullptr) {
NEARBY_LOG(INFO, "Connection established; Finalising future OK");
@@ -332,14 +335,20 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
OnEndpointFound(client, webrtc_endpoint);
}
auto endpoints = GetDiscoveredEndpoints(endpoint_id);
auto discovered_endpoints = GetDiscoveredEndpoints(endpoint_id);
std::unique_ptr<EndpointChannel> channel;
ConnectImplResult connect_impl_result;
// TODO(b/156634369): add GetRemoteBluetoothMacAddressEndpoint here for
// valid remote mac address.
auto remote_bluetooth_mac_address =
BluetoothUtils::ToString(options.remote_bluetooth_mac_address);
if (!remote_bluetooth_mac_address.empty()) {
auto additional_endpoint = GetRemoteBluetoothMacAddressEndpoint(
endpoint_id, remote_bluetooth_mac_address, discovered_endpoints);
if (additional_endpoint != nullptr)
discovered_endpoints.push_back(additional_endpoint.get());
}
for (auto connect_endpoint : endpoints) {
for (auto connect_endpoint : discovered_endpoints) {
connect_impl_result = ConnectImpl(client, connect_endpoint);
if (connect_impl_result.status.Ok()) {
channel = std::move(connect_impl_result.endpoint_channel);
@@ -625,10 +634,6 @@ Status BasePcpHandler::RejectConnection(ClientProxy* client,
client->GetClientId(), &response);
}
// proto::connections::Medium BasePcpHandler::GetBandwidthUpgradeMedium() {
// return bandwidth_upgrade_medium_.Get();
//}
void BasePcpHandler::OnIncomingFrame(OfflineFrame& frame,
const std::string& endpoint_id,
ClientProxy* client,
@@ -942,7 +947,7 @@ void BasePcpHandler::ProcessTieBreakLoss(
void BasePcpHandler::InitiateBandwidthUpgrade(
ClientProxy* client, const std::string& endpoint_id,
const std::vector<proto::connections::Medium>& supported_mediums) {
const std::vector<Medium>& supported_mediums) {
// When we successfully connect to a remote endpoint and a bandwidth upgrade
// medium has not yet been decided, we'll pick the highest bandwidth medium
// supported by both us and the remote endpoint. Once we pick a medium, all
@@ -952,16 +957,14 @@ void BasePcpHandler::InitiateBandwidthUpgrade(
// 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.
if (bandwidth_upgrade_medium_.Get() ==
proto::connections::Medium::UNKNOWN_MEDIUM) {
bandwidth_upgrade_medium_.Set(ChooseBestUpgradeMedium(supported_mediums));
Medium bwu_medium = bwu_medium_.Get();
if (bwu_medium == Medium::UNKNOWN_MEDIUM) {
bwu_medium = ChooseBestUpgradeMedium(supported_mediums);
bwu_medium_.Set(bwu_medium);
}
if (AutoUpgradeBandwidth() && (bandwidth_upgrade_medium_.Get() !=
proto::connections::Medium::UNKNOWN_MEDIUM)) {
// TODO(apolyudov): Bring bandwidth upgrade back, when it is ready.
// bandwidth_upgrade_->InitiateBandwidthUpgradeForEndpoint(
// client, endpoint_id, bandwidth_upgrade_medium_.Get());
if (AutoUpgradeBandwidth() && bwu_medium != Medium::UNKNOWN_MEDIUM) {
bwu_manager_->InitiateBwuForEndpoint(client, endpoint_id, bwu_medium);
}
}
@@ -989,6 +992,55 @@ proto::connections::Medium BasePcpHandler::ChooseBestUpgradeMedium(
return proto::connections::Medium::UNKNOWN_MEDIUM;
}
std::unique_ptr<BasePcpHandler::DiscoveredEndpoint>
BasePcpHandler::GetRemoteBluetoothMacAddressEndpoint(
std::string endpoint_id, std::string remote_bluetooth_mac_address,
std::vector<DiscoveredEndpoint*> endpoints) {
if (!discovery_options_.allowed.bluetooth) {
return nullptr;
}
if (endpoints.empty()) {
NEARBY_LOGS(INFO)
<< "Cannot append remote Bluetooth MAC Address, because endpointId "
<< endpoint_id << " has not been discovered";
return nullptr;
}
for (auto endpoint : endpoints) {
if (endpoint->medium == proto::connections::Medium::BLUETOOTH) {
NEARBY_LOGS(INFO)
<< "Cannot append remote Bluetooth MAC Address, because the "
"endpoint has already been found over Bluetooth.";
return nullptr;
}
}
auto remote_bluetooth_device =
mediums_->GetBluetoothClassic().GetRemoteDevice(
remote_bluetooth_mac_address);
if (!remote_bluetooth_device.IsValid()) {
NEARBY_LOGS(INFO)
<< "Cannot append remote Bluetooth MAC Address, because a valid "
"Bluetooth device could not be derived.";
return nullptr;
}
auto bluetooth_endpoint =
std::make_unique<BluetoothEndpoint>(BluetoothEndpoint{
{
endpoint_id,
endpoints[0]->endpoint_info,
endpoints[0]->service_id,
proto::connections::Medium::BLUETOOTH,
},
remote_bluetooth_device,
});
NEARBY_LOGS(INFO) << "Appended remote Bluetooth device "
<< remote_bluetooth_mac_address;
return bluetooth_endpoint;
}
void BasePcpHandler::EvaluateConnectionResult(ClientProxy* client,
const std::string& endpoint_id,
bool can_close_immediately) {
cpp/core_v2/internal/base_pcp_handler.h
+24 -23
View File
@@ -20,6 +20,10 @@
#include <string>
#include <vector>
#include "absl/container/btree_map.h"
#include "absl/container/flat_hash_map.h"
#include "absl/time/time.h"
#include "core_v2/internal/bwu_manager.h"
#include "core_v2/internal/client_proxy.h"
#include "core_v2/internal/encryption_runner.h"
#include "core_v2/internal/endpoint_channel_manager.h"
@@ -31,7 +35,6 @@
#include "core_v2/listeners.h"
#include "core_v2/options.h"
#include "core_v2/status.h"
#include "proto/connections/offline_wire_formats.pb.h"
#include "platform_v2/base/byte_array.h"
#include "platform_v2/base/prng.h"
#include "platform_v2/public/atomic_boolean.h"
@@ -42,12 +45,10 @@
#include "platform_v2/public/scheduled_executor.h"
#include "platform_v2/public/single_thread_executor.h"
#include "platform_v2/public/system_clock.h"
#include "proto/connections/offline_wire_formats.pb.h"
#include "proto/connections_enums.pb.h"
#include "securegcm/d2d_connection_context_v1.h"
#include "securegcm/ukey2_handshake.h"
#include "absl/container/btree_map.h"
#include "absl/container/flat_hash_map.h"
#include "absl/time/time.h"
namespace location {
namespace nearby {
@@ -95,17 +96,16 @@ class BasePcpHandler : public PcpHandler,
// TODO(apolyudov): Add SecureRandom.
BasePcpHandler(Mediums* mediums, EndpointManager* endpoint_manager,
EndpointChannelManager* channel_manager, Pcp pcp);
EndpointChannelManager* channel_manager,
BwuManager* bwu_manager, Pcp pcp);
~BasePcpHandler() override;
BasePcpHandler(BasePcpHandler&&) = delete;
BasePcpHandler& operator=(BasePcpHandler&&) = delete;
// Starts advertising. Once successfully started, changes ClientProxy's state.
// Notifies ConnectionListener (info.listener) in case of any event.
// See
// cpp/core_v2/listeners.h;l=78
Status StartAdvertising(ClientProxy* client,
const std::string& service_id,
// See cpp/core_v2/listeners.h;l=78
Status StartAdvertising(ClientProxy* client, const std::string& service_id,
const ConnectionOptions& options,
const ConnectionRequestInfo& info) override;
@@ -116,8 +116,7 @@ class BasePcpHandler : public PcpHandler,
// Starts discovery of endpoints that may be advertising.
// Updates ClientProxy state once discovery started.
// DiscoveryListener will get called in case of any event.
Status StartDiscovery(ClientProxy* client,
const std::string& service_id,
Status StartDiscovery(ClientProxy* client, const std::string& service_id,
const ConnectionOptions& options,
const DiscoveryListener& listener) override;
@@ -127,16 +126,14 @@ class BasePcpHandler : public PcpHandler,
// Requests a newly discovered remote endpoint it to form a connection.
// Updates state on ClientProxy.
Status RequestConnection(ClientProxy* client,
const std::string& endpoint_id,
Status RequestConnection(ClientProxy* client, const std::string& endpoint_id,
const ConnectionRequestInfo& info,
const ConnectionOptions& options) override;
// Called by either party to accept connection on their part.
// Until both parties call it, connection will not reach a data phase.
// Updates state in ClientProxy.
Status AcceptConnection(ClientProxy* client,
const std::string& endpoint_id,
Status AcceptConnection(ClientProxy* client, const std::string& endpoint_id,
const PayloadListener& payload_listener) override;
// Called by either party to reject connection on their part.
@@ -153,12 +150,12 @@ class BasePcpHandler : public PcpHandler,
// Called when an endpoint disconnects while we're waiting for both sides to
// approve/reject the connection.
// @EndpointManagerThread
void OnEndpointDisconnect(ClientProxy* client,
const std::string& endpoint_id,
void OnEndpointDisconnect(ClientProxy* client, const std::string& endpoint_id,
CountDownLatch* barrier) override;
Pcp GetPcp() const override { return pcp_; }
Strategy GetStrategy() const override { return strategy_; }
Medium GetBwuMedium() const { return bwu_medium_.Get(); }
void DisconnectFromEndpointManager();
protected:
@@ -241,8 +238,7 @@ class BasePcpHandler : public PcpHandler,
std::shared_ptr<DiscoveredEndpoint> endpoint);
// @PcpHandlerThread
void OnEndpointLost(ClientProxy* client,
const DiscoveredEndpoint& endpoint);
void OnEndpointLost(ClientProxy* client, const DiscoveredEndpoint& endpoint);
Exception OnIncomingConnection(
ClientProxy* client, const ByteArray& remote_endpoint_info,
@@ -284,8 +280,8 @@ class BasePcpHandler : public PcpHandler,
// Returns a vector of discovered endpoints, sorted in order of decreasing
// preference.
std::vector<BasePcpHandler::DiscoveredEndpoint*>
GetDiscoveredEndpoints(const std::string& endpoint_id);
std::vector<BasePcpHandler::DiscoveredEndpoint*> GetDiscoveredEndpoints(
const std::string& endpoint_id);
mediums::PeerId CreatePeerIdFromAdvertisement(const string& service_id,
const string& endpoint_id,
@@ -416,6 +412,11 @@ class BasePcpHandler : public PcpHandler,
proto::connections::Medium ChooseBestUpgradeMedium(
const std::vector<proto::connections::Medium>& supported_mediums);
std::unique_ptr<BasePcpHandler::DiscoveredEndpoint>
GetRemoteBluetoothMacAddressEndpoint(
std::string endpoint_id, std::string remote_bluetooth_mac_address,
std::vector<DiscoveredEndpoint*> endpoints);
void ProcessPreConnectionInitiationFailure(const std::string& endpoint_id,
EndpointChannel* channel,
Status status,
@@ -443,8 +444,7 @@ class BasePcpHandler : public PcpHandler,
Status WaitForResult(const std::string& method_name, std::int64_t client_id,
Future<Status>* future);
AtomicReference<proto::connections::Medium> bandwidth_upgrade_medium_{
proto::connections::Medium::UNKNOWN_MEDIUM};
AtomicReference<Medium> bwu_medium_{Medium::UNKNOWN_MEDIUM};
ScheduledExecutor alarm_executor_;
SingleThreadExecutor serial_executor_;
@@ -486,6 +486,7 @@ class BasePcpHandler : public PcpHandler,
Strategy strategy_{PcpToStrategy(pcp_)};
Prng prng_;
EncryptionRunner encryption_runner_;
BwuManager* bwu_manager_;
EndpointManager::FrameProcessor::Handle handle_ = nullptr;
};
cpp/core_v2/internal/base_pcp_handler_test.cc
+26 -13
View File
@@ -18,6 +18,7 @@
#include <memory>
#include "core_v2/internal/base_endpoint_channel.h"
#include "core_v2/internal/bwu_manager.h"
#include "core_v2/internal/client_proxy.h"
#include "core_v2/internal/encryption_runner.h"
#include "core_v2/internal/offline_frames.h"
@@ -90,8 +91,9 @@ class MockPcpHandler : public BasePcpHandler {
public:
using DiscoveredEndpoint = BasePcpHandler::DiscoveredEndpoint;
MockPcpHandler(Mediums* m, EndpointManager* em, EndpointChannelManager* ecm)
: BasePcpHandler(m, em, ecm, Pcp::kP2pCluster) {}
MockPcpHandler(Mediums* m, EndpointManager* em, EndpointChannelManager* ecm,
BwuManager* bwu)
: BasePcpHandler(m, em, ecm, bwu, Pcp::kP2pCluster) {}
// Expose protected inner types of a base type for mocking.
using BasePcpHandler::ConnectImplResult;
@@ -381,7 +383,8 @@ TEST_P(BasePcpHandlerTest, ConstructorDestructorWorks) {
Mediums m;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&m, &em, &ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
SUCCEED();
}
@@ -390,7 +393,8 @@ TEST_P(BasePcpHandlerTest, StartAdvertisingChangesState) {
Mediums m;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&m, &em, &ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartAdvertising(&client, &pcp_handler);
}
@@ -399,7 +403,8 @@ TEST_P(BasePcpHandlerTest, StopAdvertisingChangesState) {
Mediums m;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&m, &em, &ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartAdvertising(&client, &pcp_handler);
EXPECT_CALL(pcp_handler, StopAdvertisingImpl(&client)).Times(1);
EXPECT_TRUE(client.IsAdvertising());
@@ -412,7 +417,8 @@ TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) {
Mediums m;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&m, &em, &ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
}
@@ -421,7 +427,8 @@ TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) {
Mediums m;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&m, &em, &ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
EXPECT_CALL(pcp_handler, StopDiscoveryImpl(&client)).Times(1);
EXPECT_TRUE(client.IsDiscovering());
@@ -435,7 +442,8 @@ TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) {
Mediums m;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&m, &em, &ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
auto mediums = pcp_handler.GetDiscoveryMediums();
auto connect_medium = mediums[mediums.size() - 1];
@@ -458,7 +466,8 @@ TEST_P(BasePcpHandlerTest, AcceptConnectionChangesState) {
Mediums m;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&m, &em, &ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
auto mediums = pcp_handler.GetDiscoveryMediums();
auto connect_medium = mediums[mediums.size() - 1];
@@ -485,7 +494,8 @@ TEST_P(BasePcpHandlerTest, RejectConnectionChangesState) {
Mediums m;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&m, &em, &ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
auto mediums = pcp_handler.GetDiscoveryMediums();
auto connect_medium = mediums[mediums.size() - 1];
@@ -508,7 +518,8 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) {
Mediums m;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&m, &em, &ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
auto mediums = pcp_handler.GetDiscoveryMediums();
auto connect_medium = mediums[mediums.size() - 1];
@@ -544,7 +555,8 @@ TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) {
Mediums m;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&m, &em, &ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
auto mediums = pcp_handler.GetDiscoveryMediums();
auto connect_medium = mediums[mediums.size() - 1];
@@ -583,7 +595,8 @@ TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) {
Mediums m;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&m, &em, &ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
auto mediums = pcp_handler.GetDiscoveryMediums();
auto connect_medium = mediums[mediums.size() - 1];
cpp/core_v2/internal/base_pcp_handler_test.cc.orig
-538
View File
@@ -1,538 +0,0 @@
#include "core_v2/internal/base_pcp_handler.h"
#include <atomic>
#include <memory>
#include "core_v2/internal/base_endpoint_channel.h"
#include "core_v2/internal/client_proxy.h"
#include "core_v2/internal/encryption_runner.h"
#include "core_v2/internal/offline_frames.h"
#include "core_v2/listeners.h"
#include "core_v2/options.h"
#include "core_v2/params.h"
#include "proto/connections/offline_wire_formats.pb.h"
#include "platform_v2/base/byte_array.h"
#include "platform_v2/public/count_down_latch.h"
#include "platform_v2/public/pipe.h"
#include "gmock/gmock.h"
#include "gtest/gtest.h"
#include "absl/time/time.h"
namespace location {
namespace nearby {
namespace connections {
namespace {
using ::location::nearby::proto::connections::Medium;
using ::testing::_;
using ::testing::AtLeast;
using ::testing::Invoke;
using ::testing::MockFunction;
using ::testing::Return;
using ::testing::StrictMock;
constexpr BooleanMediumSelector kTestCases[] = {
BooleanMediumSelector{},
BooleanMediumSelector{
.bluetooth = true,
},
BooleanMediumSelector{
.wifi_lan = true,
},
BooleanMediumSelector{
.bluetooth = true,
.wifi_lan = true,
},
};
class MockEndpointChannel : public BaseEndpointChannel {
public:
explicit MockEndpointChannel(Pipe* reader, Pipe* writer)
: BaseEndpointChannel("channel", &reader->GetInputStream(),
&writer->GetOutputStream()) {}
ExceptionOr<ByteArray> DoRead() { return BaseEndpointChannel::Read(); }
Exception DoWrite(const ByteArray& data) {
return BaseEndpointChannel::Write(data);
}
absl::Time DoGetLastReadTimestamp() {
return BaseEndpointChannel::GetLastReadTimestamp();
}
MOCK_METHOD(ExceptionOr<ByteArray>, Read, (), (override));
MOCK_METHOD(Exception, Write, (const ByteArray& data), (override));
MOCK_METHOD(void, CloseImpl, (), (override));
MOCK_METHOD(proto::connections::Medium, GetMedium, (), (const override));
MOCK_METHOD(std::string, GetType, (), (const override));
MOCK_METHOD(std::string, GetName, (), (const override));
MOCK_METHOD(bool, IsPaused, (), (const override));
MOCK_METHOD(void, Pause, (), (override));
MOCK_METHOD(void, Resume, (), (override));
MOCK_METHOD(absl::Time, GetLastReadTimestamp, (), (const override));
};
class MockPcpHandler : public BasePcpHandler {
public:
using DiscoveredEndpoint = BasePcpHandler::DiscoveredEndpoint;
MockPcpHandler(EndpointManager* em, EndpointChannelManager* ecm)
: BasePcpHandler(em, ecm, Pcp::kP2pCluster) {}
// Expose protected inner types of a base type for mocking.
using BasePcpHandler::ConnectImplResult;
using BasePcpHandler::DiscoveredEndpoint;
using BasePcpHandler::StartOperationResult;
MOCK_METHOD(Strategy, GetStrategy, (), (const override));
MOCK_METHOD(Pcp, GetPcp, (), (const override));
MOCK_METHOD(bool, HasOutgoingConnections, (ClientProxy * client),
(const, override));
MOCK_METHOD(bool, HasIncomingConnections, (ClientProxy * client),
(const, override));
MOCK_METHOD(bool, CanSendOutgoingConnection, (ClientProxy * client),
(const, override));
MOCK_METHOD(bool, CanReceiveIncomingConnection, (ClientProxy * client),
(const, override));
MOCK_METHOD(StartOperationResult, StartAdvertisingImpl,
(ClientProxy * client, const string& service_id,
const string& local_endpoint_id,
const string& local_endpoint_name,
const ConnectionOptions& options),
(override));
MOCK_METHOD(Status, StopAdvertisingImpl, (ClientProxy * client), (override));
MOCK_METHOD(StartOperationResult, StartDiscoveryImpl,
(ClientProxy * client, const string& service_id,
const ConnectionOptions& options),
(override));
MOCK_METHOD(Status, StopDiscoveryImpl, (ClientProxy * client), (override));
MOCK_METHOD(ConnectImplResult, ConnectImpl,
(ClientProxy * client, DiscoveredEndpoint* endpoint), (override));
MOCK_METHOD(proto::connections::Medium, GetDefaultUpgradeMedium, (),
(override));
std::vector<proto::connections::Medium> GetConnectionMediumsByPriority()
override {
return GetDiscoveryMediums();
}
// Mock adapters for protected non-virtual methods of a base class.
void OnEndpointFound(ClientProxy* client,
std::shared_ptr<DiscoveredEndpoint> endpoint) {
BasePcpHandler::OnEndpointFound(client, std::move(endpoint));
}
void OnEndpointLost(ClientProxy* client, const DiscoveredEndpoint& endpoint) {
BasePcpHandler::OnEndpointLost(client, endpoint);
}
std::vector<proto::connections::Medium> GetDiscoveryMediums() {
std::vector<proto::connections::Medium> mediums;
auto allowed =
BasePcpHandler::GetDiscoveryOptions().CompatibleOptions().allowed;
// Mediums are sorted in order of decreasing preference.
if (allowed.wifi_lan)
mediums.push_back(proto::connections::Medium::WIFI_LAN);
if (allowed.web_rtc) mediums.push_back(proto::connections::Medium::WEB_RTC);
if (allowed.bluetooth)
mediums.push_back(proto::connections::Medium::BLUETOOTH);
return mediums;
}
std::vector<BasePcpHandler::DiscoveredEndpoint*> GetDiscoveredEndpoints(
const std::string& endpoint_id) {
return BasePcpHandler::GetDiscoveredEndpoints(endpoint_id);
}
};
class MockContext {
public:
explicit MockContext(std::atomic_int* destroyed = nullptr) {
destroyed_ = destroyed;
}
MockContext(MockContext&&) = default;
MockContext& operator=(MockContext&&) = default;
~MockContext() {
if (destroyed_) (*destroyed_)++;
}
private:
Swapper<std::atomic_int> destroyed_{nullptr};
};
struct MockDiscoveredEndpoint : public MockPcpHandler::DiscoveredEndpoint {
MockDiscoveredEndpoint(DiscoveredEndpoint endpoint, MockContext context)
: DiscoveredEndpoint(std::move(endpoint)), context(std::move(context)) {}
MockContext context;
};
class BasePcpHandlerTest
: public ::testing::TestWithParam<BooleanMediumSelector> {
protected:
struct MockConnectionListener {
StrictMock<MockFunction<void(const std::string& endpoint_id,
const ConnectionResponseInfo& info)>>
initiated_cb;
StrictMock<MockFunction<void(const std::string& endpoint_id)>> accepted_cb;
StrictMock<MockFunction<void(const std::string& endpoint_id,
const Status& status)>>
rejected_cb;
StrictMock<MockFunction<void(const std::string& endpoint_id)>>
disconnected_cb;
StrictMock<MockFunction<void(const std::string& endpoint_id,
std::int32_t quality)>>
bandwidth_changed_cb;
};
struct MockDiscoveryListener {
StrictMock<MockFunction<void(const std::string& endpoint_id,
const std::string& endpoint_name,
const std::string& service_id)>>
endpoint_found_cb;
StrictMock<MockFunction<void(const std::string& endpoint_id)>>
endpoint_lost_cb;
StrictMock<
MockFunction<void(const std::string& endpoint_id, DistanceInfo info)>>
endpoint_distance_changed_cb;
};
void StartAdvertising(ClientProxy* client, MockPcpHandler* pcp_handler,
BooleanMediumSelector allowed = GetParam()) {
std::string service_id{"service"};
ConnectionOptions options{
.strategy = Strategy::kP2pCluster,
.allowed = allowed,
.auto_upgrade_bandwidth = true,
.enforce_topology_constraints = true,
};
ConnectionRequestInfo info{
.name = "remote_endpoint_name",
.listener = connection_listener_,
};
EXPECT_CALL(*pcp_handler,
StartAdvertisingImpl(client, service_id, _, info.name, _))
.WillOnce(Return(MockPcpHandler::StartOperationResult{
.status = {Status::kSuccess},
.mediums = {Medium::BLE},
}));
EXPECT_EQ(pcp_handler->StartAdvertising(client, service_id, options, info),
Status{Status::kSuccess});
EXPECT_TRUE(client->IsAdvertising());
}
void StartDiscovery(ClientProxy* client, MockPcpHandler* pcp_handler,
BooleanMediumSelector allowed = GetParam()) {
std::string service_id{"service"};
ConnectionOptions options{
.strategy = Strategy::kP2pCluster,
.allowed = allowed,
.auto_upgrade_bandwidth = true,
.enforce_topology_constraints = true,
};
EXPECT_CALL(*pcp_handler, StartDiscoveryImpl(client, service_id, _))
.WillOnce(Return(MockPcpHandler::StartOperationResult{
.status = {Status::kSuccess},
.mediums = {Medium::BLE},
}));
EXPECT_EQ(pcp_handler->StartDiscovery(client, service_id, options,
discovery_listener_),
Status{Status::kSuccess});
EXPECT_TRUE(client->IsDiscovering());
}
std::pair<std::unique_ptr<MockEndpointChannel>,
std::unique_ptr<MockEndpointChannel>>
SetupConnection(Pipe& pipe_a, Pipe& pipe_b) { // NOLINT
auto channel_a = std::make_unique<MockEndpointChannel>(&pipe_b, &pipe_a);
auto channel_b = std::make_unique<MockEndpointChannel>(&pipe_a, &pipe_b);
// On initiator (A) side, we drop the first write, since this is a
// connection establishment packet, and we don't have the peer entity, just
// the peer channel. The rest of the exchange must happen for the benefit of
// DH key exchange.
EXPECT_CALL(*channel_a, Read())
.WillRepeatedly(Invoke(
[channel = channel_a.get()]() { return channel->DoRead(); }));
EXPECT_CALL(*channel_a, Write(_))
.WillOnce(Return(Exception{Exception::kSuccess}))
.WillRepeatedly(
Invoke([channel = channel_a.get()](const ByteArray& data) {
return channel->DoWrite(data);
}));
EXPECT_CALL(*channel_a, GetMedium).WillRepeatedly(Return(Medium::BLE));
EXPECT_CALL(*channel_a, GetLastReadTimestamp)
.WillRepeatedly(Return(absl::Now()));
EXPECT_CALL(*channel_a, IsPaused).WillRepeatedly(Return(false));
EXPECT_CALL(*channel_b, Read())
.WillRepeatedly(Invoke(
[channel = channel_b.get()]() { return channel->DoRead(); }));
EXPECT_CALL(*channel_b, Write(_))
.WillRepeatedly(
Invoke([channel = channel_b.get()](const ByteArray& data) {
return channel->DoWrite(data);
}));
EXPECT_CALL(*channel_b, GetMedium).WillRepeatedly(Return(Medium::BLE));
EXPECT_CALL(*channel_b, GetLastReadTimestamp)
.WillRepeatedly(Return(absl::Now()));
EXPECT_CALL(*channel_b, IsPaused).WillRepeatedly(Return(false));
return std::make_pair(std::move(channel_a), std::move(channel_b));
}
void RequestConnection(const std::string& endpoint_id,
std::unique_ptr<MockEndpointChannel> channel_a,
MockEndpointChannel* channel_b, ClientProxy* client,
MockPcpHandler* pcp_handler,
std::atomic_int* flag = nullptr) {
ConnectionRequestInfo info{
.name = "ABCD",
.listener = connection_listener_,
};
EXPECT_CALL(mock_discovery_listener_.endpoint_found_cb, Call);
EXPECT_CALL(*pcp_handler, CanSendOutgoingConnection)
.WillRepeatedly(Return(true));
EXPECT_CALL(*pcp_handler, GetStrategy)
.WillRepeatedly(Return(Strategy::kP2pCluster));
EXPECT_CALL(mock_connection_listener_.initiated_cb, Call).Times(1);
// Simulate successful discovery.
auto encryption_runner = std::make_unique<EncryptionRunner>();
auto allowed_mediums = pcp_handler->GetDiscoveryMediums();
EXPECT_CALL(*pcp_handler, ConnectImpl)
.WillOnce(Invoke([&channel_a, medium = allowed_mediums[0]](
ClientProxy* client,
MockPcpHandler::DiscoveredEndpoint* endpoint) {
return MockPcpHandler::ConnectImplResult{
.medium = medium,
.status = {Status::kSuccess},
.endpoint_channel = std::move(channel_a),
};
}));
for (const auto& medium : allowed_mediums) {
pcp_handler->OnEndpointFound(
client,
std::make_shared<MockDiscoveredEndpoint>(MockDiscoveredEndpoint{
{
endpoint_id,
info.name,
"service",
medium,
},
MockContext{flag},
}));
}
auto other_client = std::make_unique<ClientProxy>();
// Run peer crypto in advance, if channel_b is provided.
// Otherwise stay in not-encrypted state.
if (channel_b != nullptr) {
encryption_runner->StartServer(other_client.get(), endpoint_id, channel_b,
{});
}
EXPECT_EQ(pcp_handler->RequestConnection(client, endpoint_id, info),
Status{Status::kSuccess});
NEARBY_LOG(INFO, "Stopping Encryption Runner");
}
Pipe pipe_a_;
Pipe pipe_b_;
MockConnectionListener mock_connection_listener_;
MockDiscoveryListener mock_discovery_listener_;
ConnectionListener connection_listener_{
.initiated_cb = mock_connection_listener_.initiated_cb.AsStdFunction(),
.accepted_cb = mock_connection_listener_.accepted_cb.AsStdFunction(),
.rejected_cb = mock_connection_listener_.rejected_cb.AsStdFunction(),
.disconnected_cb =
mock_connection_listener_.disconnected_cb.AsStdFunction(),
.bandwidth_changed_cb =
mock_connection_listener_.bandwidth_changed_cb.AsStdFunction(),
};
DiscoveryListener discovery_listener_{
.endpoint_found_cb =
mock_discovery_listener_.endpoint_found_cb.AsStdFunction(),
.endpoint_lost_cb =
mock_discovery_listener_.endpoint_lost_cb.AsStdFunction(),
.endpoint_distance_changed_cb =
mock_discovery_listener_.endpoint_distance_changed_cb.AsStdFunction(),
};
};
TEST_P(BasePcpHandlerTest, ConstructorDestructorWorks) {
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&em, &ecm);
SUCCEED();
}
TEST_P(BasePcpHandlerTest, StartAdvertisingChangesState) {
ClientProxy client;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&em, &ecm);
StartAdvertising(&client, &pcp_handler);
}
TEST_P(BasePcpHandlerTest, StopAdvertisingChangesState) {
ClientProxy client;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&em, &ecm);
StartAdvertising(&client, &pcp_handler);
EXPECT_CALL(pcp_handler, StopAdvertisingImpl(&client)).Times(1);
EXPECT_TRUE(client.IsAdvertising());
pcp_handler.StopAdvertising(&client);
EXPECT_FALSE(client.IsAdvertising());
}
TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) {
ClientProxy client;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&em, &ecm);
StartDiscovery(&client, &pcp_handler);
}
TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) {
ClientProxy client;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&em, &ecm);
StartDiscovery(&client, &pcp_handler);
EXPECT_CALL(pcp_handler, StopDiscoveryImpl(&client)).Times(1);
EXPECT_TRUE(client.IsDiscovering());
pcp_handler.StopDiscovery(&client);
EXPECT_FALSE(client.IsDiscovering());
}
TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) {
std::string endpoint_id{"1234"};
ClientProxy client;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&em, &ecm);
StartDiscovery(&client, &pcp_handler);
auto channel_pair = SetupConnection(pipe_a_, pipe_b_);
auto& channel_a = channel_pair.first;
auto& channel_b = channel_pair.second;
EXPECT_CALL(*channel_a, CloseImpl).Times(1);
EXPECT_CALL(*channel_b, CloseImpl).Times(1);
EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(AtLeast(0));
RequestConnection(endpoint_id, std::move(channel_a), channel_b.get(), &client,
&pcp_handler);
NEARBY_LOG(INFO, "RequestConnection complete");
channel_b->Close();
pcp_handler.DisconnectFromEndpointManager();
}
TEST_P(BasePcpHandlerTest, AcceptConnectionChangesState) {
std::string endpoint_id{"1234"};
ClientProxy client;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&em, &ecm);
StartDiscovery(&client, &pcp_handler);
auto channel_pair = SetupConnection(pipe_a_, pipe_b_);
auto& channel_a = channel_pair.first;
auto& channel_b = channel_pair.second;
EXPECT_CALL(*channel_a, CloseImpl).Times(1);
EXPECT_CALL(*channel_b, CloseImpl).Times(1);
RequestConnection(endpoint_id, std::move(channel_a), channel_b.get(), &client,
&pcp_handler);
NEARBY_LOG(INFO, "Attempting to accept connection: id=%s",
endpoint_id.c_str());
EXPECT_EQ(pcp_handler.AcceptConnection(&client, endpoint_id, {}),
Status{Status::kSuccess});
EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(AtLeast(0));
NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id;
channel_b->Close();
pcp_handler.DisconnectFromEndpointManager();
}
TEST_P(BasePcpHandlerTest, RejectConnectionChangesState) {
std::string endpoint_id{"1234"};
ClientProxy client;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&em, &ecm);
StartDiscovery(&client, &pcp_handler);
auto channel_pair = SetupConnection(pipe_a_, pipe_b_);
auto& channel_b = channel_pair.second;
EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(1);
RequestConnection(endpoint_id, std::move(channel_pair.first), channel_b.get(),
&client, &pcp_handler);
NEARBY_LOGS(INFO) << "Attempting to reject connection: id=" << endpoint_id;
EXPECT_EQ(pcp_handler.RejectConnection(&client, endpoint_id),
Status{Status::kSuccess});
NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id;
channel_b->Close();
pcp_handler.DisconnectFromEndpointManager();
}
TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) {
std::string endpoint_id{"1234"};
ClientProxy client;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&em, &ecm);
StartDiscovery(&client, &pcp_handler);
auto channel_pair = SetupConnection(pipe_a_, pipe_b_);
auto& channel_a = channel_pair.first;
auto& channel_b = channel_pair.second;
EXPECT_CALL(*channel_a, CloseImpl).Times(1);
EXPECT_CALL(*channel_b, CloseImpl).Times(1);
RequestConnection(endpoint_id, std::move(channel_a), channel_b.get(), &client,
&pcp_handler);
NEARBY_LOGS(INFO) << "Attempting to accept connection: id=" << endpoint_id;
EXPECT_CALL(mock_connection_listener_.accepted_cb, Call).Times(1);
EXPECT_CALL(mock_connection_listener_.disconnected_cb, Call)
.Times(AtLeast(0));
EXPECT_EQ(pcp_handler.AcceptConnection(&client, endpoint_id, {}),
Status{Status::kSuccess});
NEARBY_LOG(INFO, "Simulating remote accept: id=%s", endpoint_id.c_str());
auto frame =
parser::FromBytes(parser::ForConnectionResponse(Status::kSuccess));
pcp_handler.OnIncomingFrame(frame.result(), endpoint_id, &client,
Medium::BLE);
NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id;
channel_b->Close();
pcp_handler.DisconnectFromEndpointManager();
}
TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) {
std::atomic_int destroyed_flag = 0;
int mediums_count = 0;
{
std::string endpoint_id{"1234"};
ClientProxy client;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&em, &ecm);
StartDiscovery(&client, &pcp_handler);
auto channel_pair = SetupConnection(pipe_a_, pipe_b_);
auto& channel_a = channel_pair.first;
auto& channel_b = channel_pair.second;
EXPECT_CALL(*channel_a, CloseImpl).Times(1);
EXPECT_CALL(*channel_b, CloseImpl).Times(1);
RequestConnection(endpoint_id, std::move(channel_a), channel_b.get(),
&client, &pcp_handler, &destroyed_flag);
mediums_count = pcp_handler.GetDiscoveryMediums().size();
NEARBY_LOG(INFO, "Attempting to accept connection: id=%s",
endpoint_id.c_str());
EXPECT_EQ(pcp_handler.AcceptConnection(&client, endpoint_id, {}),
Status{Status::kSuccess});
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();
}
EXPECT_EQ(destroyed_flag.load(), mediums_count);
}
INSTANTIATE_TEST_SUITE_P(ParameterizedBasePcpHandlerTest, BasePcpHandlerTest,
::testing::ValuesIn(kTestCases));
} // namespace
} // namespace connections
} // namespace nearby
} // namespace location
cpp/core_v2/internal/bwu_manager.cc
+16 -6
View File
@@ -18,6 +18,7 @@
#include "core_v2/internal/bwu_handler.h"
#include "core_v2/internal/offline_frames.h"
#include "core_v2/internal/webrtc_bwu_handler.h"
#include "platform_v2/base/byte_array.h"
#include "platform_v2/public/count_down_latch.h"
#include "proto/connections_enums.pb.h"
@@ -66,7 +67,11 @@ void BwuManager::InitBwuHandlers() {
.incoming_connection_cb =
absl::bind_front(&BwuManager::OnIncomingConnection, this),
};
// TODO(apolyudov): inject instances of supported upgrade medium handlers.
if (config_.allow_upgrade_to.web_rtc) {
handlers_.emplace(Medium::WEB_RTC,
std::make_unique<WebrtcBwuHandler>(
*mediums_, *channel_manager_, notifications));
}
}
void BwuManager::Shutdown() {
@@ -104,12 +109,17 @@ void BwuManager::Shutdown() {
}
// This is the point on the Initiator side where the
// currentBwuMedium is set.
// medium_ is set.
void BwuManager::InitiateBwuForEndpoint(ClientProxy* client,
const std::string& endpoint_id) {
RunOnBwuManagerThread([this, client, endpoint_id]() {
auto* handler = SetCurrentBwuHandler(ChooseBestUpgradeMedium(
client->GetUpgradeMediums(endpoint_id).GetMediums(true)));
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;
cpp/core_v2/internal/bwu_manager.h
+2 -1
View File
@@ -79,7 +79,8 @@ class BwuManager : public EndpointManager::FrameProcessor {
// Function initiates the bandwidth upgrade and sends an
// UPGRADE_PATH_AVAILABLE OfflineFrame.
void InitiateBwuForEndpoint(ClientProxy* client_proxy,
const std::string& endpoint_id);
const std::string& endpoint_id,
Medium new_medium = Medium::UNKNOWN_MEDIUM);
// == EndpointManager::FrameProcessor interface ==.
// This is the point on the inbound BWU protocol where the handler_ is set.
cpp/core_v2/internal/mediums/ble.cc
+3 -1
View File
@@ -120,6 +120,7 @@ bool Ble::IsAdvertisingLocked(const std::string& service_id) {
}
bool Ble::StartScanning(const std::string& service_id,
const std::string& fast_advertisement_service_uuid,
DiscoveredPeripheralCallback callback) {
MutexLock lock(&mutex_);
@@ -147,7 +148,8 @@ bool Ble::StartScanning(const std::string& service_id,
return false;
}
if (!medium_.StartScanning(service_id, callback)) {
if (!medium_.StartScanning(service_id, fast_advertisement_service_uuid,
callback)) {
NEARBY_LOGS(INFO) << "Failed to start scan of BLE services.";
return false;
}
cpp/core_v2/internal/mediums/ble.h
+1
View File
@@ -59,6 +59,7 @@ class Ble {
// range through a callback. Returns true, if scanning mode was enabled,
// false otherwise.
bool StartScanning(const std::string& service_id,
const std::string& fast_advertisement_service_uuid,
DiscoveredPeripheralCallback callback)
ABSL_LOCKS_EXCLUDED(mutex_);
cpp/core_v2/internal/mediums/ble_test.cc
+4 -1
View File
@@ -32,7 +32,7 @@ namespace {
constexpr absl::Duration kWaitDuration = absl::Milliseconds(1000);
constexpr absl::string_view kServiceID{"com.google.location.nearby.apps.test"};
constexpr absl::string_view kAdvertisementString{"\x0a\x0b\x0c\x0d"};
constexpr absl::string_view kFastAdvertisementServiceUuid{"\xff\xfe"};
constexpr absl::string_view kFastAdvertisementServiceUuid{"\xf3\xfe"};
class BleTest : public ::testing::Test {
protected:
@@ -75,6 +75,7 @@ TEST_F(BleTest, CanStartAdvertising) {
ble_b.StartScanning(
service_id,
fast_advertisement_service_uuid,
DiscoveredPeripheralCallback{
.peripheral_discovered_cb =
[&found_latch](
@@ -109,6 +110,7 @@ TEST_F(BleTest, CanStartDiscovery) {
EXPECT_TRUE(ble_a.StartScanning(
service_id,
fast_advertisement_service_uuid,
DiscoveredPeripheralCallback{
.peripheral_discovered_cb =
[&accept_latch](
@@ -153,6 +155,7 @@ TEST_F(BleTest, CanStartAcceptingConnectionsAndConnect) {
BlePeripheral discovered_peripheral;
ble_b.StartScanning(
service_id,
fast_advertisement_service_uuid,
{
.peripheral_discovered_cb =
[&found_latch, &discovered_peripheral](
cpp/core_v2/internal/mediums/bluetooth_classic.cc
+2 -2
View File
@@ -382,10 +382,10 @@ BluetoothSocket BluetoothClassic::Connect(BluetoothDevice& bluetooth_device,
return socket;
}
BluetoothDevice BluetoothClassic::FindRemoteDevice(
BluetoothDevice BluetoothClassic::GetRemoteDevice(
const std::string& mac_address) {
MutexLock lock(&mutex_);
return medium_.FindRemoteDevice(mac_address);
return medium_.GetRemoteDevice(mac_address);
}
std::string BluetoothClassic::GetMacAddress() const {
cpp/core_v2/internal/mediums/bluetooth_classic.h
+1 -1
View File
@@ -116,7 +116,7 @@ class BluetoothClassic {
std::string GetMacAddress() const ABSL_LOCKS_EXCLUDED(mutex_);
BluetoothDevice FindRemoteDevice(const std::string& mac_address)
BluetoothDevice GetRemoteDevice(const std::string& mac_address)
ABSL_LOCKS_EXCLUDED(mutex_);
private:
cpp/core_v2/internal/mediums/webrtc/connection_flow.cc
+3 -4
View File
@@ -247,8 +247,8 @@ bool ConnectionFlow::InitPeerConnection(WebRtcMedium& webrtc_medium) {
Future<bool> success_future;
webrtc_medium.CreatePeerConnection(
&peer_connection_observer_,
[this, &success_future](
rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection) {
[this, success_future](rtc::scoped_refptr<webrtc::PeerConnectionInterface>
peer_connection) mutable {
if (!peer_connection) {
success_future.Set(false);
return;
@@ -343,8 +343,7 @@ bool ConnectionFlow::CloseLocked() {
state_ = State::kEnded;
data_channel_future_.SetException({Exception::kInterrupted});
if (peer_connection_)
peer_connection_->Close();
if (peer_connection_) peer_connection_->Close();
data_channel_observer_.reset();
NEARBY_LOG(INFO, "Closed WebRTC connection.");
cpp/core_v2/internal/offline_service_controller.h
+2 -1
View File
@@ -81,9 +81,10 @@ class OfflineServiceController : public ServiceController {
EndpointChannelManager channel_manager_;
EndpointManager endpoint_manager_{&channel_manager_};
PayloadManager payload_manager_{endpoint_manager_};
PcpManager pcp_manager_{mediums_, channel_manager_, endpoint_manager_};
BwuManager bwu_manager_{
mediums_, endpoint_manager_, channel_manager_, {}, {}};
PcpManager pcp_manager_{mediums_, channel_manager_, endpoint_manager_,
bwu_manager_};
};
} // namespace connections
cpp/core_v2/internal/offline_service_controller.h.orig
-81
View File
@@ -1,81 +0,0 @@
#ifndef CORE_V2_INTERNAL_OFFLINE_SERVICE_CONTROLLER_H_
#define CORE_V2_INTERNAL_OFFLINE_SERVICE_CONTROLLER_H_
#include <cstdint>
#include <string>
#include <vector>
#include "core_v2/internal/client_proxy.h"
#include "core_v2/internal/endpoint_channel_manager.h"
#include "core_v2/internal/endpoint_manager.h"
#include "core_v2/internal/mediums/mediums.h"
#include "core_v2/internal/payload_manager.h"
#include "core_v2/internal/pcp_manager.h"
#include "core_v2/internal/service_controller.h"
#include "core_v2/listeners.h"
#include "core_v2/options.h"
#include "core_v2/payload.h"
#include "core_v2/status.h"
namespace location {
namespace nearby {
namespace connections {
class OfflineServiceController : public ServiceController {
public:
OfflineServiceController() = default;
~OfflineServiceController() override;
Status StartAdvertising(ClientProxy* client,
const std::string& service_id,
const ConnectionOptions& options,
const ConnectionRequestInfo& info) override;
void StopAdvertising(ClientProxy* client) override;
Status StartDiscovery(ClientProxy* client,
const std::string& service_id,
const ConnectionOptions& options,
const DiscoveryListener& listener) override;
void StopDiscovery(ClientProxy* client) override;
Status RequestConnection(ClientProxy* client,
const std::string& endpoint_id,
const ConnectionRequestInfo& info,
const ConnectionOptions& options) override;
Status AcceptConnection(ClientProxy* client,
const std::string& endpoint_id,
const PayloadListener& listener) override;
Status RejectConnection(ClientProxy* client,
const std::string& endpoint_id) override;
void InitiateBandwidthUpgrade(ClientProxy* client,
const std::string& endpoint_id) override;
void SendPayload(ClientProxy* client,
const std::vector<std::string>& endpoint_ids,
Payload payload) override;
Status CancelPayload(ClientProxy* client,
Payload::Id payload_id) override;
void DisconnectFromEndpoint(ClientProxy* client,
const std::string& endpoint_id) override;
void Stop();
private:
// Note that the order of declaration of these is crucial, because we depend
// on the destructors running (strictly) in the reverse order; a deviation
// from that will lead to crashes at runtime.
AtomicBoolean stop_{false};
Mediums mediums_;
EndpointChannelManager channel_manager_;
EndpointManager endpoint_manager_{&channel_manager_};
PayloadManager payload_manager_{endpoint_manager_};
PcpManager pcp_manager_{mediums_, channel_manager_, endpoint_manager_};
};
} // namespace connections
} // namespace nearby
} // namespace location
#endif // CORE_V2_INTERNAL_OFFLINE_SERVICE_CONTROLLER_H_
cpp/core_v2/internal/p2p_cluster_pcp_handler.cc
+132 -48
View File
@@ -18,6 +18,7 @@
#include "core_v2/internal/ble_advertisement.h"
#include "core_v2/internal/ble_endpoint_channel.h"
#include "core_v2/internal/bluetooth_endpoint_channel.h"
#include "core_v2/internal/bwu_manager.h"
#include "core_v2/internal/mediums/utils.h"
#include "core_v2/internal/mediums/webrtc/webrtc_socket_wrapper.h"
#include "core_v2/internal/webrtc_endpoint_channel.h"
@@ -37,10 +38,22 @@ ByteArray P2pClusterPcpHandler::GenerateHash(const std::string& source,
return Utils::Sha256Hash(source, size);
}
bool P2pClusterPcpHandler::ShouldAdvertiseBluetoothMacOverBle(
PowerLevel power_level) {
return power_level == PowerLevel::kHighPower;
}
bool P2pClusterPcpHandler::ShouldAcceptBluetoothConnections(
const ConnectionOptions& options) {
return options.enable_bluetooth_listening;
}
P2pClusterPcpHandler::P2pClusterPcpHandler(
Mediums* mediums, EndpointManager* endpoint_manager,
EndpointChannelManager* endpoint_channel_manager, Pcp pcp)
: BasePcpHandler(mediums, endpoint_manager, endpoint_channel_manager, pcp),
EndpointChannelManager* endpoint_channel_manager, BwuManager* bwu_manager,
Pcp pcp)
: BasePcpHandler(mediums, endpoint_manager, endpoint_channel_manager,
bwu_manager, pcp),
bluetooth_radio_(mediums->GetBluetoothRadio()),
bluetooth_medium_(mediums->GetBluetoothClassic()),
ble_medium_(mediums->GetBle()),
@@ -145,10 +158,12 @@ Status P2pClusterPcpHandler::StopAdvertisingImpl(ClientProxy* client) {
bluetooth_medium_.StopAcceptingConnections(client->GetAdvertisingServiceId());
ble_medium_.StopAdvertising(client->GetAdvertisingServiceId());
ble_medium_.StopAcceptingConnections(client->GetAdvertisingServiceId());
webrtc_medium_.StopAcceptingConnections();
wifi_lan_medium_.StopAdvertising(client->GetAdvertisingServiceId());
wifi_lan_medium_.StopAcceptingConnections(client->GetAdvertisingServiceId());
return {Status::kSuccess};
}
@@ -325,12 +340,11 @@ void P2pClusterPcpHandler::BlePeripheralDiscoveredHandler(
return;
}
// Parse the Ble advertisement bytes.
// Parse the BLE advertisement bytes.
BleAdvertisement advertisement(
fast_advertisement,
peripheral.GetAdvertisementBytes(service_id));
fast_advertisement, peripheral.GetAdvertisementBytes(service_id));
// Make sure the Ble advertisement points to a valid
// Make sure the BLE advertisement points to a valid
// endpoint we're discovering.
if (!IsRecognizedBleEndpoint(service_id, advertisement)) return;
@@ -357,7 +371,34 @@ void P2pClusterPcpHandler::BlePeripheralDiscoveredHandler(
peripheral,
}));
// TODO(b/156632928): Check for Bluetooth device with remote mac address.
// Make sure we can connect to this device via Classic Bluetooth.
std::string remote_bluetooth_mac_address =
advertisement.GetBluetoothMacAddress();
if (remote_bluetooth_mac_address.empty()) {
NEARBY_LOGS(INFO)
<< "No Bluetooth Classic MAC address found in advertisement";
return;
}
BluetoothDevice remote_bluetooth_device =
bluetooth_medium_.GetRemoteDevice(remote_bluetooth_mac_address);
if (!remote_bluetooth_device.IsValid()) {
NEARBY_LOGS(INFO) << "A valid Bluetooth device could not be derived from "
"the MAC address "
<< remote_bluetooth_mac_address;
return;
}
OnEndpointFound(client,
std::make_shared<BluetoothEndpoint>(BluetoothEndpoint{
{
advertisement.GetEndpointId(),
advertisement.GetEndpointInfo(),
service_id,
proto::connections::Medium::BLUETOOTH,
},
remote_bluetooth_device,
}));
});
}
@@ -561,7 +602,7 @@ BasePcpHandler::StartOperationResult P2pClusterPcpHandler::StartDiscoveryImpl(
.peripheral_lost_cb = absl::bind_front(
&P2pClusterPcpHandler::BlePeripheralLostHandler, this, client),
},
client, service_id);
client, service_id, options.fast_advertisement_service_uuid);
if (ble_medium != proto::connections::UNKNOWN_MEDIUM) {
NEARBY_LOG(INFO, "P2pClusterPcpHandler::StartDiscoveryImpl: Ble added");
mediums_started_successfully.push_back(ble_medium);
@@ -767,51 +808,90 @@ proto::connections::Medium P2pClusterPcpHandler::StartBleAdvertising(
const std::string& local_endpoint_id, const ByteArray& local_endpoint_info,
const ConnectionOptions& options) {
bool fast_advertisement = !options.fast_advertisement_service_uuid.empty();
PowerLevel power_level =
options.low_power ? PowerLevel::kLowPower : PowerLevel::kHighPower;
// Start listening for connections before advertising in case a connection
// request comes in very quickly.
// request comes in very quickly. BLE allows connecting over BLE itself, as
// well as advertising the Bluetooth MAC address to allow connecting over
// Bluetooth Classic.
NEARBY_LOGS(INFO) << "P2pClusterPcpHandler::StartBleAdvertising: service_id="
<< service_id << ": start";
if (ble_medium_.IsAcceptingConnections(service_id)) {
NEARBY_LOGS(ERROR) << "Ble is already accepting connections for service_id="
<< service_id;
return proto::connections::UNKNOWN_MEDIUM;
}
if (!ble_medium_.IsAcceptingConnections(service_id)) {
if (!bluetooth_radio_.Enable() ||
!ble_medium_.StartAcceptingConnections(
service_id, {.accepted_cb = [this, client, local_endpoint_info](
BleSocket socket,
const std::string& service_id) {
if (!socket.IsValid()) {
NEARBY_LOG(ERROR, "Invalid socket in accept callback: name=%s",
std::string(local_endpoint_info).c_str());
return;
}
RunOnPcpHandlerThread([this, client, local_endpoint_info,
service_id,
socket = std::move(socket)]() mutable {
std::string remote_peripheral_name =
socket.GetRemotePeripheral().GetName();
auto channel = absl::make_unique<BleEndpointChannel>(
remote_peripheral_name, socket);
ByteArray remote_peripheral_info =
socket.GetRemotePeripheral().GetAdvertisementBytes(
service_id);
NEARBY_LOGS(INFO) << "P2pClusterPcpHandler::StartBleAdvertising: service_id="
<< service_id << ": invoking";
if (!bluetooth_radio_.Enable() ||
!ble_medium_.StartAcceptingConnections(
service_id,
{.accepted_cb = [this, client, local_endpoint_info](
BleSocket socket, const std::string& service_id) {
if (!socket.IsValid()) {
NEARBY_LOG(ERROR, "Invalid socket in accept callback: name=%s",
std::string(local_endpoint_info).c_str());
return;
}
RunOnPcpHandlerThread([this, client, local_endpoint_info,
service_id,
socket = std::move(socket)]() mutable {
std::string remote_peripheral_name =
socket.GetRemotePeripheral().GetName();
auto channel = absl::make_unique<BleEndpointChannel>(
remote_peripheral_name, socket);
ByteArray remote_peripheral_info =
socket.GetRemotePeripheral().GetAdvertisementBytes(
service_id);
OnIncomingConnection(client, remote_peripheral_info,
std::move(channel),
proto::connections::Medium::BLE);
});
}})) {
OnIncomingConnection(client, remote_peripheral_info,
std::move(channel),
proto::connections::Medium::BLE);
});
}})) {
NEARBY_LOGS(ERROR)
<< "Ble failed to start accepting connections for service_id="
<< service_id;
return proto::connections::UNKNOWN_MEDIUM;
}
NEARBY_LOGS(ERROR)
<< "Ble failed to start accepting connections for service_id="
<< "Ble succeed to start accepting connections for service_id="
<< service_id;
return proto::connections::UNKNOWN_MEDIUM;
}
// TODO(b/156632928): Should check for Bluetooth connection here
if (ShouldAdvertiseBluetoothMacOverBle(power_level) ||
ShouldAcceptBluetoothConnections(options)) {
if (bluetooth_medium_.IsAvailable() &&
!bluetooth_medium_.IsAcceptingConnections(service_id)) {
if (!bluetooth_radio_.Enable() ||
!bluetooth_medium_.StartAcceptingConnections(
service_id, {.accepted_cb = [this, client, local_endpoint_info](
BluetoothSocket socket) {
if (!socket.IsValid()) {
NEARBY_LOG(ERROR,
"Invalid socket in accept callback: name=%s",
std::string(local_endpoint_info).c_str());
return;
}
RunOnPcpHandlerThread([this, client, local_endpoint_info,
socket = std::move(socket)]() mutable {
std::string remote_device_name =
socket.GetRemoteDevice().GetName();
auto channel = absl::make_unique<BluetoothEndpointChannel>(
remote_device_name, socket);
ByteArray remote_device_info{remote_device_name};
OnIncomingConnection(client, remote_device_info,
std::move(channel),
proto::connections::Medium::BLUETOOTH);
});
}})) {
NEARBY_LOGS(ERROR)
<< "BT failed to start accepting connections for service_id="
<< service_id;
ble_medium_.StopAcceptingConnections(service_id);
return proto::connections::UNKNOWN_MEDIUM;
}
NEARBY_LOGS(ERROR)
<< "BT succeed to start accepting connections for service_id="
<< service_id;
}
}
NEARBY_LOG(INFO,
"P2pClusterPcpHandler::StartBleAdvertising: service=%s: "
@@ -828,8 +908,10 @@ proto::connections::Medium P2pClusterPcpHandler::StartBleAdvertising(
} else {
const ByteArray service_id_hash =
GenerateHash(service_id, BleAdvertisement::kServiceIdHashLength);
// TODO(b/156632928): Should advertise Bluetooth MacAddress Over Ble
std::string bluetooth_mac_address;
if (bluetooth_medium_.IsAvailable() &&
ShouldAdvertiseBluetoothMacOverBle(power_level))
bluetooth_mac_address = bluetooth_medium_.GetMacAddress();
advertisement_bytes = ByteArray(BleAdvertisement(
kBleAdvertisementVersion, GetPcp(), service_id_hash, local_endpoint_id,
@@ -866,9 +948,11 @@ proto::connections::Medium P2pClusterPcpHandler::StartBleAdvertising(
proto::connections::Medium P2pClusterPcpHandler::StartBleScanning(
BleDiscoveredPeripheralCallback callback, ClientProxy* client,
const std::string& service_id) {
const std::string& service_id,
const std::string& fast_advertisement_service_uuid) {
if (bluetooth_radio_.Enable() &&
ble_medium_.StartScanning(service_id, std::move(callback))) {
ble_medium_.StartScanning(service_id, fast_advertisement_service_uuid,
std::move(callback))) {
NEARBY_LOGS(INFO) << "P2pClusterPcpHandler::StartBleScanning: ok";
return proto::connections::BLE;
} else {
cpp/core_v2/internal/p2p_cluster_pcp_handler.h
+7 -1
View File
@@ -21,6 +21,7 @@
#include "core_v2/internal/base_pcp_handler.h"
#include "core_v2/internal/ble_advertisement.h"
#include "core_v2/internal/bluetooth_device_name.h"
#include "core_v2/internal/bwu_manager.h"
#include "core_v2/internal/client_proxy.h"
#include "core_v2/internal/endpoint_channel_manager.h"
#include "core_v2/internal/endpoint_manager.h"
@@ -52,6 +53,7 @@ class P2pClusterPcpHandler : public BasePcpHandler {
public:
P2pClusterPcpHandler(Mediums* mediums, EndpointManager* endpoint_manager,
EndpointChannelManager* channel_manager,
BwuManager* bwu_manager,
Pcp pcp = Pcp::kP2pCluster);
~P2pClusterPcpHandler() override = default;
@@ -131,6 +133,9 @@ class P2pClusterPcpHandler : public BasePcpHandler {
WifiLanServiceInfo::Version::kV1;
static ByteArray GenerateHash(const std::string& source, size_t size);
static bool ShouldAdvertiseBluetoothMacOverBle(PowerLevel power_level);
static bool ShouldAcceptBluetoothConnections(
const ConnectionOptions& options);
// Bluetooth
bool IsRecognizedBluetoothEndpoint(const std::string& name_string,
@@ -169,7 +174,8 @@ class P2pClusterPcpHandler : public BasePcpHandler {
const ByteArray& local_endpoint_info, const ConnectionOptions& options);
proto::connections::Medium StartBleScanning(
BleDiscoveredPeripheralCallback callback, ClientProxy* client,
const std::string& service_id);
const std::string& service_id,
const std::string& fast_advertisement_service_uuid);
BasePcpHandler::ConnectImplResult BleConnectImpl(ClientProxy* client,
BleEndpoint* endpoint);
cpp/core_v2/internal/p2p_cluster_pcp_handler_test.cc
+21 -8
View File
@@ -16,6 +16,7 @@
#include <memory>
#include "core_v2/internal/bwu_manager.h"
#include "core_v2/options.h"
#include "platform_v2/base/medium_environment.h"
#include "platform_v2/public/count_down_latch.h"
@@ -75,7 +76,8 @@ TEST_P(P2pClusterPcpHandlerTest, CanConstructOne) {
Mediums mediums;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
P2pClusterPcpHandler handler(&mediums, &em, &ecm);
BwuManager bwu(mediums, em, ecm, {}, {});
P2pClusterPcpHandler handler(&mediums, &em, &ecm, &bwu);
env_.Stop();
}
@@ -87,8 +89,10 @@ TEST_P(P2pClusterPcpHandlerTest, CanConstructMultiple) {
EndpointChannelManager ecm_b;
EndpointManager em_a(&ecm_a);
EndpointManager em_b(&ecm_b);
P2pClusterPcpHandler handler_a(&mediums_a, &em_a, &ecm_a);
P2pClusterPcpHandler handler_b(&mediums_b, &em_b, &ecm_b);
BwuManager bwu_a(mediums_a, em_a, ecm_a, {}, {});
BwuManager bwu_b(mediums_b, em_b, ecm_b, {}, {});
P2pClusterPcpHandler handler_a(&mediums_a, &em_a, &ecm_a, &bwu_a);
P2pClusterPcpHandler handler_b(&mediums_b, &em_b, &ecm_b, &bwu_b);
env_.Stop();
}
@@ -98,7 +102,8 @@ TEST_P(P2pClusterPcpHandlerTest, CanAdvertise) {
Mediums mediums_a;
EndpointChannelManager ecm_a;
EndpointManager em_a(&ecm_a);
P2pClusterPcpHandler handler_a(&mediums_a, &em_a, &ecm_a);
BwuManager bwu_a(mediums_a, em_a, ecm_a, {}, {});
P2pClusterPcpHandler handler_a(&mediums_a, &em_a, &ecm_a, &bwu_a);
EXPECT_EQ(
handler_a.StartAdvertising(&client_a_, service_id_, options_,
{.endpoint_info = ByteArray{endpoint_name}}),
@@ -115,8 +120,10 @@ TEST_P(P2pClusterPcpHandlerTest, CanDiscover) {
EndpointChannelManager ecm_b;
EndpointManager em_a(&ecm_a);
EndpointManager em_b(&ecm_b);
P2pClusterPcpHandler handler_a(&mediums_a, &em_a, &ecm_a);
P2pClusterPcpHandler handler_b(&mediums_b, &em_b, &ecm_b);
BwuManager bwu_a(mediums_a, em_a, ecm_a, {}, {});
BwuManager bwu_b(mediums_b, em_b, ecm_b, {}, {});
P2pClusterPcpHandler handler_a(&mediums_a, &em_a, &ecm_a, &bwu_a);
P2pClusterPcpHandler handler_b(&mediums_b, &em_b, &ecm_b, &bwu_b);
CountDownLatch latch(1);
EXPECT_EQ(
handler_a.StartAdvertising(&client_a_, service_id_, options_,
@@ -155,8 +162,12 @@ TEST_P(P2pClusterPcpHandlerTest, CanConnect) {
EndpointChannelManager ecm_b;
EndpointManager em_a(&ecm_a);
EndpointManager em_b(&ecm_b);
P2pClusterPcpHandler handler_a(&mediums_a, &em_a, &ecm_a);
P2pClusterPcpHandler handler_b(&mediums_b, &em_b, &ecm_b);
BwuManager bwu_a(mediums_a, em_a, ecm_a, {},
{.allow_upgrade_to = {.bluetooth = true}});
BwuManager bwu_b(mediums_b, em_b, ecm_b, {},
{.allow_upgrade_to = {.bluetooth = true}});
P2pClusterPcpHandler handler_a(&mediums_a, &em_a, &ecm_a, &bwu_a);
P2pClusterPcpHandler handler_b(&mediums_b, &em_b, &ecm_b, &bwu_b);
CountDownLatch discover_latch(1);
CountDownLatch connect_latch(2);
struct DiscoveredInfo {
@@ -221,6 +232,8 @@ TEST_P(P2pClusterPcpHandlerTest, CanConnect) {
},
options_);
EXPECT_TRUE(connect_latch.Await(absl::Milliseconds(1000)).result());
bwu_a.Shutdown();
bwu_b.Shutdown();
env_.Stop();
}
cpp/core_v2/internal/p2p_point_to_point_pcp_handler.cc
+3 -2
View File
@@ -20,8 +20,9 @@ namespace connections {
P2pPointToPointPcpHandler::P2pPointToPointPcpHandler(
Mediums& mediums, EndpointManager& endpoint_manager,
EndpointChannelManager& channel_manager, Pcp pcp)
: P2pStarPcpHandler(mediums, endpoint_manager, channel_manager, pcp) {}
EndpointChannelManager& channel_manager, BwuManager& bwu_manager, Pcp pcp)
: P2pStarPcpHandler(mediums, endpoint_manager, channel_manager, bwu_manager,
pcp) {}
std::vector<proto::connections::Medium>
P2pPointToPointPcpHandler::GetConnectionMediumsByPriority() {
cpp/core_v2/internal/p2p_point_to_point_pcp_handler.h
+1
View File
@@ -36,6 +36,7 @@ class P2pPointToPointPcpHandler : public P2pStarPcpHandler {
public:
P2pPointToPointPcpHandler(Mediums& mediums, EndpointManager& endpoint_manager,
EndpointChannelManager& channel_manager,
BwuManager& bwu_manager,
Pcp pcp = Pcp::kP2pPointToPoint);
protected:
cpp/core_v2/internal/p2p_star_pcp_handler.cc
+3 -3
View File
@@ -23,9 +23,9 @@ namespace connections {
P2pStarPcpHandler::P2pStarPcpHandler(Mediums& mediums,
EndpointManager& endpoint_manager,
EndpointChannelManager& channel_manager,
Pcp pcp)
: P2pClusterPcpHandler(&mediums, &endpoint_manager, &channel_manager, pcp) {
}
BwuManager& bwu_manager, Pcp pcp)
: P2pClusterPcpHandler(&mediums, &endpoint_manager, &channel_manager,
&bwu_manager, pcp) {}
std::vector<proto::connections::Medium>
P2pStarPcpHandler::GetConnectionMediumsByPriority() {
cpp/core_v2/internal/p2p_star_pcp_handler.h
+1
View File
@@ -39,6 +39,7 @@ class P2pStarPcpHandler : public P2pClusterPcpHandler {
public:
P2pStarPcpHandler(Mediums& mediums, EndpointManager& endpoint_manager,
EndpointChannelManager& channel_manager,
BwuManager& bwu_manager,
Pcp pcp = Pcp::kP2pStar);
protected:
cpp/core_v2/internal/pcp_manager.cc
+5 -4
View File
@@ -25,14 +25,15 @@ namespace connections {
PcpManager::PcpManager(Mediums& mediums,
EndpointChannelManager& channel_manager,
EndpointManager& endpoint_manager) {
EndpointManager& endpoint_manager,
BwuManager& bwu_manager) {
handlers_[Pcp::kP2pCluster] = std::make_unique<P2pClusterPcpHandler>(
&mediums, &endpoint_manager, &channel_manager);
&mediums, &endpoint_manager, &channel_manager, &bwu_manager);
handlers_[Pcp::kP2pStar] = std::make_unique<P2pStarPcpHandler>(
mediums, endpoint_manager, channel_manager);
mediums, endpoint_manager, channel_manager, bwu_manager);
handlers_[Pcp::kP2pPointToPoint] =
std::make_unique<P2pPointToPointPcpHandler>(mediums, endpoint_manager,
channel_manager);
channel_manager, bwu_manager);
}
void PcpManager::DisconnectFromEndpointManager() {
cpp/core_v2/internal/pcp_manager.h
+2 -1
View File
@@ -18,6 +18,7 @@
#include <string>
#include "core_v2/internal/base_pcp_handler.h"
#include "core_v2/internal/bwu_manager.h"
#include "core_v2/internal/client_proxy.h"
#include "core_v2/internal/endpoint_channel_manager.h"
#include "core_v2/internal/endpoint_manager.h"
@@ -43,7 +44,7 @@ namespace connections {
class PcpManager {
public:
PcpManager(Mediums& mediums, EndpointChannelManager& channel_manager,
EndpointManager& endpoint_manager);
EndpointManager& endpoint_manager, BwuManager& bwu_manager);
~PcpManager();
Status StartAdvertising(ClientProxy* client, const string& service_id,
cpp/core_v2/internal/simulation_user.h
+3 -1
View File
@@ -17,6 +17,7 @@
#include <string>
#include "core_v2/internal/bwu_manager.h"
#include "core_v2/internal/client_proxy.h"
#include "core_v2/internal/endpoint_channel_manager.h"
#include "core_v2/internal/endpoint_manager.h"
@@ -144,7 +145,8 @@ class SimulationUser {
ClientProxy client_;
EndpointChannelManager ecm_;
EndpointManager em_{&ecm_};
PcpManager mgr_{mediums_, ecm_, em_};
BwuManager bwu_{mediums_, em_, ecm_, {}, {}};
PcpManager mgr_{mediums_, ecm_, em_, bwu_};
PayloadManager pm_{em_};
};
cpp/core_v2/options.h
+9
View File
@@ -79,6 +79,13 @@ struct MediumSelector {
// Feature On/Off switch for mediums.
using BooleanMediumSelector = MediumSelector<bool>;
// Represents the various power levels that can be used, on mediums that support
// it.
enum class PowerLevel {
kHighPower = 0,
kLowPower = 1,
};
// Connection Options: used for both Advertising and Discovery.
// All fields are mutable, to make the type copy-assignable.
struct ConnectionOptions {
@@ -86,6 +93,8 @@ struct ConnectionOptions {
BooleanMediumSelector allowed{BooleanMediumSelector().SetAll(true)};
bool auto_upgrade_bandwidth;
bool enforce_topology_constraints;
bool low_power;
bool enable_bluetooth_listening;
ByteArray remote_bluetooth_mac_address;
std::string fast_advertisement_service_uuid;
// Verify if ConnectionOptions is in a not-initialized (Empty) state.
cpp/platform/BUILD
-1
View File
@@ -75,7 +75,6 @@ cc_library(
visibility = [
"//googlemac/iPhone/Shared/Nearby/Connections:__subpackages__",
"//core:__subpackages__",
"//platform_v2/base:__pkg__",
],
deps = [
"//absl/base",
cpp/platform_v2/api/ble.h
+1
View File
@@ -89,6 +89,7 @@ class BleMedium {
// Returns true once the BLE scan has been initiated.
virtual bool StartScanning(const std::string& service_id,
const std::string& fast_advertisement_service_uuid,
DiscoveredPeripheralCallback callback) = 0;
// Returns true once BLE scanning for service_id is well and truly stopped;
cpp/platform_v2/api/bluetooth_classic.h
+1 -1
View File
@@ -150,7 +150,7 @@ class BluetoothClassicMedium {
virtual std::unique_ptr<BluetoothServerSocket> ListenForService(
const std::string& service_name, const std::string& service_uuid) = 0;
virtual BluetoothDevice* FindRemoteDevice(const std::string& mac_address) = 0;
virtual BluetoothDevice* GetRemoteDevice(const std::string& mac_address) = 0;
};
} // namespace api
cpp/platform_v2/api/platform.h
+2 -3
View File
@@ -56,7 +56,7 @@ class ImplementationPlatform {
// - synchronization primitives:
// - mutex (regular, and recursive)
// - condition variable (must work with regular mutex only)
// - Future<T> : to synchronize on Callable<T> schduled to execute.
// - Future<T> : to synchronize on Callable<T> scheduled to execute.
// - CountDownLatch : to ensure at least N threads are waiting.
// - file I/O
// - Logging
@@ -72,8 +72,7 @@ class ImplementationPlatform {
// Supports enums and integers up to 32-bit.
// Does not use locking, if platform supports 32-bit atimics natively.
// Does not use dynamic memory allocations in operations.
static std::unique_ptr<AtomicUint32>
CreateAtomicUint32(std::uint32_t value);
static std::unique_ptr<AtomicUint32> CreateAtomicUint32(std::uint32_t value);
static std::unique_ptr<CountDownLatch> CreateCountDownLatch(
std::int32_t count);
cpp/platform_v2/base/medium_environment.cc
+9 -5
View File
@@ -354,10 +354,12 @@ void MediumEnvironment::UpdateBleMediumForAdvertising(
void MediumEnvironment::UpdateBleMediumForScanning(
api::BleMedium& medium, const std::string& service_id,
const std::string& fast_advertisement_service_uuid,
BleDiscoveredPeripheralCallback callback, bool enabled) {
if (!enabled_) return;
RunOnMediumEnvironmentThread(
[this, &medium, service_id, callback = std::move(callback), enabled]() {
[this, &medium, service_id, fast_advertisement_service_uuid,
callback = std::move(callback), enabled]() {
auto item = ble_mediums_.find(&medium);
if (item == ble_mediums_.end()) {
NEARBY_LOG(INFO,
@@ -367,10 +369,12 @@ void MediumEnvironment::UpdateBleMediumForScanning(
}
auto& context = item->second;
context.discovery_callback = std::move(callback);
NEARBY_LOG(INFO,
"Update Ble medium for scanning: this=%p; medium=%p; "
"service_id=%s; enabled=%d ;",
this, &medium, service_id.c_str(), enabled);
NEARBY_LOG(
INFO,
"Update Ble medium for scanning: this=%p; medium=%p; "
"service_id=%s; fast_advertisement_service_uuid=%s; enabled=%d ;",
this, &medium, service_id.c_str(),
fast_advertisement_service_uuid.c_str(), enabled);
for (auto& medium_info : ble_mediums_) {
auto& local_medium = medium_info.first;
auto& info = medium_info.second;
cpp/platform_v2/base/medium_environment.h
+4 -4
View File
@@ -165,10 +165,10 @@ class MediumEnvironment {
// This should be called when discoverable state changes.
// with user-specified callback when discovery is enabled, and with default
// (empty) callback otherwise.
void UpdateBleMediumForScanning(api::BleMedium& medium,
const std::string& service_id,
BleDiscoveredPeripheralCallback callback,
bool enabled);
void UpdateBleMediumForScanning(
api::BleMedium& medium, const std::string& service_id,
const std::string& fast_advertisement_service_uuid,
BleDiscoveredPeripheralCallback callback, bool enabled);
// Updates Accepted connection callback info to allow for dispatch of
// advertising events.
cpp/platform_v2/impl/g3/ble.cc
+8 -4
View File
@@ -266,11 +266,15 @@ bool BleMedium::StopAdvertising(const std::string& service_id) {
return true;
}
bool BleMedium::StartScanning(const std::string& service_id,
DiscoveredPeripheralCallback callback) {
bool BleMedium::StartScanning(
const std::string& service_id,
const std::string& fast_advertisement_service_uuid,
DiscoveredPeripheralCallback callback) {
NEARBY_LOGS(INFO) << "G3 Ble StartScanning: service_id=" << service_id;
auto& env = MediumEnvironment::Instance();
env.UpdateBleMediumForScanning(*this, service_id, std::move(callback), true);
env.UpdateBleMediumForScanning(*this, service_id,
fast_advertisement_service_uuid,
std::move(callback), true);
{
absl::MutexLock lock(&mutex_);
scanning_info_.service_id = service_id;
@@ -291,7 +295,7 @@ bool BleMedium::StopScanning(const std::string& service_id) {
}
auto& env = MediumEnvironment::Instance();
env.UpdateBleMediumForScanning(*this, service_id, {}, false);
env.UpdateBleMediumForScanning(*this, service_id, {}, {}, false);
return true;
}
cpp/platform_v2/impl/g3/ble.h
+1
View File
@@ -160,6 +160,7 @@ class BleMedium : public api::BleMedium {
// Returns true once the Ble scanning has been initiated.
bool StartScanning(const std::string& service_id,
const std::string& fast_advertisement_service_uuid,
DiscoveredPeripheralCallback callback) override
ABSL_LOCKS_EXCLUDED(mutex_);
cpp/platform_v2/impl/g3/bluetooth_classic.cc
+1 -1
View File
@@ -254,7 +254,7 @@ BluetoothClassicMedium::ListenForService(const std::string& service_name,
return socket;
}
api::BluetoothDevice* BluetoothClassicMedium::FindRemoteDevice(
api::BluetoothDevice* BluetoothClassicMedium::GetRemoteDevice(
const std::string& mac_address) {
auto& env = MediumEnvironment::Instance();
return env.FindBluetoothDevice(mac_address);
cpp/platform_v2/impl/g3/bluetooth_classic.h
+1 -1
View File
@@ -221,7 +221,7 @@ class BluetoothClassicMedium : public api::BluetoothClassicMedium {
const std::string& service_name, const std::string& service_uuid) override
ABSL_LOCKS_EXCLUDED(mutex_);
api::BluetoothDevice* FindRemoteDevice(
api::BluetoothDevice* GetRemoteDevice(
const std::string& mac_address) override;
private:
cpp/platform_v2/impl/ios/BUILD
+56
View File
@@ -0,0 +1,56 @@
objc_library(
name = "types",
srcs = [
"log_message.mm",
"scheduled_executor.mm",
],
hdrs = [
"atomic_boolean.h",
"atomic_reference.h",
"condition_variable.h",
"count_down_latch.h",
"log_message.h",
"multi_thread_executor.h",
"mutex.h",
"scheduled_executor.h",
"single_thread_executor.h",
],
visibility = [
"//platform_v2/impl/ios:__pkg__",
],
deps = [
"//base",
"//platform_v2/api:platform",
"//platform_v2/api:types",
"//platform_v2/base",
"//platform_v2/base:util",
"//platform_v2/impl/shared:posix_mutex",
"//absl/base:core_headers",
"//absl/synchronization",
"//absl/time",
"//thread",
],
)
objc_library(
name = "ios",
srcs = [
"platform.mm",
],
visibility = [
"//googlemac/iPhone/Shared/Nearby/Connections:__subpackages__",
"//core_v2:__subpackages__",
"//platform_v2:__subpackages__",
],
deps = [
":types",
"//platform_v2/api:comm",
"//platform_v2/api:platform",
"//platform_v2/api:types",
"//platform_v2/impl/shared:file",
"//absl/base:core_headers",
"//absl/memory",
"//absl/strings",
"//absl/time",
],
)
cpp/platform_v2/impl/ios/atomic_boolean.h
+28
View File
@@ -0,0 +1,28 @@
#ifndef PLATFORM_V2_IMPL_IOS_ATOMIC_BOOLEAN_H_
#define PLATFORM_V2_IMPL_IOS_ATOMIC_BOOLEAN_H_
#include <atomic>
#include "platform_v2/api/atomic_boolean.h"
namespace location {
namespace nearby {
namespace ios {
class AtomicBoolean : public api::AtomicBoolean {
public:
explicit AtomicBoolean(bool initial_value) : value_(initial_value) {}
~AtomicBoolean() override = default;
bool Get() const override { return value_.load(); }
bool Set(bool value) override { return value_.exchange(value); }
private:
std::atomic_bool value_;
};
} // namespace ios
} // namespace nearby
} // namespace location
#endif // PLATFORM_V2_IMPL_IOS_ATOMIC_BOOLEAN_H_
cpp/platform_v2/impl/ios/atomic_reference.h
+33
View File
@@ -0,0 +1,33 @@
#ifndef PLATFORM_V2_IMPL_IOS_ATOMIC_REFERENCE_H_
#define PLATFORM_V2_IMPL_IOS_ATOMIC_REFERENCE_H_
#include <atomic>
#include <cstdint>
#include "platform_v2/api/atomic_reference.h"
namespace location {
namespace nearby {
namespace ios {
class AtomicUint32 : public api::AtomicUint32 {
public:
explicit AtomicUint32(std::int32_t value) : value_(value) {}
~AtomicUint32() override = default;
std::uint32_t Get() const override {
return value_;
}
void Set(std::uint32_t value) override {
value_ = value;
}
private:
std::atomic<std::uint32_t> value_;
};
} // namespace ios
} // namespace nearby
} // namespace location
#endif // PLATFORM_V2_IMPL_IOS_ATOMIC_REFERENCE_H_
cpp/platform_v2/impl/ios/condition_variable.h
+37
View File
@@ -0,0 +1,37 @@
#ifndef PLATFORM_V2_IMPL_IOS_CONDITION_VARIABLE_H_
#define PLATFORM_V2_IMPL_IOS_CONDITION_VARIABLE_H_
#include "platform_v2/api/condition_variable.h"
#include "platform_v2/base/exception.h"
#include "platform_v2/impl/ios/mutex.h"
#include "absl/synchronization/mutex.h"
namespace location {
namespace nearby {
namespace ios {
class ConditionVariable : public api::ConditionVariable {
public:
explicit ConditionVariable(ios::Mutex* mutex) : mutex_(&mutex->mutex_) {}
~ConditionVariable() override = default;
Exception Wait() override {
cond_var_.Wait(mutex_);
return {Exception::kSuccess};
}
Exception Wait(absl::Duration timeout) override {
cond_var_.WaitWithTimeout(mutex_, timeout);
return {Exception::kSuccess};
}
void Notify() override { cond_var_.SignalAll(); }
private:
absl::Mutex* mutex_;
absl::CondVar cond_var_;
};
} // namespace ios
} // namespace nearby
} // namespace location
#endif // PLATFORM_V2_IMPL_IOS_CONDITION_VARIABLE_H_
cpp/platform_v2/impl/ios/count_down_latch.h
+55
View File
@@ -0,0 +1,55 @@
#ifndef PLATFORM_V2_IMPL_IOS_COUNT_DOWN_LATCH_H_
#define PLATFORM_V2_IMPL_IOS_COUNT_DOWN_LATCH_H_
#include "platform_v2/api/count_down_latch.h"
#include "absl/base/thread_annotations.h"
#include "absl/synchronization/mutex.h"
#include "absl/time/clock.h"
namespace location {
namespace nearby {
namespace ios {
class CountDownLatch final : public api::CountDownLatch {
public:
explicit CountDownLatch(int count) : count_(count) {}
CountDownLatch(const CountDownLatch&) = delete;
CountDownLatch& operator=(const CountDownLatch&) = delete;
CountDownLatch(CountDownLatch&&) = delete;
CountDownLatch& operator=(CountDownLatch&&) = delete;
ExceptionOr<bool> Await(absl::Duration timeout) override {
absl::MutexLock lock(&mutex_);
absl::Time deadline = absl::Now() + timeout;
while (count_ > 0) {
if (cond_.WaitWithDeadline(&mutex_, deadline)) {
return ExceptionOr<bool>(false);
}
}
return ExceptionOr<bool>(true);
}
Exception Await() override {
absl::MutexLock lock(&mutex_);
while (count_ > 0) {
cond_.Wait(&mutex_);
}
return {Exception::kSuccess};
}
void CountDown() override {
absl::MutexLock lock(&mutex_);
if (count_ > 0 && --count_ == 0) {
cond_.SignalAll();
}
}
private:
absl::Mutex mutex_; // Mutex to be used with cond_.Wait...() method family.
absl::CondVar cond_; // Condition to synchronize up to N waiting threads.
int count_
ABSL_GUARDED_BY(mutex_); // When zero, latch should release all waiters.
};
} // namespace ios
} // namespace nearby
} // namespace location
#endif // PLATFORM_V2_IMPL_IOS_COUNT_DOWN_LATCH_H_
cpp/platform_v2/impl/ios/log_message.h
+28
View File
@@ -0,0 +1,28 @@
#ifndef PLATFORM_V2_IMPL_IOS_LOG_MESSAGE_H_
#define PLATFORM_V2_IMPL_IOS_LOG_MESSAGE_H_
#include "base/logging.h"
#include "platform_v2/api/log_message.h"
namespace location {
namespace nearby {
namespace ios {
class LogMessage : public api::LogMessage {
public:
LogMessage(const char* file, int line, Severity severity);
~LogMessage() override;
void Print(const char* format, ...) override;
std::ostream& Stream() override;
private:
absl::LogStreamer log_streamer_;
};
} // namespace ios
} // namespace nearby
} // namespace location
#endif // PLATFORM_V2_IMPL_IOS_LOG_MESSAGE_H_
cpp/platform_v2/impl/ios/log_message.mm
+56
View File
@@ -0,0 +1,56 @@
#include "platform_v2/impl/ios/log_message.h"
#include <algorithm>
#include "base/stringprintf.h"
namespace location {
namespace nearby {
namespace ios {
api::LogMessage::Severity kMinLogSeverity = api::LogMessage::Severity::kInfo;
inline absl::LogSeverity ConvertSeverity(api::LogMessage::Severity severity) {
switch (severity) {
case api::LogMessage::Severity::kInfo:
return absl::LogSeverity::kInfo;
case api::LogMessage::Severity::kWarning:
return absl::LogSeverity::kWarning;
case api::LogMessage::Severity::kError:
return absl::LogSeverity::kError;
case api::LogMessage::Severity::kFatal:
return absl::LogSeverity::kFatal;
}
}
LogMessage::LogMessage(const char* file, int line, Severity severity)
: log_streamer_(ConvertSeverity(severity), file, line) {}
LogMessage::~LogMessage() = default;
void LogMessage::Print(const char* format, ...) {
va_list ap;
va_start(ap, format);
std::string result;
StringAppendV(&result, format, ap);
log_streamer_.stream() << result;
va_end(ap);
}
std::ostream& LogMessage::Stream() { return log_streamer_.stream(); }
} // namespace ios
namespace api {
void LogMessage::SetMinLogSeverity(Severity severity) {
ios::kMinLogSeverity = severity;
}
bool LogMessage::ShouldCreateLogMessage(Severity severity) {
return severity >= ios::kMinLogSeverity;
}
} // namespace api
} // namespace nearby
} // namespace location
cpp/platform_v2/impl/ios/multi_thread_executor.h
+57
View File
@@ -0,0 +1,57 @@
#ifndef PLATFORM_V2_IMPL_IOS_MULTI_THREAD_EXECUTOR_H_
#define PLATFORM_V2_IMPL_IOS_MULTI_THREAD_EXECUTOR_H_
#include <atomic>
#include "platform_v2/api/submittable_executor.h"
#include "platform_v2/impl/ios/count_down_latch.h"
#include "absl/time/clock.h"
#include "thread/threadpool.h"
namespace location {
namespace nearby {
namespace ios {
class MultiThreadExecutor : public api::SubmittableExecutor {
public:
explicit MultiThreadExecutor(int max_parallelism)
: thread_pool_(max_parallelism) {
thread_pool_.StartWorkers();
}
void Execute(Runnable&& runnable) override {
if (!shutdown_) {
thread_pool_.Schedule(std::move(runnable));
}
}
bool DoSubmit(Runnable&& runnable) override {
if (shutdown_) return false;
thread_pool_.Schedule(std::move(runnable));
return true;
}
void Shutdown() override { DoShutdown(); }
~MultiThreadExecutor() override { DoShutdown(); }
int GetTid(int index) const override {
const auto* thread = thread_pool_.thread(index);
return thread ? *(int*)(thread->tid()) : 0;
}
void ScheduleAfter(absl::Duration delay, Runnable&& runnable) {
if (shutdown_) return;
thread_pool_.ScheduleAt(absl::Now() + delay, std::move(runnable));
}
bool InShutdown() const { return shutdown_; }
private:
void DoShutdown() {
shutdown_ = true;
}
std::atomic_bool shutdown_ = false;
ThreadPool thread_pool_;
};
} // namespace ios
} // namespace nearby
} // namespace location
#endif // PLATFORM_V2_IMPL_IOS_MULTI_THREAD_EXECUTOR_H_
cpp/platform_v2/impl/ios/mutex.h
+47
View File
@@ -0,0 +1,47 @@
#ifndef PLATFORM_V2_IMPL_IOS_MUTEX_H_
#define PLATFORM_V2_IMPL_IOS_MUTEX_H_
#include "platform_v2/api/mutex.h"
#include "platform_v2/impl/shared/posix_mutex.h"
#include "absl/synchronization/mutex.h"
namespace location {
namespace nearby {
namespace ios {
class ABSL_LOCKABLE Mutex : public api::Mutex {
public:
explicit Mutex(bool check) : check_(check) {}
~Mutex() override = default;
Mutex(Mutex&&) = delete;
Mutex& operator=(Mutex&&) = delete;
Mutex(const Mutex&) = delete;
Mutex& operator=(const Mutex&) = delete;
void Lock() ABSL_EXCLUSIVE_LOCK_FUNCTION() override {
mutex_.Lock();
if (!check_) mutex_.ForgetDeadlockInfo();
}
void Unlock() ABSL_UNLOCK_FUNCTION() override { mutex_.Unlock(); }
private:
friend class ConditionVariable;
absl::Mutex mutex_;
bool check_;
};
class ABSL_LOCKABLE RecursiveMutex : public posix::Mutex {
public:
~RecursiveMutex() override = default;
RecursiveMutex() = default;
RecursiveMutex(RecursiveMutex&&) = delete;
RecursiveMutex& operator=(RecursiveMutex&&) = delete;
RecursiveMutex(const RecursiveMutex&) = delete;
RecursiveMutex& operator=(const RecursiveMutex&) = delete;
};
} // namespace ios
} // namespace nearby
} // namespace location
#endif // PLATFORM_V2_IMPL_IOS_MUTEX_H_
cpp/platform_v2/impl/ios/platform.mm
+124
View File
@@ -0,0 +1,124 @@
#include "platform_v2/api/platform.h"
#include <atomic>
#include <memory>
#include "platform_v2/api/atomic_boolean.h"
#include "platform_v2/api/atomic_reference.h"
#include "platform_v2/api/condition_variable.h"
#include "platform_v2/api/count_down_latch.h"
#include "platform_v2/api/log_message.h"
#include "platform_v2/api/mutex.h"
#include "platform_v2/api/scheduled_executor.h"
#include "platform_v2/api/submittable_executor.h"
#include "platform_v2/impl/ios/atomic_boolean.h"
#include "platform_v2/impl/ios/atomic_reference.h"
#include "platform_v2/impl/ios/condition_variable.h"
#include "platform_v2/impl/ios/count_down_latch.h"
#include "platform_v2/impl/ios/log_message.h"
#include "platform_v2/impl/ios/multi_thread_executor.h"
#include "platform_v2/impl/ios/mutex.h"
#include "platform_v2/impl/ios/scheduled_executor.h"
#include "platform_v2/impl/ios/single_thread_executor.h"
#include "platform_v2/impl/shared/file.h"
#include "absl/memory/memory.h"
namespace location {
namespace nearby {
namespace api {
namespace {
std::string GetPayloadPath(PayloadId payload_id) {
return absl::StrCat("/tmp/", payload_id);
}
} // namespace
std::unique_ptr<AtomicBoolean> ImplementationPlatform::CreateAtomicBoolean(bool initial_value) {
return absl::make_unique<ios::AtomicBoolean>(initial_value);
}
std::unique_ptr<AtomicUint32> ImplementationPlatform::CreateAtomicUint32(std::uint32_t value) {
return absl::make_unique<ios::AtomicUint32>(value);
}
std::unique_ptr<CountDownLatch> ImplementationPlatform::CreateCountDownLatch(
std::int32_t count) {
return absl::make_unique<ios::CountDownLatch>(count);
}
std::unique_ptr<Mutex> ImplementationPlatform::CreateMutex(Mutex::Mode mode) {
if (mode == Mutex::Mode::kRecursive)
return absl::make_unique<ios::RecursiveMutex>();
else
return absl::make_unique<ios::Mutex>(mode == Mutex::Mode::kRegular);
}
std::unique_ptr<ConditionVariable> ImplementationPlatform::CreateConditionVariable(Mutex* mutex) {
return std::unique_ptr<ConditionVariable>(
new ios::ConditionVariable(static_cast<ios::Mutex*>(mutex)));
}
std::unique_ptr<InputFile> ImplementationPlatform::CreateInputFile(PayloadId payload_id,
std::int64_t total_size) {
return absl::make_unique<shared::InputFile>(GetPayloadPath(payload_id), total_size);
}
std::unique_ptr<OutputFile> ImplementationPlatform::CreateOutputFile(PayloadId payload_id) {
return absl::make_unique<shared::OutputFile>(GetPayloadPath(payload_id));
}
std::unique_ptr<LogMessage> ImplementationPlatform::CreateLogMessage(
const char* file, int line, LogMessage::Severity severity) {
return absl::make_unique<ios::LogMessage>(file, line, severity);
}
std::unique_ptr<SubmittableExecutor> ImplementationPlatform::CreateSingleThreadExecutor() {
return absl::make_unique<ios::SingleThreadExecutor>();
}
std::unique_ptr<SubmittableExecutor> ImplementationPlatform::CreateMultiThreadExecutor(
int max_concurrency) {
return absl::make_unique<ios::MultiThreadExecutor>(max_concurrency);
}
std::unique_ptr<ScheduledExecutor> ImplementationPlatform::CreateScheduledExecutor() {
return absl::make_unique<ios::ScheduledExecutor>();
}
std::unique_ptr<BluetoothAdapter> ImplementationPlatform::CreateBluetoothAdapter() {
return std::unique_ptr<BluetoothAdapter>();
}
std::unique_ptr<BluetoothClassicMedium> ImplementationPlatform::CreateBluetoothClassicMedium(
api::BluetoothAdapter& adapter) {
return std::unique_ptr<BluetoothClassicMedium>();
}
std::unique_ptr<BleMedium> ImplementationPlatform::CreateBleMedium(api::BluetoothAdapter& adapter) {
return std::unique_ptr<BleMedium>();
}
std::unique_ptr<ble_v2::BleMedium> ImplementationPlatform::CreateBleV2Medium(
api::BluetoothAdapter& adapter) {
return std::unique_ptr<ble_v2::BleMedium>();
}
std::unique_ptr<ServerSyncMedium> ImplementationPlatform::CreateServerSyncMedium() {
return std::unique_ptr<ServerSyncMedium>();
}
std::unique_ptr<WifiMedium> ImplementationPlatform::CreateWifiMedium() {
return std::unique_ptr<WifiMedium>();
}
std::unique_ptr<WifiLanMedium> ImplementationPlatform::CreateWifiLanMedium() {
return std::unique_ptr<WifiLanMedium>();
}
std::unique_ptr<WebRtcMedium> ImplementationPlatform::CreateWebRtcMedium() {
return std::unique_ptr<WebRtcMedium>();
}
} // namespace api
} // namespace nearby
} // namespace location
cpp/platform_v2/impl/ios/scheduled_executor.h
+43
View File
@@ -0,0 +1,43 @@
#ifndef PLATFORM_V2_IMPL_IOS_SCHEDULED_EXECUTOR_H_
#define PLATFORM_V2_IMPL_IOS_SCHEDULED_EXECUTOR_H_
#include <atomic>
#include <memory>
#include "platform_v2/api/cancelable.h"
#include "platform_v2/api/scheduled_executor.h"
#include "platform_v2/base/runnable.h"
#include "platform_v2/impl/ios/single_thread_executor.h"
#include "absl/time/clock.h"
#include "thread/threadpool.h"
namespace location {
namespace nearby {
namespace ios {
class ScheduledExecutor final : public api::ScheduledExecutor {
public:
ScheduledExecutor() = default;
~ScheduledExecutor() override {
executor_.Shutdown();
}
void Execute(Runnable&& runnable) override {
executor_.Execute(std::move(runnable));
}
std::shared_ptr<api::Cancelable> Schedule(Runnable&& runnable,
absl::Duration delay) override;
void Shutdown() override { executor_.Shutdown(); }
int GetTid(int index) const override {
return executor_.GetTid(index);
}
private:
SingleThreadExecutor executor_;
};
} // namespace ios
} // namespace nearby
} // namespace location
#endif // PLATFORM_V2_IMPL_IOS_SCHEDULED_EXECUTOR_H_
cpp/platform_v2/impl/ios/scheduled_executor.mm
+65
View File
@@ -0,0 +1,65 @@
#include "platform_v2/impl/ios/scheduled_executor.h"
#include <atomic>
#include <memory>
#include "platform_v2/api/cancelable.h"
#include "platform_v2/base/runnable.h"
#include "absl/time/clock.h"
namespace location {
namespace nearby {
namespace ios {
namespace {
class ScheduledCancelable : public api::Cancelable {
public:
bool Cancel() override {
Status expected = kNotRun;
while (expected == kNotRun) {
if (status_.compare_exchange_strong(expected, kCanceled)) {
return true;
}
}
return false;
}
bool MarkExecuted() {
Status expected = kNotRun;
while (expected == kNotRun) {
if (status_.compare_exchange_strong(expected, kExecuted)) {
return true;
}
}
return false;
}
private:
enum Status {
kNotRun,
kExecuted,
kCanceled,
};
std::atomic<Status> status_ = kNotRun;
};
} // namespace
std::shared_ptr<api::Cancelable> ScheduledExecutor::Schedule(
Runnable&& runnable, absl::Duration delay) {
auto scheduled_cancelable = std::make_shared<ScheduledCancelable>();
if (executor_.InShutdown()) {
return scheduled_cancelable;
}
executor_.ScheduleAfter(
delay, [this, scheduled_cancelable, runnable(std::move(runnable))]() {
if (!executor_.InShutdown() && scheduled_cancelable->MarkExecuted()) {
runnable();
}
});
return scheduled_cancelable;
}
} // namespace ios
} // namespace nearby
} // namespace location
cpp/platform_v2/impl/ios/single_thread_executor.h
+20
View File
@@ -0,0 +1,20 @@
#ifndef PLATFORM_V2_IMPL_IOS_SINGLE_THREAD_EXECUTOR_H_
#define PLATFORM_V2_IMPL_IOS_SINGLE_THREAD_EXECUTOR_H_
#include "platform_v2/impl/ios/multi_thread_executor.h"
namespace location {
namespace nearby {
namespace ios {
class SingleThreadExecutor final : public MultiThreadExecutor {
public:
SingleThreadExecutor() : MultiThreadExecutor(1) {}
~SingleThreadExecutor() override = default;
};
} // namespace ios
} // namespace nearby
} // namespace location
#endif // PLATFORM_V2_IMPL_IOS_SINGLE_THREAD_EXECUTOR_H_
cpp/platform_v2/public/ble.cc
+5 -2
View File
@@ -31,8 +31,10 @@ bool BleMedium::StopAdvertising(const std::string& service_id) {
return impl_->StopAdvertising(service_id);
}
bool BleMedium::StartScanning(const std::string& service_id,
DiscoveredPeripheralCallback callback) {
bool BleMedium::StartScanning(
const std::string& service_id,
const std::string& fast_advertisement_service_uuid,
DiscoveredPeripheralCallback callback) {
{
MutexLock lock(&mutex_);
discovered_peripheral_callback_ = std::move(callback);
@@ -40,6 +42,7 @@ bool BleMedium::StartScanning(const std::string& service_id,
}
return impl_->StartScanning(
service_id,
fast_advertisement_service_uuid,
{
.peripheral_discovered_cb =
[this](api::BlePeripheral& peripheral,
cpp/platform_v2/public/ble.h
+1
View File
@@ -120,6 +120,7 @@ class BleMedium final {
// Returns true once the BLE scan has been initiated.
bool StartScanning(const std::string& service_id,
const std::string& fast_advertisement_service_uuid,
DiscoveredPeripheralCallback callback);
// Returns true once BLE scanning for service_id is well and truly stopped;
cpp/platform_v2/public/ble_test.cc
+5 -1
View File
@@ -29,7 +29,7 @@ namespace {
constexpr absl::Duration kWaitDuration = absl::Milliseconds(1000);
constexpr absl::string_view kServiceID{"com.google.location.nearby.apps.test"};
constexpr absl::string_view kAdvertisementString{"\x0a\x0b\x0c\x0d"};
constexpr absl::string_view kFastAdvertisementServiceUuid{"\xff\xfe"};
constexpr absl::string_view kFastAdvertisementServiceUuid{"\xf3\xfe"};
class BleMediumTest : public ::testing::Test {
protected:
@@ -73,6 +73,7 @@ TEST_F(BleMediumTest, CanStartAdvertising) {
EXPECT_TRUE(ble_b.StartScanning(
service_id,
fast_advertisement_service_uuid,
DiscoveredPeripheralCallback{
.peripheral_discovered_cb =
[&found_latch](
@@ -99,6 +100,7 @@ TEST_F(BleMediumTest, CanStartScanning) {
ble_a.StartScanning(
service_id,
fast_advertisement_service_uuid,
DiscoveredPeripheralCallback{
.peripheral_discovered_cb =
[&found_latch](
@@ -133,6 +135,7 @@ TEST_F(BleMediumTest, CanStopDiscovery) {
ble_a.StartScanning(
service_id,
fast_advertisement_service_uuid,
DiscoveredPeripheralCallback{
.peripheral_discovered_cb =
[&found_latch](
@@ -168,6 +171,7 @@ TEST_F(BleMediumTest, CanStartAcceptingConnectionsAndConnect) {
BlePeripheral* discovered_peripheral = nullptr;
ble_a.StartScanning(
service_id,
fast_advertisement_service_uuid,
DiscoveredPeripheralCallback{
.peripheral_discovered_cb =
[&found_latch, &discovered_peripheral](
cpp/platform_v2/public/bluetooth_classic.h
+2 -2
View File
@@ -202,8 +202,8 @@ class BluetoothClassicMedium final {
api::BluetoothClassicMedium& GetImpl() { return *impl_; }
BluetoothAdapter& GetAdapter() { return adapter_; }
std::string GetMacAddress() const { return adapter_.GetMacAddress(); }
BluetoothDevice FindRemoteDevice(const std::string& mac_address) {
return BluetoothDevice(impl_->FindRemoteDevice(mac_address));
BluetoothDevice GetRemoteDevice(const std::string& mac_address) {
return BluetoothDevice(impl_->GetRemoteDevice(mac_address));
}
private:
proto/connections/offline_wire_formats.proto
+12 -1
View File
@@ -95,8 +95,19 @@ message ConnectionResponseFrame {
//
// - ConnectionsStatusCodes.STATUS_OK
// - ConnectionsStatusCodes.STATUS_CONNECTION_REJECTED.
optional int32 status = 1;
optional int32 status = 1 [deprecated = true];
optional bytes handshake_data = 2;
// Used to replace the status integer parameter with a meaningful enum item.
// Map ConnectionsStatusCodes.STATUS_OK to ACCEPT and
// ConnectionsStatusCodes.STATUS_CONNECTION_REJECTED to REJECT.
// Flag: connection_replace_status_with_response_connectionResponseFrame
enum ResponseStatus {
UNKNOWN_RESPONSE_STATUS = 0;
ACCEPT = 1;
REJECT = 2;
}
optional ResponseStatus response = 3;
}
message PayloadTransferFrame {
proto/discovery_enums.proto
+5 -1
View File
@@ -20,7 +20,7 @@ option optimize_for = LITE_RUNTIME;
option java_package = "com.google.location.nearby.proto";
option java_outer_classname = "DiscoveryEnums";
// NEXT ID: 132
// NEXT ID: 133
enum DiscoveryEvent {
UNKNOWN_DISCOVERY_EVENT = 0;
@@ -404,6 +404,10 @@ enum DiscoveryEvent {
// User has seen a low battery notification.
FAST_PAIR_LOW_BATTERY_NOTIFICATION_SHOWN = 131;
// Connection Tracker Manager (Baymax) recovered the connection of the
// companion app.
FAST_PAIR_CONNECTION_TRACKER_RECOVER_COMPANION_APP = 132;
// Deprecated.
reserved 65, 67 to 72;
}
proto/error_code_enums.proto
+5
View File
@@ -422,4 +422,9 @@ enum Description {
SOCKET_NOT_BOUND = 141;
INVALID_REMOTE_ADDRESS = 142;
SOCKET_ALREADY_BOUND = 143;
HOTSPOT_NOT_STARTED = 144;
WEBRTC_ALREADY_INITIALIZED = 145;
INVALID_WEBRTC_STATE = 146;
NULL_DATA_CHANNEL = 147;
CREATE_OFFER_FAILED = 148;
}