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

Roll forward to cl/318932159

Browse Source 
Signed-off-by: Alexey Polyudov <apolyudov@google.com>
Change-Id: Ia0cddfc8cf46c66d5739bdb45ab7825422912082
This commit is contained in:
Alexey Polyudov
2020-06-30 00:27:57 -07:00
parent a4ef9ee564
commit 5142ef11af
39 changed files with 1229 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/core_v2/core.h
+3 -1
View File
@@ -4,6 +4,7 @@
#include <string>
#include "core_v2/internal/client_proxy.h"
#include "core_v2/internal/offline_service_controller.h"
#include "core_v2/internal/service_controller.h"
#include "core_v2/internal/service_controller_router.h"
#include "core_v2/listeners.h"
@@ -19,7 +20,8 @@ namespace connections {
// This class defines the API of the Nearby Connections Core library.
class Core {
public:
explicit Core(std::function<ServiceController*()> factory)
explicit Core(std::function<ServiceController*()> factory =
[]() { return new OfflineServiceController; })
: router_(factory) {}
~Core();
Core(Core&&) = default;
cpp/core_v2/internal/BUILD
+6
View File
@@ -13,6 +13,7 @@ cc_library(
"internal_payload.cc",
"internal_payload_factory.cc",
"offline_frames.cc",
"offline_service_controller.cc",
"p2p_cluster_pcp_handler.cc",
"p2p_point_to_point_pcp_handler.cc",
"p2p_star_pcp_handler.cc",
@@ -37,6 +38,7 @@ cc_library(
"internal_payload.h",
"internal_payload_factory.h",
"offline_frames.h",
"offline_service_controller.h",
"p2p_cluster_pcp_handler.h",
"p2p_point_to_point_pcp_handler.h",
"p2p_star_pcp_handler.h",
@@ -80,10 +82,12 @@ cc_library(
name = "internal_test",
testonly = True,
srcs = [
"offline_simulation_user.cc",
"simulation_user.cc",
],
hdrs = [
"mock_service_controller.h",
"offline_simulation_user.h",
"simulation_user.h",
],
visibility = [
@@ -96,6 +100,7 @@ cc_library(
"//platform_v2/public:types",
"//testing/base/public:gunit",
"//absl/functional:bind_front",
"//absl/strings",
],
)
@@ -113,6 +118,7 @@ cc_test(
"endpoint_manager_test.cc",
"internal_payload_factory_test.cc",
"offline_frames_test.cc",
"offline_service_controller_test.cc",
"p2p_cluster_pcp_handler_test.cc",
"payload_manager_test.cc",
"pcp_manager_test.cc",
cpp/core_v2/internal/base_pcp_handler.cc
+46 -33
View File
@@ -32,12 +32,9 @@ BasePcpHandler::BasePcpHandler(EndpointManager* endpoint_manager,
pcp_(pcp) {}
BasePcpHandler::~BasePcpHandler() {
// Unregister ourselves from the FrameProcessors.
NEARBY_LOGS(INFO) << "BasePcpHandler: going down; strategy="
<< strategy_.GetName();
endpoint_manager_->UnregisterFrameProcessor(V1Frame::CONNECTION_RESPONSE,
handle_);
<< strategy_.GetName() << "; handle=" << handle_;
DisconnectFromEndpointManager();
// Stop all the ongoing Runnables (as gracefully as possible).
NEARBY_LOGS(INFO) << "BasePcpHandler: bringing down executors; strategy="
<< strategy_.GetName();
@@ -47,8 +44,17 @@ BasePcpHandler::~BasePcpHandler() {
<< strategy_.GetName();
}
void BasePcpHandler::DisconnectFromEndpointManager() {
if (stop_.Set(true)) return;
NEARBY_LOGS(INFO) << "BasePcpHandler: Unregister from EPM; strategy="
<< strategy_.GetName() << "; handle=" << handle_;
// Unregister ourselves from EPM message dispatcher.
endpoint_manager_->UnregisterFrameProcessor(V1Frame::CONNECTION_RESPONSE,
handle_, true);
}
Status BasePcpHandler::StartAdvertising(ClientProxy* client,
const string& service_id,
const std::string& service_id,
const ConnectionOptions& options,
const ConnectionRequestInfo& info) {
Future<Status> response;
@@ -85,7 +91,7 @@ void BasePcpHandler::StopAdvertising(ClientProxy* client) {
}
Status BasePcpHandler::StartDiscovery(ClientProxy* client,
const string& service_id,
const std::string& service_id,
const ConnectionOptions& options,
const DiscoveryListener& listener) {
Future<Status> response;
@@ -122,7 +128,7 @@ void BasePcpHandler::StopDiscovery(ClientProxy* client) {
WaitForLatch("stopDiscovery", &latch);
}
void BasePcpHandler::WaitForLatch(const string& method_name,
void BasePcpHandler::WaitForLatch(const std::string& method_name,
CountDownLatch* latch) {
Exception await_exception = latch->Await();
if (!await_exception.Ok()) {
@@ -132,7 +138,7 @@ void BasePcpHandler::WaitForLatch(const string& method_name,
}
}
Status BasePcpHandler::WaitForResult(const string& method_name,
Status BasePcpHandler::WaitForResult(const std::string& method_name,
std::int64_t client_id,
Future<Status>* future) {
if (!future) {
@@ -156,9 +162,10 @@ void BasePcpHandler::RunOnPcpHandlerThread(Runnable runnable) {
EncryptionRunner::ResultListener BasePcpHandler::GetResultListener() {
return {
.on_success_cb =
[this](const string& endpoint_id,
[this](const std::string& endpoint_id,
std::unique_ptr<UKey2Handshake> ukey2,
const string& auth_token, const ByteArray& raw_auth_token) {
const std::string& auth_token,
const ByteArray& raw_auth_token) {
RunOnPcpHandlerThread([this, endpoint_id,
raw_ukey2 = ukey2.release(), auth_token,
raw_auth_token]() mutable {
@@ -168,7 +175,7 @@ EncryptionRunner::ResultListener BasePcpHandler::GetResultListener() {
});
},
.on_failure_cb =
[this](const string& endpoint_id, EndpointChannel* channel) {
[this](const std::string& endpoint_id, EndpointChannel* channel) {
RunOnPcpHandlerThread([this, endpoint_id, channel]() {
OnEncryptionFailureRunnable(endpoint_id, channel);
});
@@ -177,8 +184,8 @@ EncryptionRunner::ResultListener BasePcpHandler::GetResultListener() {
}
void BasePcpHandler::OnEncryptionSuccessRunnable(
const string& endpoint_id, std::unique_ptr<UKey2Handshake> ukey2,
const string& auth_token, const ByteArray& raw_auth_token) {
const std::string& endpoint_id, std::unique_ptr<UKey2Handshake> ukey2,
const std::string& auth_token, const ByteArray& raw_auth_token) {
// Quick fail if we've been removed from pending connections while we were
// busy running UKEY2.
auto it = pending_connections_.find(endpoint_id);
@@ -203,7 +210,8 @@ void BasePcpHandler::OnEncryptionSuccessRunnable(
// Set ourselves up so that we receive all acceptance/rejection messages
handle_ = endpoint_manager_->RegisterFrameProcessor(
V1Frame::CONNECTION_RESPONSE, this);
V1Frame::CONNECTION_RESPONSE,
static_cast<EndpointManager::FrameProcessor*>(this));
// Now we register our endpoint so that we can listen for both sides to
// accept.
@@ -225,7 +233,7 @@ void BasePcpHandler::OnEncryptionSuccessRunnable(
}
void BasePcpHandler::OnEncryptionFailureRunnable(
const string& endpoint_id, EndpointChannel* endpoint_channel) {
const std::string& endpoint_id, EndpointChannel* endpoint_channel) {
auto it = pending_connections_.find(endpoint_id);
if (it == pending_connections_.end()) {
NEARBY_LOG(INFO,
@@ -256,7 +264,7 @@ void BasePcpHandler::OnEncryptionFailureRunnable(
}
Status BasePcpHandler::RequestConnection(ClientProxy* client,
const string& endpoint_id,
const std::string& endpoint_id,
const ConnectionRequestInfo& info) {
Future<Status> result;
RunOnPcpHandlerThread([this, client, &info, endpoint_id, &result]() {
@@ -358,7 +366,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
}
BasePcpHandler::DiscoveredEndpoint* BasePcpHandler::GetDiscoveredEndpoint(
const string& endpoint_id) {
const std::string& endpoint_id) {
auto it = discovered_endpoints_.find(endpoint_id);
if (it == discovered_endpoints_.end()) {
return nullptr;
@@ -400,15 +408,15 @@ bool BasePcpHandler::CanReceiveIncomingConnection(ClientProxy* client) const {
}
Exception BasePcpHandler::WriteConnectionRequestFrame(
EndpointChannel* endpoint_channel, const string& local_endpoint_id,
const string& local_endpoint_name, std::int32_t nonce,
EndpointChannel* endpoint_channel, const std::string& local_endpoint_id,
const std::string& local_endpoint_name, std::int32_t nonce,
const std::vector<proto::connections::Medium>& supported_mediums) {
return endpoint_channel->Write(parser::ForConnectionRequest(
local_endpoint_id, local_endpoint_name, nonce, supported_mediums));
}
void BasePcpHandler::ProcessPreConnectionInitiationFailure(
const string& endpoint_id, EndpointChannel* channel, Status status,
const std::string& endpoint_id, EndpointChannel* channel, Status status,
Future<Status>* result) {
if (channel != nullptr) {
channel->Close();
@@ -423,7 +431,7 @@ void BasePcpHandler::ProcessPreConnectionInitiationFailure(
}
void BasePcpHandler::ProcessPreConnectionResultFailure(
ClientProxy* client, const string& endpoint_id) {
ClientProxy* client, const std::string& endpoint_id) {
auto item = pending_connections_.extract(endpoint_id);
endpoint_manager_->DiscardEndpoint(client, endpoint_id);
client->OnConnectionRejected(endpoint_id, {Status::kError});
@@ -448,7 +456,7 @@ bool BasePcpHandler::AutoUpgradeBandwidth() const {
}
Status BasePcpHandler::AcceptConnection(
ClientProxy* client, const string& endpoint_id,
ClientProxy* client, const std::string& endpoint_id,
const PayloadListener& payload_listener) {
Future<Status> response;
RunOnPcpHandlerThread(
@@ -503,7 +511,7 @@ Status BasePcpHandler::AcceptConnection(
}
Status BasePcpHandler::RejectConnection(ClientProxy* client,
const string& endpoint_id) {
const std::string& endpoint_id) {
Future<Status> response;
RunOnPcpHandlerThread([this, client, endpoint_id, &response]() {
NEARBY_LOG(INFO, "RejectConnection: id=%s", endpoint_id.c_str());
@@ -559,7 +567,7 @@ Status BasePcpHandler::RejectConnection(ClientProxy* client,
//}
void BasePcpHandler::OnIncomingFrame(OfflineFrame& frame,
const string& endpoint_id,
const std::string& endpoint_id,
ClientProxy* client,
proto::connections::Medium medium) {
CountDownLatch latch(1);
@@ -595,8 +603,12 @@ void BasePcpHandler::OnIncomingFrame(OfflineFrame& frame,
}
void BasePcpHandler::OnEndpointDisconnect(ClientProxy* client,
const string& endpoint_id,
const std::string& endpoint_id,
CountDownLatch* barrier) {
if (stop_.Get()) {
if (barrier) barrier->CountDown();
return;
}
RunOnPcpHandlerThread([this, client, endpoint_id, barrier]() {
auto item = pending_alarms_.find(endpoint_id);
if (item != pending_alarms_.end()) {
@@ -702,7 +714,7 @@ bool BasePcpHandler::IsPreferred(
}
Exception BasePcpHandler::OnIncomingConnection(
ClientProxy* client, const string& remote_device_name,
ClientProxy* client, const std::string& remote_device_name,
std::unique_ptr<EndpointChannel> channel,
proto::connections::Medium medium) {
absl::Time start_time = SystemClock::ElapsedRealtime();
@@ -797,7 +809,8 @@ Exception BasePcpHandler::OnIncomingConnection(
return {Exception::kSuccess};
}
bool BasePcpHandler::BreakTie(ClientProxy* client, const string& endpoint_id,
bool BasePcpHandler::BreakTie(ClientProxy* client,
const std::string& endpoint_id,
std::int32_t incoming_nonce,
EndpointChannel* endpoint_channel) {
auto it = pending_connections_.find(endpoint_id);
@@ -835,7 +848,7 @@ bool BasePcpHandler::BreakTie(ClientProxy* client, const string& endpoint_id,
}
void BasePcpHandler::ProcessTieBreakLoss(
ClientProxy* client, const string& endpoint_id,
ClientProxy* client, const std::string& endpoint_id,
BasePcpHandler::PendingConnectionInfo* info) {
ProcessPreConnectionInitiationFailure(endpoint_id, info->channel.get(),
{Status::kEndpointIoError},
@@ -845,7 +858,7 @@ void BasePcpHandler::ProcessTieBreakLoss(
}
void BasePcpHandler::InitiateBandwidthUpgrade(
ClientProxy* client, const string& endpoint_id,
ClientProxy* client, const std::string& endpoint_id,
const std::vector<proto::connections::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
@@ -894,7 +907,7 @@ proto::connections::Medium BasePcpHandler::ChooseBestUpgradeMedium(
}
void BasePcpHandler::EvaluateConnectionResult(ClientProxy* client,
const string& endpoint_id,
const std::string& endpoint_id,
bool can_close_immediately) {
// Short-circuit immediately if we're not in an actionable state yet. We will
// be called again once the other side has made their decision.
@@ -1032,12 +1045,12 @@ BasePcpHandler::PendingConnectionInfo::~PendingConnectionInfo() {
}
void BasePcpHandler::PendingConnectionInfo::LocalEndpointAcceptedConnection(
const string& endpoint_id, const PayloadListener& payload_listener) {
const std::string& endpoint_id, const PayloadListener& payload_listener) {
client->LocalEndpointAcceptedConnection(endpoint_id, payload_listener);
}
void BasePcpHandler::PendingConnectionInfo::LocalEndpointRejectedConnection(
const string& endpoint_id) {
const std::string& endpoint_id) {
client->LocalEndpointRejectedConnection(endpoint_id);
}
cpp/core_v2/internal/base_pcp_handler.h
+4
View File
@@ -17,6 +17,7 @@
#include "core_v2/status.h"
#include "proto/connections/offline_wire_formats.pb.h"
#include "platform_v2/base/prng.h"
#include "platform_v2/public/atomic_boolean.h"
#include "platform_v2/public/atomic_reference.h"
#include "platform_v2/public/cancelable_alarm.h"
#include "platform_v2/public/count_down_latch.h"
@@ -139,6 +140,7 @@ class BasePcpHandler : public PcpHandler,
Pcp GetPcp() const override { return pcp_; }
Strategy GetStrategy() const override { return strategy_; }
void DisconnectFromEndpointManager();
protected:
// The result of a call to startAdvertisingImpl() or startDiscoveryImpl().
@@ -423,6 +425,8 @@ class BasePcpHandler : public PcpHandler,
// stops discovering because it might still be useful downstream of
// discovery (eg: connection speed, etc.)
ConnectionOptions discovery_options_;
AtomicBoolean stop_{false};
Pcp pcp_;
Strategy strategy_{PcpToStrategy(pcp_)};
Prng prng_;
cpp/core_v2/internal/base_pcp_handler_test.cc
+94 -93
View File
@@ -24,6 +24,7 @@ namespace {
using ::location::nearby::proto::connections::Medium;
using ::testing::_;
using ::testing::AtLeast;
using ::testing::Invoke;
using ::testing::MockFunction;
using ::testing::Return;
@@ -125,8 +126,7 @@ class MockContext {
struct MockDiscoveredEndpoint : public MockPcpHandler::DiscoveredEndpoint {
MockDiscoveredEndpoint(DiscoveredEndpoint endpoint, MockContext context)
: DiscoveredEndpoint(std::move(endpoint)),
context(std::move(context)) {}
: DiscoveredEndpoint(std::move(endpoint)), context(std::move(context)) {}
MockContext context;
};
@@ -310,148 +310,144 @@ class BasePcpHandlerTest : public ::testing::Test {
};
TEST_F(BasePcpHandlerTest, ConstructorDestructorWorks) {
auto ecm = std::make_unique<EndpointChannelManager>();
auto em = std::make_unique<EndpointManager>(ecm.get());
auto pcp_handler = std::make_unique<MockPcpHandler>(em.get(), ecm.get());
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&em, &ecm);
SUCCEED();
}
TEST_F(BasePcpHandlerTest, StartAdvertisingChangesState) {
auto client = std::make_unique<ClientProxy>();
auto ecm = std::make_unique<EndpointChannelManager>();
auto em = std::make_unique<EndpointManager>(ecm.get());
auto pcp_handler = std::make_unique<MockPcpHandler>(em.get(), ecm.get());
StartAdvertising(client.get(), pcp_handler.get());
ClientProxy client;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&em, &ecm);
StartAdvertising(&client, &pcp_handler);
}
TEST_F(BasePcpHandlerTest, StopAdvertisingChangesState) {
auto client = std::make_unique<ClientProxy>();
auto ecm = std::make_unique<EndpointChannelManager>();
auto em = std::make_unique<EndpointManager>(ecm.get());
auto pcp_handler = std::make_unique<MockPcpHandler>(em.get(), ecm.get());
StartAdvertising(client.get(), pcp_handler.get());
EXPECT_CALL(*pcp_handler, StopAdvertisingImpl(client.get())).Times(1);
EXPECT_TRUE(client->IsAdvertising());
pcp_handler->StopAdvertising(client.get());
EXPECT_FALSE(client->IsAdvertising());
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_F(BasePcpHandlerTest, StartDiscoveryChangesState) {
auto client = std::make_unique<ClientProxy>();
auto ecm = std::make_unique<EndpointChannelManager>();
auto em = std::make_unique<EndpointManager>(ecm.get());
auto pcp_handler = std::make_unique<MockPcpHandler>(em.get(), ecm.get());
StartDiscovery(client.get(), pcp_handler.get());
ClientProxy client;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
MockPcpHandler pcp_handler(&em, &ecm);
StartDiscovery(&client, &pcp_handler);
}
TEST_F(BasePcpHandlerTest, StopDiscoveryChangesState) {
auto client = std::make_unique<ClientProxy>();
auto ecm = std::make_unique<EndpointChannelManager>();
auto em = std::make_unique<EndpointManager>(ecm.get());
auto pcp_handler = std::make_unique<MockPcpHandler>(em.get(), ecm.get());
StartDiscovery(client.get(), pcp_handler.get());
EXPECT_CALL(*pcp_handler, StopDiscoveryImpl(client.get())).Times(1);
EXPECT_TRUE(client->IsDiscovering());
pcp_handler->StopDiscovery(client.get());
EXPECT_FALSE(client->IsDiscovering());
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_F(BasePcpHandlerTest, RequestConnectionChangesState) {
std::string endpoint_id{"1234"};
auto client = std::make_unique<ClientProxy>();
auto ecm = std::make_unique<EndpointChannelManager>();
auto em = std::make_unique<EndpointManager>(ecm.get());
auto pcp_handler = std::make_unique<MockPcpHandler>(em.get(), ecm.get());
StartDiscovery(client.get(), pcp_handler.get());
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;
RequestConnection(endpoint_id, std::move(channel_pair.first), channel_b.get(),
client.get(), pcp_handler.get());
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_F(BasePcpHandlerTest, AcceptConnectionChangesState) {
std::string endpoint_id{"1234"};
auto client = std::make_unique<ClientProxy>();
auto ecm = std::make_unique<EndpointChannelManager>();
auto em = std::make_unique<EndpointManager>(ecm.get());
auto pcp_handler = std::make_unique<MockPcpHandler>(em.get(), ecm.get());
StartDiscovery(client.get(), pcp_handler.get());
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;
RequestConnection(endpoint_id, std::move(channel_pair.first), channel_b.get(),
client.get(), pcp_handler.get());
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.get(), endpoint_id, {}),
EXPECT_EQ(pcp_handler.AcceptConnection(&client, endpoint_id, {}),
Status{Status::kSuccess});
NEARBY_LOG(INFO, "Closing connection: id=%s", endpoint_id.c_str());
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_F(BasePcpHandlerTest, RejectConnectionChangesState) {
std::string endpoint_id{"1234"};
auto client = std::make_unique<ClientProxy>();
auto ecm = std::make_unique<EndpointChannelManager>();
auto em = std::make_unique<EndpointManager>(ecm.get());
auto pcp_handler = std::make_unique<MockPcpHandler>(em.get(), ecm.get());
StartDiscovery(client.get(), pcp_handler.get());
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.get(), pcp_handler.get());
NEARBY_LOG(INFO, "Attempting to reject connection: id=%s",
endpoint_id.c_str());
EXPECT_EQ(pcp_handler->RejectConnection(client.get(), endpoint_id),
&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_LOG(INFO, "Closing connection: id=%s", endpoint_id.c_str());
NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id;
channel_b->Close();
pcp_handler.DisconnectFromEndpointManager();
}
TEST_F(BasePcpHandlerTest, OnIncomingFrameChangesState) {
std::string endpoint_id{"1234"};
auto client = std::make_unique<ClientProxy>();
auto ecm = std::make_unique<EndpointChannelManager>();
auto em = std::make_unique<EndpointManager>(ecm.get());
auto pcp_handler = std::make_unique<MockPcpHandler>(em.get(), ecm.get());
StartDiscovery(client.get(), pcp_handler.get());
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;
RequestConnection(endpoint_id, std::move(channel_pair.first), channel_b.get(),
client.get(), pcp_handler.get());
NEARBY_LOG(INFO, "Attempting to accept connection: id=%s",
endpoint_id.c_str());
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_EQ(pcp_handler->AcceptConnection(client.get(), endpoint_id, {}),
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.get(),
Medium::BLE);
NEARBY_LOG(INFO, "Closing connection: id=%s", endpoint_id.c_str());
pcp_handler.OnIncomingFrame(frame.result(), endpoint_id, &client,
Medium::BLE);
NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id;
channel_b->Close();
}
TEST_F(BasePcpHandlerTest, OnEndpointDisconnectChangesState) {
std::string endpoint_id{"1234"};
auto client = std::make_unique<ClientProxy>();
auto ecm = std::make_unique<EndpointChannelManager>();
auto em = std::make_unique<EndpointManager>(ecm.get());
auto pcp_handler = std::make_unique<MockPcpHandler>(em.get(), ecm.get());
StartDiscovery(client.get(), pcp_handler.get());
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.get(), pcp_handler.get());
NEARBY_LOG(INFO, "Simulating disconnect event: id=%s", endpoint_id.c_str());
CountDownLatch latch(1);
pcp_handler->OnEndpointDisconnect(client.get(), endpoint_id, &latch);
channel_b->Close();
EXPECT_TRUE(latch.Await(absl::Milliseconds(5000)).result());
pcp_handler.DisconnectFromEndpointManager();
}
TEST_F(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) {
@@ -464,15 +460,20 @@ TEST_F(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) {
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;
RequestConnection(endpoint_id, std::move(channel_pair.first),
channel_b.get(), &client, &pcp_handler, &destroyed_flag);
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);
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_TRUE(destroyed_flag.load());
}
cpp/core_v2/internal/endpoint_manager.cc
+57 -19
View File
@@ -200,6 +200,7 @@ EndpointManager::EndpointManager(EndpointChannelManager* manager)
: channel_manager_(manager) {}
EndpointManager::~EndpointManager() {
NEARBY_LOG(INFO, "EndpointManager going down");
CountDownLatch latch(1);
RunOnEndpointManagerThread([this, &latch]() {
NEARBY_LOG(INFO, "Bringing down endpoints");
@@ -208,10 +209,8 @@ EndpointManager::~EndpointManager() {
EndpointState& state = item.second;
// This will close the channel; all workers will sense that and
// terminate.
NEARBY_LOG(INFO, "Bringing down endpoint channels: id=%s",
endpoint_id.c_str());
WaitForEndpointDisconnectionProcessing(state.client, endpoint_id);
channel_manager_->UnregisterChannelForEndpoint(endpoint_id);
state.barrier.Await();
}
latch.CountDown();
});
@@ -236,9 +235,12 @@ EndpointManager::RegisterFrameProcessor(
RunOnEndpointManagerThread([this, frame_type, &latch, processor]() {
auto it = frame_processors_.find(frame_type);
if (it != frame_processors_.end()) {
NEARBY_LOG(INFO, "Frame processor found, updated; type=%d", frame_type);
NEARBY_LOGS(INFO) << "Frame processor found: updated; type=" << frame_type
<< "; processor=" << processor << "; self=" << this;
it->second = processor;
} else {
NEARBY_LOGS(INFO) << "Frame processor added; type=" << frame_type
<< "; processor=" << processor << "; self=" << this;
frame_processors_.emplace(frame_type, processor);
}
latch.CountDown();
@@ -249,14 +251,22 @@ EndpointManager::RegisterFrameProcessor(
void EndpointManager::UnregisterFrameProcessor(V1Frame::FrameType frame_type,
const void* handle, bool sync) {
NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [enter]: handle=" << handle;
if (handle == nullptr) return;
CountDownLatch latch(1);
RunOnEndpointManagerThread([this, frame_type, handle, &latch, sync]() {
auto it = frame_processors_.find(frame_type);
if (it == frame_processors_.end()) return;
if (it == frame_processors_.end()) {
NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [not found]: handle="
<< handle;
if (sync) latch.CountDown();
return;
}
NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [found]: handle=" << handle;
if (it->second == handle) {
frame_processors_.erase(it);
NEARBY_LOG(INFO, "Unregistered: type=%d", frame_type);
NEARBY_LOGS(INFO) << "Unregistered: type=" << frame_type
<< "; processor=" << handle << "; self=" << this;
} else {
NEARBY_LOG(INFO,
"Failed to unregister: type=%d; handle mismatch: passed=%p, "
@@ -267,7 +277,8 @@ void EndpointManager::UnregisterFrameProcessor(V1Frame::FrameType frame_type,
});
if (sync) {
latch.Await();
NEARBY_LOG(INFO, "Unregistered: [sync done] type=%d", frame_type);
NEARBY_LOGS(INFO) << "Unregistered [sync done]: type=" << frame_type
<< "; processor=" << handle << "; self=" << this;
}
}
@@ -294,10 +305,13 @@ void EndpointManager::EnsureWorkersTerminated(const std::string& endpoint_id) {
// If another instance of data and keep-alive handlers is running, it will
// terminate soon; we should block until it happens.
EndpointState& endpoint_state = item->second;
NEARBY_LOG(INFO, "Waiting for workers to terminate for endpoint_id='%s'",
endpoint_id.c_str());
NEARBY_LOGS(INFO) << "Waiting for workers to terminate for id: "
<< endpoint_id;
endpoint_state.barrier.Await();
endpoints_.erase(item);
NEARBY_LOGS(INFO) << "Workers terminated for id: " << endpoint_id;
} else {
NEARBY_LOGS(INFO) << "EndpointState not found for id: " << endpoint_id;
}
}
@@ -378,8 +392,8 @@ void EndpointManager::UnregisterEndpoint(ClientProxy* client,
const std::string& endpoint_id) {
CountDownLatch latch(1);
RunOnEndpointManagerThread([this, client, endpoint_id, &latch]() {
channel_manager_->UnregisterChannelForEndpoint(endpoint_id);
RemoveEndpoint(client, endpoint_id, /*notify=*/false);
RemoveEndpoint(client, endpoint_id,
client->IsConnectedToEndpoint(endpoint_id));
latch.CountDown();
});
latch.Await();
@@ -391,7 +405,6 @@ void EndpointManager::UnregisterEndpoint(ClientProxy* client,
void EndpointManager::DiscardEndpoint(ClientProxy* client,
const std::string& endpoint_id) {
RunOnEndpointManagerThread([this, client, endpoint_id]() {
channel_manager_->UnregisterChannelForEndpoint(endpoint_id);
RemoveEndpoint(client, endpoint_id,
/*notify=*/
client->IsConnectedToEndpoint(endpoint_id));
@@ -435,25 +448,50 @@ void EndpointManager::RemoveEndpoint(ClientProxy* client,
// should be no further interactions with the endpoint.
// (See b/37352254 for history)
WaitForEndpointDisconnectionProcessing(client, endpoint_id);
EnsureWorkersTerminated(endpoint_id);
client->OnDisconnected(endpoint_id, notify);
NEARBY_LOG(INFO, "Removed endpoint; id=%s",
endpoint_id.c_str());
NEARBY_LOG(INFO, "Removed endpoint; id=%s", endpoint_id.c_str());
}
}
// @EndpointManagerThread
void EndpointManager::WaitForEndpointDisconnectionProcessing(
ClientProxy* client, const std::string& endpoint_id) {
CountDownLatch barrier(frame_processors_.size());
NEARBY_LOGS(INFO) << "Wait: client=" << client << "; id=" << endpoint_id;
auto total_size = frame_processors_.size();
NEARBY_LOGS(INFO) << "Total frame processors: " << total_size;
if (!total_size) return;
CountDownLatch barrier(total_size);
int valid = 0;
for (auto& item : frame_processors_) {
auto& processor = item.second;
processor->OnEndpointDisconnect(client, endpoint_id, &barrier);
auto* processor = item.second;
NEARBY_LOGS(INFO) << "processor=" << processor << "; type=" << item.first;
if (processor) {
valid++;
processor->OnEndpointDisconnect(client, endpoint_id, &barrier);
} else {
barrier.CountDown();
}
}
barrier.Await(kProcessEndpointDisconnectionTimeout);
if (!valid) {
NEARBY_LOGS(INFO) << "No valid frame processors.";
return;
} else {
NEARBY_LOGS(INFO) << "Valid frame processors: " << valid;
}
NEARBY_LOGS(INFO) << "Waiting for " << valid
<< " frame processors to disconnect from: " << endpoint_id;
if (!barrier.Await(kProcessEndpointDisconnectionTimeout).result()) {
NEARBY_LOGS(INFO) << "Failed to disconnect frame processors from: "
<< endpoint_id;
} else {
NEARBY_LOGS(INFO)
<< "Finished waiting for frame processors to disconnect from: "
<< endpoint_id;
}
}
std::vector<std::string> EndpointManager::SendTransferFrameBytes(
cpp/core_v2/internal/mediums/webrtc.h
+1 -1
View File
@@ -138,11 +138,11 @@ class WebRtc {
std::vector<::location::nearby::mediums::IceCandidate>
pending_local_ice_candidates_ ABSL_GUARDED_BY(mutex_);
WebRtcMedium medium_;
std::unique_ptr<ConnectionFlow> connection_flow_;
std::unique_ptr<WebRtcSignalingMessenger> signaling_messenger_
ABSL_GUARDED_BY(mutex_);
WebRtcSocketWrapper socket_ ABSL_GUARDED_BY(mutex_);
WebRtcMedium medium_;
SingleThreadExecutor single_thread_executor_;
};
cpp/core_v2/internal/offline_service_controller.cc
+93
View File
@@ -0,0 +1,93 @@
// Copyright 2020 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#include "core_v2/internal/offline_service_controller.h"
#include <string>
namespace location {
namespace nearby {
namespace connections {
OfflineServiceController::~OfflineServiceController() {
Stop();
}
void OfflineServiceController::Stop() {
if (stop_.Set(true)) return;
payload_manager_.DisconnectFromEndpointManager();
pcp_manager_.DisconnectFromEndpointManager();
}
Status OfflineServiceController::StartAdvertising(
ClientProxy* client, const std::string& service_id,
const ConnectionOptions& options, const ConnectionRequestInfo& info) {
return pcp_manager_.StartAdvertising(client, service_id, options, info);
}
void OfflineServiceController::StopAdvertising(ClientProxy* client) {
pcp_manager_.StopAdvertising(client);
}
Status OfflineServiceController::StartDiscovery(
ClientProxy* client, const std::string& service_id,
const ConnectionOptions& options, const DiscoveryListener& listener) {
return pcp_manager_.StartDiscovery(client, service_id, options, listener);
}
void OfflineServiceController::StopDiscovery(ClientProxy* client) {
pcp_manager_.StopDiscovery(client);
}
Status OfflineServiceController::RequestConnection(
ClientProxy* client, const std::string& endpoint_id,
const ConnectionRequestInfo& info) {
return pcp_manager_.RequestConnection(client, endpoint_id, info);
}
Status OfflineServiceController::AcceptConnection(
ClientProxy* client, const std::string& endpoint_id,
const PayloadListener& listener) {
return pcp_manager_.AcceptConnection(client, endpoint_id, listener);
}
Status OfflineServiceController::RejectConnection(
ClientProxy* client, const std::string& endpoint_id) {
return pcp_manager_.RejectConnection(client, endpoint_id);
}
void OfflineServiceController::InitiateBandwidthUpgrade(
ClientProxy* client, const std::string& endpoint_id) {
// TODO(apolyudov): implement.
}
void OfflineServiceController::SendPayload(
ClientProxy* client, const std::vector<std::string>& endpoint_ids,
Payload payload) {
payload_manager_.SendPayload(client, endpoint_ids, std::move(payload));
}
Status OfflineServiceController::CancelPayload(ClientProxy* client,
std::int64_t payload_id) {
return payload_manager_.CancelPayload(client, payload_id);
}
void OfflineServiceController::DisconnectFromEndpoint(
ClientProxy* client, const std::string& endpoint_id) {
endpoint_manager_.UnregisterEndpoint(client, endpoint_id);
}
} // namespace connections
} // namespace nearby
} // namespace location
cpp/core_v2/internal/offline_service_controller.h
+94
View File
@@ -0,0 +1,94 @@
// Copyright 2020 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#ifndef CORE_V2_INTERNAL_OFFLINE_SERVICE_CONTROLLER_H_
#define CORE_V2_INTERNAL_OFFLINE_SERVICE_CONTROLLER_H_
#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) 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/offline_service_controller_test.cc
+335
View File
@@ -0,0 +1,335 @@
// Copyright 2020 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#include "core_v2/internal/offline_service_controller.h"
#include "core_v2/internal/offline_simulation_user.h"
#include "platform_v2/base/medium_environment.h"
#include "platform_v2/base/output_stream.h"
#include "platform_v2/public/count_down_latch.h"
#include "platform_v2/public/logging.h"
#include "platform_v2/public/pipe.h"
#include "platform_v2/public/system_clock.h"
#include "gmock/gmock.h"
#include "gtest/gtest.h"
namespace location {
namespace nearby {
namespace connections {
namespace {
using ::testing::Eq;
constexpr absl::string_view kServiceId = "service-id";
constexpr absl::string_view kDeviceA = "device-a";
constexpr absl::string_view kDeviceB = "device-b";
constexpr absl::string_view kMessage = "message";
constexpr absl::Duration kProgressTimeout = absl::Milliseconds(1000);
constexpr absl::Duration kDefaultTimeout = absl::Milliseconds(1000);
constexpr absl::Duration kDisconnectTimeout = absl::Milliseconds(15000);
class OfflineServiceControllerTest : public ::testing::Test {
protected:
OfflineServiceControllerTest() { env_.Stop(); }
bool SetupConnection(OfflineSimulationUser& user_a,
OfflineSimulationUser& user_b) {
user_a.StartAdvertising(std::string(kServiceId), &connect_latch_);
user_b.StartDiscovery(std::string(kServiceId), &discover_latch_);
EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result());
EXPECT_EQ(user_b.GetDiscovered().service_id, kServiceId);
EXPECT_EQ(user_b.GetDiscovered().endpoint_name, user_a.GetName());
EXPECT_FALSE(user_b.GetDiscovered().endpoint_id.empty());
NEARBY_LOG(INFO, "EP-B: [discovered] %s",
user_b.GetDiscovered().endpoint_id.c_str());
user_b.RequestConnection(&connect_latch_);
EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result());
EXPECT_FALSE(user_a.GetDiscovered().endpoint_id.empty());
NEARBY_LOG(INFO, "EP-A: [discovered] %s",
user_a.GetDiscovered().endpoint_id.c_str());
NEARBY_LOG(INFO, "Both users discovered their peers.");
user_a.AcceptConnection(&accept_latch_);
user_b.AcceptConnection(&accept_latch_);
EXPECT_TRUE(accept_latch_.Await(kDefaultTimeout).result());
NEARBY_LOG(INFO, "Both users reached connected state.");
return user_a.IsConnected() && user_b.IsConnected();
}
CountDownLatch discover_latch_{1};
CountDownLatch connect_latch_{2};
CountDownLatch accept_latch_{2};
CountDownLatch payload_latch_{1};
MediumEnvironment& env_ = MediumEnvironment::Instance();
};
TEST_F(OfflineServiceControllerTest, CanCreateOne) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA);
env_.Stop();
}
TEST_F(OfflineServiceControllerTest, CanCreateMany) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA);
OfflineSimulationUser user_b(kDeviceB);
env_.Stop();
}
TEST_F(OfflineServiceControllerTest, CanStartAdvertising) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA);
OfflineSimulationUser user_b(kDeviceB);
EXPECT_FALSE(user_a.IsAdvertising());
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(user_a.IsAdvertising());
env_.Stop();
}
TEST_F(OfflineServiceControllerTest, CanStartDiscoveryBeforeAdvertising) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA);
OfflineSimulationUser user_b(kDeviceB);
EXPECT_FALSE(user_b.IsDiscovering());
EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(user_b.IsDiscovering());
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_F(OfflineServiceControllerTest, CanStartDiscoveryAfterAdvertising) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA);
OfflineSimulationUser user_b(kDeviceB);
EXPECT_FALSE(user_b.IsDiscovering());
EXPECT_FALSE(user_b.IsAdvertising());
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr),
Eq(Status{Status::kSuccess}));
EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(user_a.IsAdvertising());
EXPECT_TRUE(user_b.IsDiscovering());
EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_F(OfflineServiceControllerTest, CanStopAdvertising) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA);
OfflineSimulationUser user_b(kDeviceB);
EXPECT_FALSE(user_a.IsAdvertising());
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(user_a.IsAdvertising());
user_a.StopAdvertising();
EXPECT_FALSE(user_a.IsAdvertising());
EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(user_b.IsDiscovering());
EXPECT_FALSE(discover_latch_.Await(kDefaultTimeout).result());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_F(OfflineServiceControllerTest, CanStopDiscovery) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA);
OfflineSimulationUser user_b(kDeviceB);
EXPECT_FALSE(user_b.IsDiscovering());
EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(user_b.IsDiscovering());
user_b.StopDiscovery();
EXPECT_FALSE(user_b.IsDiscovering());
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr),
Eq(Status{Status::kSuccess}));
EXPECT_FALSE(discover_latch_.Await(kDefaultTimeout).result());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_F(OfflineServiceControllerTest, CanConnect) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA);
OfflineSimulationUser user_b(kDeviceB);
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), &connect_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result());
EXPECT_THAT(user_b.RequestConnection(&connect_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_F(OfflineServiceControllerTest, CanAcceptConnection) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA);
OfflineSimulationUser user_b(kDeviceB);
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), &connect_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result());
EXPECT_THAT(user_b.RequestConnection(&connect_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result());
EXPECT_THAT(user_a.AcceptConnection(&accept_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_THAT(user_b.AcceptConnection(&accept_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(accept_latch_.Await(kDefaultTimeout).result());
EXPECT_TRUE(user_a.IsConnected());
EXPECT_TRUE(user_b.IsConnected());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_F(OfflineServiceControllerTest, CanRejectConnection) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA);
OfflineSimulationUser user_b(kDeviceB);
CountDownLatch reject_latch(1);
EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), &connect_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result());
EXPECT_THAT(user_b.RequestConnection(&connect_latch_),
Eq(Status{Status::kSuccess}));
EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result());
user_a.ExpectRejectedConnection(reject_latch);
EXPECT_THAT(user_b.RejectConnection(nullptr), Eq(Status{Status::kSuccess}));
EXPECT_TRUE(reject_latch.Await(kDefaultTimeout).result());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_F(OfflineServiceControllerTest, CanSendBytePayload) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA);
OfflineSimulationUser user_b(kDeviceB);
ASSERT_TRUE(SetupConnection(user_a, user_b));
ByteArray message(std::string{kMessage});
user_a.SendPayload(Payload(message));
user_b.ExpectPayload(payload_latch_);
EXPECT_TRUE(payload_latch_.Await(kDefaultTimeout).result());
EXPECT_EQ(user_b.GetPayload().AsBytes(), message);
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_F(OfflineServiceControllerTest, CanSendStreamPayload) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA);
OfflineSimulationUser user_b(kDeviceB);
ASSERT_TRUE(SetupConnection(user_a, user_b));
ByteArray message(std::string{kMessage});
auto pipe = std::make_shared<Pipe>();
OutputStream& tx = pipe->GetOutputStream();
user_a.SendPayload(Payload([pipe]() -> InputStream& {
return pipe->GetInputStream(); // NOLINT
}));
user_b.ExpectPayload(payload_latch_);
tx.Write(message);
EXPECT_TRUE(payload_latch_.Await(kDefaultTimeout).result());
EXPECT_NE(user_b.GetPayload().AsStream(), nullptr);
InputStream& rx = *user_b.GetPayload().AsStream();
ASSERT_TRUE(user_b.WaitForProgress(
[size = message.size()](const PayloadProgressInfo& info) -> bool {
return info.bytes_transferred >= size;
},
kProgressTimeout));
EXPECT_EQ(rx.Read(Pipe::kChunkSize).result(), message);
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_F(OfflineServiceControllerTest, CanCancelStreamPayload) {
env_.Start();
OfflineSimulationUser user_a(kDeviceA);
OfflineSimulationUser user_b(kDeviceB);
ASSERT_TRUE(SetupConnection(user_a, user_b));
ByteArray message(std::string{kMessage});
auto pipe = std::make_shared<Pipe>();
OutputStream& tx = pipe->GetOutputStream();
user_a.SendPayload(Payload([pipe]() -> InputStream& {
return pipe->GetInputStream(); // NOLINT
}));
user_b.ExpectPayload(payload_latch_);
tx.Write(message);
EXPECT_TRUE(payload_latch_.Await(kDefaultTimeout).result());
EXPECT_NE(user_b.GetPayload().AsStream(), nullptr);
InputStream& rx = *user_b.GetPayload().AsStream();
ASSERT_TRUE(user_b.WaitForProgress(
[size = message.size()](const PayloadProgressInfo& info) -> bool {
return info.bytes_transferred >= size;
},
kProgressTimeout));
EXPECT_EQ(rx.Read(Pipe::kChunkSize).result(), message);
user_b.CancelPayload();
int count = 0;
while (true) {
count++;
if (!tx.Write(message).Ok()) break;
SystemClock::Sleep(kDefaultTimeout);
}
EXPECT_TRUE(user_a.WaitForProgress(
[](const PayloadProgressInfo& info) -> bool {
return info.status == PayloadProgressInfo::Status::kCanceled;
},
kProgressTimeout));
EXPECT_LT(count, 10);
user_a.Stop();
user_b.Stop();
env_.Stop();
}
TEST_F(OfflineServiceControllerTest, CanDisconnect) {
env_.Start();
CountDownLatch disconnect_latch(1);
OfflineSimulationUser user_a(kDeviceA);
OfflineSimulationUser user_b(kDeviceB);
ASSERT_TRUE(SetupConnection(user_a, user_b));
NEARBY_LOGS(INFO) << "Disconnecting";
user_b.ExpectDisconnect(disconnect_latch);
user_b.Disconnect();
EXPECT_TRUE(disconnect_latch.Await(kDisconnectTimeout).result());
NEARBY_LOGS(INFO) << "Disconnected";
EXPECT_FALSE(user_b.IsConnected());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
} // namespace
} // namespace connections
} // namespace nearby
} // namespace location
cpp/core_v2/internal/offline_simulation_user.cc
+191
View File
@@ -0,0 +1,191 @@
// Copyright 2020 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#include "core_v2/internal/offline_simulation_user.h"
#include "core_v2/listeners.h"
#include "platform_v2/public/count_down_latch.h"
#include "platform_v2/public/system_clock.h"
#include "absl/functional/bind_front.h"
namespace location {
namespace nearby {
namespace connections {
void OfflineSimulationUser::OnConnectionInitiated(
const std::string& endpoint_id, const ConnectionResponseInfo& info,
bool is_outgoing) {
if (is_outgoing) {
NEARBY_LOG(INFO, "RequestConnection: initiated_cb called");
} else {
NEARBY_LOG(INFO, "StartAdvertising: initiated_cb called");
discovered_ = DiscoveredInfo{
.endpoint_id = endpoint_id,
.endpoint_name = name_,
.service_id = service_id_,
};
}
if (initiated_latch_) initiated_latch_->CountDown();
}
void OfflineSimulationUser::OnConnectionAccepted(
const std::string& endpoint_id) {
if (accept_latch_) accept_latch_->CountDown();
}
void OfflineSimulationUser::OnConnectionRejected(const std::string& endpoint_id,
Status status) {
if (reject_latch_) reject_latch_->CountDown();
}
void OfflineSimulationUser::OnEndpointDisconnect(
const std::string& endpoint_id) {
NEARBY_LOGS(INFO) << "OnEndpointDisconnect: self=" << this
<< "; id=" << endpoint_id;
if (disconnect_latch_) disconnect_latch_->CountDown();
}
void OfflineSimulationUser::OnEndpointFound(const std::string& endpoint_id,
const std::string& endpoint_name,
const std::string& service_id) {
NEARBY_LOG(INFO, "Device discovered: id=%s", endpoint_id.c_str());
discovered_ = DiscoveredInfo{
.endpoint_id = endpoint_id,
.endpoint_name = endpoint_name,
.service_id = service_id,
};
if (found_latch_) found_latch_->CountDown();
}
void OfflineSimulationUser::OnEndpointLost(const std::string& endpoint_id) {
if (lost_latch_) lost_latch_->CountDown();
}
void OfflineSimulationUser::OnPayload(const std::string& endpoint_id,
Payload payload) {
payload_ = std::move(payload);
if (payload_latch_) payload_latch_->CountDown();
}
void OfflineSimulationUser::OnPayloadProgress(const std::string& endpoint_id,
const PayloadProgressInfo& info) {
MutexLock lock(&progress_mutex_);
progress_info_ = info;
if (future_ && predicate_ && predicate_(info)) future_->Set(true);
}
bool OfflineSimulationUser::WaitForProgress(
std::function<bool(const PayloadProgressInfo&)> predicate,
absl::Duration timeout) {
Future<bool> future;
{
MutexLock lock(&progress_mutex_);
if (predicate(progress_info_)) return true;
future_ = &future;
predicate_ = std::move(predicate);
}
auto response = future.Get(timeout);
{
MutexLock lock(&progress_mutex_);
future_ = nullptr;
predicate_ = nullptr;
}
return response.ok() && response.result();
}
Status OfflineSimulationUser::StartAdvertising(const std::string& service_id,
CountDownLatch* latch) {
initiated_latch_ = latch;
service_id_ = service_id;
ConnectionListener listener = {
.initiated_cb =
std::bind(&OfflineSimulationUser::OnConnectionInitiated, this,
std::placeholders::_1, std::placeholders::_2, false),
.accepted_cb =
absl::bind_front(&OfflineSimulationUser::OnConnectionAccepted, this),
.rejected_cb =
absl::bind_front(&OfflineSimulationUser::OnConnectionRejected, this),
.disconnected_cb =
absl::bind_front(&OfflineSimulationUser::OnEndpointDisconnect, this),
};
return ctrl_.StartAdvertising(&client_, service_id_, options_,
{
.name = name_,
.listener = std::move(listener),
});
}
void OfflineSimulationUser::StopAdvertising() {
ctrl_.StopAdvertising(&client_);
}
Status OfflineSimulationUser::StartDiscovery(const std::string& service_id,
CountDownLatch* latch) {
found_latch_ = latch;
DiscoveryListener listener = {
.endpoint_found_cb =
absl::bind_front(&OfflineSimulationUser::OnEndpointFound, this),
.endpoint_lost_cb =
absl::bind_front(&OfflineSimulationUser::OnEndpointLost, this),
};
return ctrl_.StartDiscovery(&client_, service_id, options_,
std::move(listener));
}
void OfflineSimulationUser::StopDiscovery() { ctrl_.StopDiscovery(&client_); }
Status OfflineSimulationUser::RequestConnection(CountDownLatch* latch) {
initiated_latch_ = latch;
ConnectionListener listener = {
.initiated_cb =
std::bind(&OfflineSimulationUser::OnConnectionInitiated, this,
std::placeholders::_1, std::placeholders::_2, true),
.accepted_cb =
absl::bind_front(&OfflineSimulationUser::OnConnectionAccepted, this),
.rejected_cb =
absl::bind_front(&OfflineSimulationUser::OnConnectionRejected, this),
.disconnected_cb =
absl::bind_front(&OfflineSimulationUser::OnEndpointDisconnect, this),
};
return ctrl_.RequestConnection(&client_, discovered_.endpoint_id,
{
.name = discovered_.endpoint_name,
.listener = std::move(listener),
});
}
Status OfflineSimulationUser::AcceptConnection(CountDownLatch* latch) {
accept_latch_ = latch;
PayloadListener listener = {
.payload_cb = absl::bind_front(&OfflineSimulationUser::OnPayload, this),
.payload_progress_cb =
absl::bind_front(&OfflineSimulationUser::OnPayloadProgress, this),
};
return ctrl_.AcceptConnection(&client_, discovered_.endpoint_id,
std::move(listener));
}
Status OfflineSimulationUser::RejectConnection(CountDownLatch* latch) {
reject_latch_ = latch;
return ctrl_.RejectConnection(&client_, discovered_.endpoint_id);
}
void OfflineSimulationUser::Disconnect() {
NEARBY_LOGS(INFO) << "Disconnecting from id=" << discovered_.endpoint_id;
ctrl_.DisconnectFromEndpoint(&client_, discovered_.endpoint_id);
}
} // namespace connections
} // namespace nearby
} // namespace location
cpp/core_v2/internal/offline_simulation_user.h
+175
View File
@@ -0,0 +1,175 @@
// Copyright 2020 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#ifndef CORE_V2_INTERNAL_OFFLINE_SIMULATION_USER_H_
#define CORE_V2_INTERNAL_OFFLINE_SIMULATION_USER_H_
#include <string>
#include "core_v2/internal/client_proxy.h"
#include "core_v2/internal/offline_service_controller.h"
#include "platform_v2/public/atomic_boolean.h"
#include "platform_v2/public/condition_variable.h"
#include "platform_v2/public/count_down_latch.h"
#include "platform_v2/public/future.h"
#include "gtest/gtest.h"
#include "absl/strings/string_view.h"
// Test-only class to help run end-to-end simulations for nearby connections
// protocol.
//
// This is a "standalone" version of PcpManager. It can run independently,
// provided MediumEnvironment has adequate support for all medium types in use.
namespace location {
namespace nearby {
namespace connections {
class OfflineSimulationUser {
public:
struct DiscoveredInfo {
std::string endpoint_id;
std::string endpoint_name;
std::string service_id;
bool Empty() const { return endpoint_id.empty(); }
void Clear() { endpoint_id.clear(); }
};
explicit OfflineSimulationUser(absl::string_view device_name)
: name_(device_name) {}
virtual ~OfflineSimulationUser() = default;
// Calls PcpManager::StartAdvertising().
// If latch is provided, will call latch->CountDown() in the initiated_cb
// callback.
Status StartAdvertising(const std::string& service_id, CountDownLatch* latch);
// Calls PcpManager::StopAdvertising().
void StopAdvertising();
// Calls PcpManager::StartDiscovery().
// If latch is provided, will call latch->CountDown() in the endpoint_found_cb
// callback.
Status StartDiscovery(const std::string& service_id, CountDownLatch* latch);
// Calls PcpManager::StopDiscovery().
void StopDiscovery();
// Calls PcpManager::RequestConnection().
// If latch is provided, latch->CountDown() will be called in the initiated_cb
// callback.
Status RequestConnection(CountDownLatch* latch);
// Calls PcpManager::AcceptConnection.
// If latch is provided, latch->CountDown() will be called in the accepted_cb
// callback.
Status AcceptConnection(CountDownLatch* latch);
// Calls PcpManager::RejectConnection.
// If latch is provided, latch->CountDown() will be called in the rejected_cb
// callback.
Status RejectConnection(CountDownLatch* latch);
// Unlike acceptance, rejection does not have to be mutual, in order to work.
// This method will allow to synchronize on the remote rejection, without
// performing a local rejection.
// latch.CountDown() will be called in the rejected_cb callback.
void ExpectRejectedConnection(CountDownLatch& latch) {
reject_latch_ = &latch;
}
void ExpectPayload(CountDownLatch& latch) { payload_latch_ = &latch; }
void ExpectDisconnect(CountDownLatch& latch) { disconnect_latch_ = &latch; }
const DiscoveredInfo& GetDiscovered() const { return discovered_; }
std::string GetName() const { return name_; }
bool WaitForProgress(std::function<bool(const PayloadProgressInfo&)> pred,
absl::Duration timeout);
Payload& GetPayload() { return payload_; }
void SendPayload(Payload payload) {
sender_payload_id_ = payload.GetId();
ctrl_.SendPayload(&client_, {discovered_.endpoint_id}, std::move(payload));
}
Status CancelPayload() {
if (sender_payload_id_) {
return ctrl_.CancelPayload(&client_, sender_payload_id_);
} else {
return ctrl_.CancelPayload(&client_, payload_.GetId());
}
}
void Disconnect();
bool IsAdvertising() const { return client_.IsAdvertising(); }
bool IsDiscovering() const { return client_.IsDiscovering(); }
bool IsConnected() const {
return client_.IsConnectedToEndpoint(discovered_.endpoint_id);
}
void Stop() {
ctrl_.Stop();
}
protected:
// ConnectionListener callbacks
void OnConnectionInitiated(const std::string& endpoint_id,
const ConnectionResponseInfo& info,
bool is_outgoing);
void OnConnectionAccepted(const std::string& endpoint_id);
void OnConnectionRejected(const std::string& endpoint_id, Status status);
void OnEndpointDisconnect(const std::string& endpoint_id);
// DiscoveryListener callbacks
void OnEndpointFound(const std::string& endpoint_id,
const std::string& endpoint_name,
const std::string& service_id);
void OnEndpointLost(const std::string& endpoint_id);
// PayloadListener callbacks
void OnPayload(const std::string& endpoint_id, Payload payload);
void OnPayloadProgress(const std::string& endpoint_id,
const PayloadProgressInfo& info);
std::string service_id_;
DiscoveredInfo discovered_;
Mutex progress_mutex_;
ConditionVariable progress_sync_{&progress_mutex_};
PayloadProgressInfo progress_info_;
Payload payload_;
Payload::Id sender_payload_id_ = 0;
CountDownLatch* initiated_latch_ = nullptr;
CountDownLatch* accept_latch_ = nullptr;
CountDownLatch* reject_latch_ = nullptr;
CountDownLatch* found_latch_ = nullptr;
CountDownLatch* lost_latch_ = nullptr;
CountDownLatch* payload_latch_ = nullptr;
CountDownLatch* disconnect_latch_ = nullptr;
Future<bool>* future_ = nullptr;
std::function<bool(const PayloadProgressInfo&)> predicate_;
std::string name_;
ConnectionOptions options_{.strategy = Strategy::kP2pCluster};
ClientProxy client_;
OfflineServiceController ctrl_;
};
} // namespace connections
} // namespace nearby
} // namespace location
#endif // CORE_V2_INTERNAL_OFFLINE_SIMULATION_USER_H_
cpp/core_v2/internal/payload_manager.cc
+17 -9
View File
@@ -44,7 +44,7 @@ bool PayloadManager::SendPayloadLoop(
for (const auto& endpoint : unavailable_endpoints) {
HandleFinishedOutgoingPayload(
client, {endpoint->id}, payload_header, next_chunk_offset,
EndpointInfoStatusToPayloadStatus(endpoint->status));
EndpointInfoStatusToPayloadStatus(endpoint->status.Get()));
}
// Update the still-active recipients of this payload.
@@ -142,8 +142,8 @@ PayloadManager::GetAvailableAndUnavailableEndpoints(
Endpoints unavailable;
for (auto* endpoint_info : pending_payload.GetEndpoints()) {
NEARBY_LOG(INFO, "EndpointInfo: %p; id=%s; status=%d", endpoint_info,
endpoint_info->id.c_str(), endpoint_info->status);
if (endpoint_info->status ==
endpoint_info->id.c_str(), endpoint_info->status.Get());
if (endpoint_info->status.Get() ==
PayloadManager::EndpointInfo::Status::kAvailable) {
available.push_back(endpoint_info);
} else {
@@ -239,12 +239,16 @@ void PayloadManager::CancelAllPayloads() {
}
}
PayloadManager::~PayloadManager() {
NEARBY_LOG(INFO, "PayloadManager: going down; self=%p", this);
shutdown_.Set(true);
void PayloadManager::DisconnectFromEndpointManager() {
if (shutdown_.Set(true)) return;
// Unregister ourselves from the FrameProcessors.
endpoint_manager_->UnregisterFrameProcessor(V1Frame::PAYLOAD_TRANSFER,
handle_, true);
}
PayloadManager::~PayloadManager() {
NEARBY_LOG(INFO, "PayloadManager: going down; self=%p", this);
DisconnectFromEndpointManager();
CancelAllPayloads();
NEARBY_LOG(INFO, "PayloadManager: turn down payload executors; self=%p",
this);
@@ -385,7 +389,11 @@ void PayloadManager::OnIncomingFrame(
void PayloadManager::OnEndpointDisconnect(ClientProxy* client,
const std::string& endpoint_id,
CountDownLatch* barrier) {
RunOnStatusUpdateThread([this, client, endpoint_id, &barrier]() {
if (shutdown_.Get()) {
if (barrier) barrier->CountDown();
return;
}
RunOnStatusUpdateThread([this, client, endpoint_id, barrier]() {
// Iterate through all our payloads and look for payloads associated
// with this endpoint.
MutexLock lock(&mutex_);
@@ -907,7 +915,7 @@ PayloadManager::EndpointInfo::ControlMessageEventToEndpointInfoStatus(
void PayloadManager::EndpointInfo::SetStatusFromControlMessage(
const PayloadTransferFrame::ControlMessage& control_message) {
status = ControlMessageEventToEndpointInfoStatus(control_message.event());
status.Set(ControlMessageEventToEndpointInfoStatus(control_message.event()));
}
//////////////////////////////// PendingPayload ////////////////////////////////
@@ -924,7 +932,7 @@ PayloadManager::PendingPayload::PendingPayload(
for (const auto& id : endpoint_ids) {
endpoints_.emplace(id, EndpointInfo{
.id = id,
.status = EndpointInfo::Status::kAvailable,
.status {EndpointInfo::Status::kAvailable},
});
}
}
cpp/core_v2/internal/payload_manager.h
+4 -1
View File
@@ -15,6 +15,7 @@
#include "proto/connections/offline_wire_formats.pb.h"
#include "platform_v2/base/byte_array.h"
#include "platform_v2/public/atomic_boolean.h"
#include "platform_v2/public/atomic_reference.h"
#include "platform_v2/public/count_down_latch.h"
#include "platform_v2/public/mutex.h"
#include "proto/connections_enums.pb.h"
@@ -47,6 +48,8 @@ class PayloadManager : public EndpointManager::FrameProcessor {
void OnEndpointDisconnect(ClientProxy* client, const std::string& endpoint_id,
CountDownLatch* barrier) override;
void DisconnectFromEndpointManager();
private:
// Information about an endpoint for a particular payload.
struct EndpointInfo {
@@ -65,7 +68,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
PayloadTransferFrame::ControlMessage::EventType event);
std::string id;
Status status = Status::kUnknown;
AtomicReference<Status> status {Status::kUnknown};
std::int64_t offset = 0;
};
cpp/core_v2/internal/payload_manager_test.cc
+9
View File
@@ -25,6 +25,7 @@ class PayloadSimulationUser : public SimulationUser {
explicit PayloadSimulationUser(absl::string_view name)
: SimulationUser(std::string(name)) {}
~PayloadSimulationUser() override {
NEARBY_LOGS(INFO) << "PayloadSimulationUser: [down] name=" << name_;
// SystemClock::Sleep(kDefaultTimeout);
}
@@ -109,6 +110,8 @@ TEST_F(PayloadManagerTest, CanSendBytePayload) {
EXPECT_EQ(user_a.GetPayload().AsBytes(), ByteArray(std::string(kMessage)));
NEARBY_LOG(INFO, "Test completed.");
user_a.Stop();
user_b.Stop();
env_.Stop();
}
@@ -157,6 +160,8 @@ TEST_F(PayloadManagerTest, CanSendStreamPayload) {
rx.Close();
tx.Close();
NEARBY_LOG(INFO, "Test completed.");
user_a.Stop();
user_b.Stop();
env_.Stop();
}
@@ -213,6 +218,8 @@ TEST_F(PayloadManagerTest, CanCancelPayloadOnReceiverSide) {
rx.Close();
NEARBY_LOG(INFO, "Test completed.");
user_a.Stop();
user_b.Stop();
env_.Stop();
}
@@ -269,6 +276,8 @@ TEST_F(PayloadManagerTest, CanCancelPayloadOnSenderSide) {
rx.Close();
NEARBY_LOG(INFO, "Test completed.");
user_a.Stop();
user_b.Stop();
env_.Stop();
}
cpp/core_v2/internal/pcp_manager.cc
+12
View File
@@ -21,6 +21,18 @@ PcpManager::PcpManager(Mediums& mediums,
channel_manager);
}
void PcpManager::DisconnectFromEndpointManager() {
if (shutdown_.Set(true)) return;
for (auto& item : handlers_) {
if (!item.second) continue;
item.second->DisconnectFromEndpointManager();
}
}
PcpManager::~PcpManager() {
DisconnectFromEndpointManager();
}
Status PcpManager::StartAdvertising(ClientProxy* client,
const string& service_id,
const ConnectionOptions& options,
cpp/core_v2/internal/pcp_manager.h
+7 -4
View File
@@ -3,15 +3,16 @@
#include <string>
#include "core_v2/internal/base_pcp_handler.h"
#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/pcp_handler.h"
#include "core_v2/listeners.h"
#include "core_v2/options.h"
#include "core_v2/status.h"
#include "core_v2/strategy.h"
#include "platform_v2/public/atomic_boolean.h"
#include "absl/container/flat_hash_map.h"
namespace location {
@@ -29,7 +30,7 @@ class PcpManager {
public:
PcpManager(Mediums& mediums, EndpointChannelManager& channel_manager,
EndpointManager& endpoint_manager);
~PcpManager() = default;
~PcpManager();
Status StartAdvertising(ClientProxy* client_proxy, const string& service_id,
const ConnectionOptions& options,
@@ -48,13 +49,15 @@ class PcpManager {
Status RejectConnection(ClientProxy* client_proxy, const string& endpoint_id);
proto::connections::Medium GetBandwidthUpgradeMedium();
void DisconnectFromEndpointManager();
private:
bool SetCurrentPcpHandler(Strategy strategy);
PcpHandler* GetPcpHandler(Pcp pcp) const;
absl::flat_hash_map<Pcp, std::unique_ptr<PcpHandler>> handlers_;
PcpHandler* current_;
AtomicBoolean shutdown_{false};
absl::flat_hash_map<Pcp, std::unique_ptr<BasePcpHandler>> handlers_;
PcpHandler* current_ = nullptr;
};
} // namespace connections
cpp/core_v2/internal/pcp_manager_test.cc
+6
View File
@@ -73,6 +73,8 @@ TEST_F(PcpManagerTest, CanConnect) {
EXPECT_EQ(user_b.GetDiscovered().endpoint_name, user_a.GetName());
user_b.RequestConnection(&connection_latch);
EXPECT_TRUE(connection_latch.Await(absl::Milliseconds(1000)).result());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
@@ -93,6 +95,8 @@ TEST_F(PcpManagerTest, CanAccept) {
user_a.AcceptConnection(&accept_latch);
user_b.AcceptConnection(&accept_latch);
EXPECT_TRUE(accept_latch.Await(absl::Milliseconds(1000)).result());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
@@ -113,6 +117,8 @@ TEST_F(PcpManagerTest, CanReject) {
user_b.ExpectRejectedConnection(reject_latch);
user_a.RejectConnection(nullptr);
EXPECT_TRUE(reject_latch.Await(absl::Milliseconds(1000)).result());
user_a.Stop();
user_b.Stop();
env_.Stop();
}
cpp/core_v2/internal/service_controller.h
+12 -12
View File
@@ -35,38 +35,38 @@ class ServiceController {
ServiceController& operator=(const ServiceController&) = delete;
// Starts advertising an endpoint for a local app.
virtual Status StartAdvertising(ClientProxy* client_proxy,
virtual Status StartAdvertising(ClientProxy* client,
const std::string& service_id,
const ConnectionOptions& options,
const ConnectionRequestInfo& info) = 0;
virtual void StopAdvertising(ClientProxy* client_proxy) = 0;
virtual void StopAdvertising(ClientProxy* client) = 0;
virtual Status StartDiscovery(ClientProxy* client_proxy,
virtual Status StartDiscovery(ClientProxy* client,
const std::string& service_id,
const ConnectionOptions& options,
const DiscoveryListener& listener) = 0;
virtual void StopDiscovery(ClientProxy* client_proxy) = 0;
virtual void StopDiscovery(ClientProxy* client) = 0;
virtual Status RequestConnection(ClientProxy* client_proxy,
virtual Status RequestConnection(ClientProxy* client,
const std::string& endpoint_id,
const ConnectionRequestInfo& info) = 0;
virtual Status AcceptConnection(ClientProxy* client_proxy,
virtual Status AcceptConnection(ClientProxy* client,
const std::string& endpoint_id,
const PayloadListener& listener) = 0;
virtual Status RejectConnection(ClientProxy* client_proxy,
virtual Status RejectConnection(ClientProxy* client,
const std::string& endpoint_id) = 0;
virtual void InitiateBandwidthUpgrade(ClientProxy* client_proxy,
virtual void InitiateBandwidthUpgrade(ClientProxy* client,
const std::string& endpoint_id) = 0;
virtual void SendPayload(ClientProxy* client_proxy,
virtual void SendPayload(ClientProxy* client,
const std::vector<std::string>& endpoint_ids,
Payload payload) = 0;
virtual Status CancelPayload(ClientProxy* client_proxy,
std::int64_t payload_id) = 0;
virtual Status CancelPayload(ClientProxy* client,
Payload::Id payload_id) = 0;
virtual void DisconnectFromEndpoint(ClientProxy* client_proxy,
virtual void DisconnectFromEndpoint(ClientProxy* client,
const std::string& endpoint_id) = 0;
};
cpp/core_v2/internal/simulation_user.h
+7 -1
View File
@@ -36,7 +36,13 @@ class SimulationUser {
explicit SimulationUser(const std::string& device_name)
: name_(device_name) {}
virtual ~SimulationUser() = default;
virtual ~SimulationUser() {
Stop();
}
void Stop() {
pm_.DisconnectFromEndpointManager();
mgr_.DisconnectFromEndpointManager();
}
// Calls PcpManager::StartAdvertising.
// If latch is provided, will call latch->CountDown() in the initiated_cb
cpp/platform/impl/shared/BUILD
+1
View File
@@ -69,5 +69,6 @@ cc_test(
":file",
"//file/util:temp_path",
"//testing/base/public:gunit_main",
"//absl/strings",
],
)
cpp/platform/impl/shared/file_impl_test.cc
+2 -1
View File
@@ -7,6 +7,7 @@
#include "file/util/temp_path.h"
#include "gtest/gtest.h"
#include "absl/strings/string_view.h"
namespace location {
namespace nearby {
@@ -20,7 +21,7 @@ class FileImplTest : public ::testing::Test {
file_ = std::fstream(path_, std::fstream::in | std::fstream::out);
}
void WriteToFile(const std::string& text) {
void WriteToFile(absl::string_view text) {
file_ << text;
file_.flush();
size_ += text.size();
cpp/platform_v2/api/BUILD
-1
View File
@@ -72,6 +72,5 @@ cc_library(
":types",
"//platform_v2/base",
"//absl/strings",
"//absl/types:any",
],
)
cpp/platform_v2/api/condition_variable.h
-1
View File
@@ -27,7 +27,6 @@ class ConditionVariable {
// Waits while timeout has not expired for Notify to be called.
// May return prematurely in case of interrupt, if supported by platform.
// Returns kSuccess, or kInterrupted on interrupt.
// If Timeout expired, and Notify was not called, returns kTimeout.
virtual Exception Wait(absl::Duration timeout) = 0;
};
cpp/platform_v2/api/executor.h
+4
View File
@@ -7,6 +7,8 @@ namespace location {
namespace nearby {
namespace api {
int GetCurrentTid();
// This abstract class is the superclass of all classes representing an
// Executor.
class Executor {
@@ -19,6 +21,8 @@ class Executor {
// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html#shutdown--
virtual void Shutdown() = 0;
virtual int GetTid(int index) const = 0;
};
} // namespace api
cpp/platform_v2/base/logging.h
+6 -1
View File
@@ -29,7 +29,12 @@ class LogMessageVoidify {
location::nearby::api::LogMessage::Severity::kError
#define NEARBY_SEVERITY_FATAL \
location::nearby::api::LogMessage::Severity::kFatal
#if defined(_WIN32)
// wingdi.h defines ERROR to be 0. When we call LOG(ERROR), it gets substituted
// with 0, and it expands to NEARBY_SEVERITY_0. To allow us to keep using this
// syntax, we define this macro to do the same thing as NEARBY_SEVERITY_ERROR.
#define NEARBY_SEVERITY_0 location::nearby::api::LogMessage::Severity::kError
#endif // defined(_WIN32)
#define NEARBY_SEVERITY(severity) NEARBY_SEVERITY_##severity
// Log enabling
cpp/platform_v2/impl/g3/BUILD
-1
View File
@@ -31,7 +31,6 @@ cc_library(
"//absl/base:core_headers",
"//absl/synchronization",
"//absl/time",
"//absl/types:any",
"//thread",
],
)
cpp/platform_v2/impl/g3/condition_variable.h
+2 -3
View File
@@ -20,9 +20,8 @@ class ConditionVariable : public api::ConditionVariable {
return {Exception::kSuccess};
}
Exception Wait(absl::Duration timeout) override {
return cond_var_.WaitWithTimeout(mutex_, timeout)
? Exception{Exception::kTimeout}
: Exception{Exception::kSuccess};
cond_var_.WaitWithTimeout(mutex_, timeout);
return {Exception::kSuccess};
}
void Notify() override { cond_var_.SignalAll(); }
cpp/platform_v2/impl/g3/multi_thread_executor.h
+5
View File
@@ -33,6 +33,11 @@ class MultiThreadExecutor : public api::SubmittableExecutor {
void Shutdown() override { DoShutdown(); }
~MultiThreadExecutor() override { DoShutdown(); }
int GetTid(int index) const override {
const auto* thread = thread_pool_.thread(index);
return thread ? thread->tid() : 0;
}
void ScheduleAfter(absl::Duration delay, Runnable&& runnable) {
if (shutdown_) return;
thread_pool_.ScheduleAt(absl::Now() + delay, std::move(runnable));
cpp/platform_v2/impl/g3/platform.cc
+5
View File
@@ -47,6 +47,11 @@ std::string GetPayloadPath(PayloadId payload_id) {
}
} // namespace
int GetCurrentTid() {
const LiveThread* my = Thread_GetMyLiveThread();
return LiveThread_Pthread_TID(my);
}
std::unique_ptr<SubmittableExecutor>
ImplementationPlatform::CreateSingleThreadExecutor() {
return absl::make_unique<g3::SingleThreadExecutor>();
cpp/platform_v2/impl/g3/scheduled_executor.h
+3
View File
@@ -31,6 +31,9 @@ class ScheduledExecutor final : public api::ScheduledExecutor {
absl::Duration delay) override;
void Shutdown() override { executor_.Shutdown(); }
int GetTid(int index) const override {
return executor_.GetTid(index);
}
private:
SingleThreadExecutor executor_;
};
cpp/platform_v2/impl/shared/BUILD
+1
View File
@@ -51,5 +51,6 @@ cc_test(
"//file/util:temp_path",
"//platform_v2/base",
"//testing/base/public:gunit_main",
"//absl/strings",
],
)
cpp/platform_v2/impl/shared/file_test.cc
+2 -1
View File
@@ -8,6 +8,7 @@
#include "file/util/temp_path.h"
#include "platform_v2/base/byte_array.h"
#include "gtest/gtest.h"
#include "absl/strings/string_view.h"
namespace location {
namespace nearby {
@@ -22,7 +23,7 @@ class FileTest : public ::testing::Test {
file_ = std::fstream(path_, std::fstream::in | std::fstream::out);
}
void WriteToFile(const std::string& text) {
void WriteToFile(absl::string_view text) {
file_ << text;
file_.flush();
size_ += text.size();
cpp/platform_v2/public/BUILD
-1
View File
@@ -39,7 +39,6 @@ cc_library(
"//absl/base:core_headers",
"//absl/container:flat_hash_map",
"//absl/time",
"//absl/types:any",
],
)
cpp/platform_v2/public/condition_variable_test.cc
+7 -1
View File
@@ -3,6 +3,7 @@
#include "platform_v2/public/logging.h"
#include "platform_v2/public/mutex.h"
#include "platform_v2/public/single_thread_executor.h"
#include "platform_v2/public/system_clock.h"
#include "gmock/gmock.h"
#include "gtest/gtest.h"
#include "absl/time/time.h"
@@ -54,7 +55,12 @@ TEST(ConditionVariableTest, WaitTerminatesOnTimeoutWithoutNotify) {
Mutex mutex;
ConditionVariable cond{&mutex};
MutexLock lock(&mutex);
EXPECT_EQ(cond.Wait(absl::Milliseconds(100)), Exception{Exception::kTimeout});
const absl::Duration kWaitTime = absl::Milliseconds(100);
absl::Time start = SystemClock::ElapsedRealtime();
cond.Wait(kWaitTime);
absl::Duration duration = SystemClock::ElapsedRealtime() - start;
EXPECT_GE(duration, kWaitTime);
}
} // namespace
cpp/platform_v2/public/scheduled_executor.h
+7 -1
View File
@@ -50,6 +50,12 @@ class ScheduledExecutor final {
DoShutdown();
}
int GetTid(int index) const {
MutexLock lock(&mutex_);
return impl_->GetTid(index);
}
int Tid() const { return GetTid(0); }
Cancelable Schedule(Runnable&& runnable, absl::Duration duration)
ABSL_LOCKS_EXCLUDED(mutex_) {
MutexLock lock(&mutex_);
@@ -65,7 +71,7 @@ class ScheduledExecutor final {
}
}
Mutex mutex_;
mutable Mutex mutex_;
std::unique_ptr<api::ScheduledExecutor> ABSL_GUARDED_BY(mutex_) impl_;
};
cpp/platform_v2/public/settable_future.h
+2 -2
View File
@@ -57,8 +57,8 @@ class SettableFuture : public api::SettableFuture<T> {
MutexLock lock(&mutex_);
while (!done_) {
absl::Time start_time = SystemClock::ElapsedRealtime();
if (completed_.Wait(timeout).Raised(Exception::kTimeout)) {
SetExceptionLocked({Exception::kTimeout});
if (completed_.Wait(timeout).Raised(Exception::kInterrupted)) {
SetExceptionLocked({Exception::kInterrupted});
break;
}
absl::Duration spent = SystemClock::ElapsedRealtime() - start_time;
cpp/platform_v2/public/single_thread_executor.h
+1
View File
@@ -18,6 +18,7 @@ class SingleThreadExecutor final : public SubmittableExecutor {
~SingleThreadExecutor() override = default;
SingleThreadExecutor(SingleThreadExecutor&&) = default;
SingleThreadExecutor& operator=(SingleThreadExecutor&&) = default;
int Tid() const { return GetTid(0); }
};
} // namespace nearby
cpp/platform_v2/public/submittable_executor.h
+8 -1
View File
@@ -17,6 +17,8 @@
namespace location {
namespace nearby {
inline int GetCurrentTid() { return api::GetCurrentTid(); }
// Main interface to be used by platform as a base class for
// - MultiThreadExecutor
// - SingleThreadExecutor
@@ -41,6 +43,11 @@ class SubmittableExecutor : public api::SubmittableExecutor {
if (impl_) impl_->Execute(std::move(runnable));
}
int GetTid(int index) const ABSL_LOCKS_EXCLUDED(mutex_) override {
MutexLock lock(&mutex_);
return impl_ ? impl_->GetTid(index) : 0;
}
void Shutdown() ABSL_LOCKS_EXCLUDED(mutex_) override {
MutexLock lock(&mutex_);
DoShutdown();
@@ -86,7 +93,7 @@ class SubmittableExecutor : public api::SubmittableExecutor {
ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_) override {
return impl_ ? impl_->DoSubmit(std::move(wrapped_callable)) : false;
}
Mutex mutex_;
mutable Mutex mutex_;
std::unique_ptr<api::SubmittableExecutor> ABSL_GUARDED_BY(mutex_) impl_;
};