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

move payload listeners to absl::AnyInvocable

Browse Source 
PiperOrigin-RevId: 533365325
This commit is contained in:
Anay Wadhera
2023-05-18 23:30:01 -07:00
committed by Copybara-Service
parent fee648aa91
commit f2cadc2815
29 changed files with 151 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connections/BUILD
+1 -1
View File
@@ -78,7 +78,7 @@ cc_library(
"//internal/platform:types",
"//internal/platform:util",
"//proto:connections_enums_cc_proto",
"@com_google_absl//absl/strings",
"@com_google_absl//absl/functional:any_invocable",
"@com_google_absl//absl/types:variant",
],
)
connections/c/core_adapter.cc
+2 -1
View File
@@ -213,7 +213,8 @@ void AcceptConnection(connections::Core *pCore, const char *endpoint_id,
}
connections::PayloadListener payload_listener =
std::move(*listener.GetImpl());
pCore->AcceptConnection(endpoint_id, payload_listener, *callback.GetImpl());
pCore->AcceptConnection(endpoint_id, std::move(payload_listener),
*callback.GetImpl());
}
void RejectConnection(connections::Core *pCore, const char *endpoint_id,
connections/c/listeners_w.cc
+4 -4
View File
@@ -193,7 +193,7 @@ PayloadListenerW::PayloadListenerW(PayloadCB payloadCB,
new connections::PayloadListener())) {
CHECK(payload_cb != nullptr);
auto pcb = payload_cb;
impl_->payload_cb = [pcb](const std::string &endpoint_id,
impl_->payload_cb = [pcb](absl::string_view endpoint_id,
connections::Payload payload) {
PayloadW payloadW;
@@ -221,13 +221,13 @@ PayloadListenerW::PayloadListenerW(PayloadCB payloadCB,
break;
}
}
pcb(endpoint_id.c_str(), payloadW);
pcb(std::string(endpoint_id).c_str(), payloadW);
};
CHECK(payload_progress_cb != nullptr);
auto ppcb = payload_progress_cb;
impl_->payload_progress_cb =
[ppcb](const std::string &endpoint_id,
[ppcb](absl::string_view endpoint_id,
connections::PayloadProgressInfo payload_progress_info) {
PayloadProgressInfoW payload_progress_info_w;
payload_progress_info_w.payload_id = payload_progress_info.payload_id;
@@ -254,7 +254,7 @@ PayloadListenerW::PayloadListenerW(PayloadCB payloadCB,
break;
}
ppcb(endpoint_id.c_str(), payload_progress_info_w);
ppcb(std::string(endpoint_id).c_str(), payload_progress_info_w);
};
}
connections/clients/ios/Internal/GNCPayloadListener.h
+2 -2
View File
@@ -41,8 +41,8 @@ class GNCPayloadListener : public PayloadListener {
GNCPayloadsProvider payloadsProvider)
: core_(core), handlers_provider_(handlersProvider), payloads_provider_(payloadsProvider) {}
void OnPayload(const std::string &endpoint_id, Payload payload);
void OnPayloadProgress(const std::string &endpoint_id, const PayloadProgressInfo &info);
void OnPayload(absl::string_view endpoint_id, Payload payload);
void OnPayloadProgress(absl::string_view endpoint_id, const PayloadProgressInfo &info);
private:
GNCCore *core_;
connections/clients/ios/Internal/GNCPayloadListener.mm
+2 -2
View File
@@ -35,7 +35,7 @@ NS_ASSUME_NONNULL_BEGIN
namespace nearby {
namespace connections {
void GNCPayloadListener::OnPayload(const std::string &endpoint_id, Payload payload) {
void GNCPayloadListener::OnPayload(absl::string_view endpoint_id, Payload payload) {
GNCConnectionHandlers *handlers = handlers_provider_();
int64_t payloadId = payload.GetId();
@@ -180,7 +180,7 @@ void GNCPayloadListener::OnPayload(const std::string &endpoint_id, Payload paylo
}
}
void GNCPayloadListener::OnPayloadProgress(const std::string &endpoint_id,
void GNCPayloadListener::OnPayloadProgress(absl::string_view endpoint_id,
const PayloadProgressInfo &info) {
// Note: The logic in this callback for handling progress updates and payload completion is
// identical for Bytes, Stream and File payloads.
connections/core.cc
+2 -1
View File
@@ -128,7 +128,8 @@ void Core::AcceptConnection(absl::string_view endpoint_id,
PayloadListener listener, ResultCallback callback) {
assert(!endpoint_id.empty());
router_->AcceptConnection(&client_, endpoint_id, listener, callback);
router_->AcceptConnection(&client_, endpoint_id, std::move(listener),
callback);
}
void Core::RejectConnection(absl::string_view endpoint_id,
connections/implementation/base_pcp_handler.cc
+11 -9
View File
@@ -870,13 +870,14 @@ bool BasePcpHandler::AutoUpgradeBandwidth(
return local_advertising_options.auto_upgrade_bandwidth;
}
Status BasePcpHandler::AcceptConnection(
ClientProxy* client, const std::string& endpoint_id,
const PayloadListener& payload_listener) {
Status BasePcpHandler::AcceptConnection(ClientProxy* client,
const std::string& endpoint_id,
PayloadListener payload_listener) {
Future<Status> response;
RunOnPcpHandlerThread(
"accept-connection", [this, client, endpoint_id, payload_listener,
&response]() RUN_ON_PCP_HANDLER_THREAD() {
"accept-connection", [this, client, endpoint_id,
payload_listener = std::move(payload_listener),
&response]() RUN_ON_PCP_HANDLER_THREAD() mutable {
NEARBY_LOGS(INFO) << "AcceptConnection: endpoint_id=" << endpoint_id;
if (!pending_connections_.count(endpoint_id)) {
NEARBY_LOGS(INFO)
@@ -918,8 +919,8 @@ Status BasePcpHandler::AcceptConnection(
NEARBY_LOGS(INFO) << "AcceptConnection: accepting locally: endpoint_id="
<< endpoint_id;
connection_info.LocalEndpointAcceptedConnection(endpoint_id,
payload_listener);
connection_info.LocalEndpointAcceptedConnection(
endpoint_id, std::move(payload_listener));
EvaluateConnectionResult(client, endpoint_id,
false /* can_close_immediately */);
response.Set({Status::kSuccess});
@@ -1722,8 +1723,9 @@ BasePcpHandler::PendingConnectionInfo::~PendingConnectionInfo() {
}
void BasePcpHandler::PendingConnectionInfo::LocalEndpointAcceptedConnection(
const std::string& endpoint_id, const PayloadListener& payload_listener) {
client->LocalEndpointAcceptedConnection(endpoint_id, payload_listener);
const std::string& endpoint_id, PayloadListener payload_listener) {
client->LocalEndpointAcceptedConnection(endpoint_id,
std::move(payload_listener));
}
void BasePcpHandler::PendingConnectionInfo::LocalEndpointRejectedConnection(
connections/implementation/base_pcp_handler.h
+3 -5
View File
@@ -69,7 +69,6 @@ class BasePcpHandler : public PcpHandler,
public:
using FrameProcessor = EndpointManager::FrameProcessor;
// TODO(apolyudov): Add SecureRandom.
BasePcpHandler(Mediums* mediums, EndpointManager* endpoint_manager,
EndpointChannelManager* channel_manager,
BwuManager* bwu_manager, Pcp pcp);
@@ -118,7 +117,7 @@ class BasePcpHandler : public PcpHandler,
// 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,
const PayloadListener& payload_listener) override;
PayloadListener payload_listener) override;
// Called by either party to reject connection on their part.
// If either party does call it, connection will terminate.
@@ -326,9 +325,8 @@ class BasePcpHandler : public PcpHandler,
void SetCryptoContext(std::unique_ptr<securegcm::UKey2Handshake> ukey2);
// Pass Accept notification to client.
void LocalEndpointAcceptedConnection(
const std::string& endpoint_id,
const PayloadListener& payload_listener);
void LocalEndpointAcceptedConnection(const std::string& endpoint_id,
PayloadListener payload_listener);
// Pass Reject notification to client.
void LocalEndpointRejectedConnection(const std::string& endpoint_id);
connections/implementation/client_proxy.cc
+65 -56
View File
@@ -83,9 +83,9 @@ std::string ClientProxy::GetLocalEndpointId() {
}
std::string ClientProxy::GetConnectionToken(const std::string& endpoint_id) {
Connection* item = LookupConnection(endpoint_id);
ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
return item->connection_token;
return item->first.connection_token;
}
return {};
}
@@ -278,12 +278,18 @@ void ClientProxy::OnConnectionInitiated(
// still need to accept this connection, so set its establishment status to
// PENDING.
auto result = connections_.emplace(
endpoint_id, Connection{
.is_incoming = info.is_incoming_connection,
.connection_listener = listener,
.connection_options = connection_options,
.connection_token = connection_token,
});
endpoint_id, std::make_pair(
Connection{
.is_incoming = info.is_incoming_connection,
.connection_listener = listener,
.connection_options = connection_options,
.connection_token = connection_token,
},
PayloadListener{
.payload_cb = [](absl::string_view, Payload) {},
.payload_progress_cb = [](absl::string_view,
PayloadProgressInfo) {},
}));
// Instead of using structured binding which is nice, but banned
// (can not use c++17 features, until chromium does) we unpack manually.
auto& pair_iter = result.first;
@@ -293,12 +299,12 @@ void ClientProxy::OnConnectionInitiated(
<< GetClientId() << "; endpoint_id=" << endpoint_id
<< "; inserted=" << inserted;
DCHECK(inserted);
const Connection& item = pair_iter->second;
const ConnectionPair& item = pair_iter->second;
// Notify the client.
//
// Note: we allow devices to connect to an advertiser even after it stops
// advertising, so no need to check IsAdvertising() here.
item.connection_listener.initiated_cb(endpoint_id, info);
item.first.connection_listener.initiated_cb(endpoint_id, info);
if (info.is_incoming_connection) {
// Add CancellationFlag for advertisers once encryption succeeds.
@@ -320,10 +326,10 @@ void ClientProxy::OnConnectionAccepted(const std::string& endpoint_id) {
}
// Notify the client.
Connection* item = LookupConnection(endpoint_id);
ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
item->connection_listener.accepted_cb(endpoint_id);
item->status = Connection::kConnected;
item->first.connection_listener.accepted_cb(endpoint_id);
item->first.status = Connection::kConnected;
}
}
@@ -339,9 +345,9 @@ void ClientProxy::OnConnectionRejected(const std::string& endpoint_id,
}
// Notify the client.
const Connection* item = LookupConnection(endpoint_id);
const ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
item->connection_listener.rejected_cb(endpoint_id, status);
item->first.connection_listener.rejected_cb(endpoint_id, status);
OnDisconnected(endpoint_id, false /* notify */);
}
}
@@ -350,9 +356,10 @@ void ClientProxy::OnBandwidthChanged(const std::string& endpoint_id,
Medium new_medium) {
MutexLock lock(&mutex_);
const Connection* item = LookupConnection(endpoint_id);
const ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
item->connection_listener.bandwidth_changed_cb(endpoint_id, new_medium);
item->first.connection_listener.bandwidth_changed_cb(endpoint_id,
new_medium);
NEARBY_LOGS(INFO) << "ClientProxy [reporting onBandwidthChanged]: client="
<< GetClientId() << "; endpoint_id=" << endpoint_id;
}
@@ -361,10 +368,10 @@ void ClientProxy::OnBandwidthChanged(const std::string& endpoint_id,
void ClientProxy::OnDisconnected(const std::string& endpoint_id, bool notify) {
MutexLock lock(&mutex_);
const Connection* item = LookupConnection(endpoint_id);
const ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
if (notify) {
item->connection_listener.disconnected_cb({endpoint_id});
item->first.connection_listener.disconnected_cb({endpoint_id});
}
connections_.erase(endpoint_id);
OnSessionComplete();
@@ -377,9 +384,9 @@ bool ClientProxy::ConnectionStatusMatches(const std::string& endpoint_id,
Connection::Status status) const {
MutexLock lock(&mutex_);
const Connection* item = LookupConnection(endpoint_id);
const ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
return item->status == status;
return item->first.status == status;
}
return false;
}
@@ -388,9 +395,9 @@ BooleanMediumSelector ClientProxy::GetUpgradeMediums(
const std::string& endpoint_id) const {
MutexLock lock(&mutex_);
const Connection* item = LookupConnection(endpoint_id);
const ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
return item->connection_options.allowed;
return item->first.connection_options.allowed;
}
return {};
}
@@ -398,9 +405,9 @@ BooleanMediumSelector ClientProxy::GetUpgradeMediums(
bool ClientProxy::Is5GHzSupported(const std::string& endpoint_id) const {
MutexLock lock(&mutex_);
const Connection* item = LookupConnection(endpoint_id);
const ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
return item->connection_options.connection_info.supports_5_ghz;
return item->first.connection_options.connection_info.supports_5_ghz;
}
return false;
}
@@ -408,9 +415,9 @@ bool ClientProxy::Is5GHzSupported(const std::string& endpoint_id) const {
std::string ClientProxy::GetBssid(const std::string& endpoint_id) const {
MutexLock lock(&mutex_);
const Connection* item = LookupConnection(endpoint_id);
const ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
return item->connection_options.connection_info.bssid;
return item->first.connection_options.connection_info.bssid;
}
return {};
}
@@ -418,9 +425,9 @@ std::string ClientProxy::GetBssid(const std::string& endpoint_id) const {
std::int32_t ClientProxy::GetApFrequency(const std::string& endpoint_id) const {
MutexLock lock(&mutex_);
const Connection* item = LookupConnection(endpoint_id);
const ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
return item->connection_options.connection_info.ap_frequency;
return item->first.connection_options.connection_info.ap_frequency;
}
return -1;
}
@@ -428,9 +435,9 @@ std::int32_t ClientProxy::GetApFrequency(const std::string& endpoint_id) const {
std::string ClientProxy::GetIPAddress(const std::string& endpoint_id) const {
MutexLock lock(&mutex_);
const Connection* item = LookupConnection(endpoint_id);
const ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
return item->connection_options.connection_info.ip_address;
return item->first.connection_options.connection_info.ip_address;
}
return {};
}
@@ -447,8 +454,8 @@ std::vector<std::string> ClientProxy::GetMatchingEndpoints(
for (const auto& pair : connections_) {
const auto& endpoint_id = pair.first;
const auto& connection = pair.second;
if (pred(connection)) {
const auto& connection_pair = pair.second;
if (pred(connection_pair.first)) {
connected_endpoints.push_back(endpoint_id);
}
}
@@ -487,9 +494,9 @@ bool ClientProxy::HasPendingConnectionToEndpoint(
const std::string& endpoint_id) const {
MutexLock lock(&mutex_);
const Connection* item = LookupConnection(endpoint_id);
const ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
return item->status != Connection::kConnected;
return item->first.status != Connection::kConnected;
}
return false;
}
@@ -515,7 +522,7 @@ bool ClientProxy::HasRemoteEndpointResponded(
}
void ClientProxy::LocalEndpointAcceptedConnection(
const std::string& endpoint_id, const PayloadListener& listener) {
const std::string& endpoint_id, PayloadListener listener) {
MutexLock lock(&mutex_);
if (HasLocalEndpointResponded(endpoint_id)) {
@@ -526,9 +533,9 @@ void ClientProxy::LocalEndpointAcceptedConnection(
}
AppendConnectionStatus(endpoint_id, Connection::kLocalEndpointAccepted);
Connection* item = LookupConnection(endpoint_id);
ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
item->payload_listener = listener;
item->second = std::move(listener);
}
analytics_recorder_->OnLocalEndpointAccepted(endpoint_id);
}
@@ -642,18 +649,18 @@ const OsInfo& ClientProxy::GetLocalOsInfo() const {
std::optional<OsInfo> ClientProxy::GetRemoteOsInfo(
absl::string_view endpoint_id) const {
const Connection* item = LookupConnection(endpoint_id);
const ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
return item->os_info;
return item->first.os_info;
}
return std::nullopt;
}
void ClientProxy::SetRemoteOsInfo(absl::string_view endpoint_id,
const OsInfo& remote_os_info) {
Connection* item = LookupConnection(endpoint_id);
ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
item->os_info.emplace(remote_os_info);
item->first.os_info.emplace(remote_os_info);
}
}
void ClientProxy::CancelAllEndpoints() {
@@ -671,23 +678,24 @@ void ClientProxy::OnPayload(const std::string& endpoint_id, Payload payload) {
MutexLock lock(&mutex_);
if (IsConnectedToEndpoint(endpoint_id)) {
const Connection* item = LookupConnection(endpoint_id);
const std::pair<ClientProxy::Connection, PayloadListener>* item =
LookupConnection(endpoint_id);
if (item != nullptr) {
NEARBY_LOGS(INFO) << "ClientProxy [reporting onPayloadReceived]: client="
<< GetClientId() << "; endpoint_id=" << endpoint_id
<< " ; payload_id=" << payload.GetId();
item->payload_listener.payload_cb(endpoint_id, std::move(payload));
item->second.payload_cb(endpoint_id, std::move(payload));
}
}
}
const ClientProxy::Connection* ClientProxy::LookupConnection(
const ClientProxy::ConnectionPair* ClientProxy::LookupConnection(
absl::string_view endpoint_id) const {
auto item = connections_.find(endpoint_id);
return item != connections_.end() ? &item->second : nullptr;
}
ClientProxy::Connection* ClientProxy::LookupConnection(
ClientProxy::ConnectionPair* ClientProxy::LookupConnection(
absl::string_view endpoint_id) {
auto item = connections_.find(endpoint_id);
return item != connections_.end() ? &item->second : nullptr;
@@ -698,9 +706,10 @@ void ClientProxy::OnPayloadProgress(const std::string& endpoint_id,
MutexLock lock(&mutex_);
if (IsConnectedToEndpoint(endpoint_id)) {
Connection* item = LookupConnection(endpoint_id);
std::pair<ClientProxy::Connection, PayloadListener>* item =
LookupConnection(endpoint_id);
if (item != nullptr) {
item->payload_listener.payload_progress_cb(endpoint_id, info);
item->second.payload_progress_cb(endpoint_id, info);
if (info.status == PayloadProgressInfo::Status::kInProgress) {
NEARBY_LOGS(VERBOSE)
@@ -742,19 +751,19 @@ void ClientProxy::OnSessionComplete() {
bool ClientProxy::ConnectionStatusesContains(
const std::string& endpoint_id, Connection::Status status_to_match) const {
const Connection* item = LookupConnection(endpoint_id);
const ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
return (item->status & status_to_match) != 0;
return (item->first.status & status_to_match) != 0;
}
return false;
}
void ClientProxy::AppendConnectionStatus(const std::string& endpoint_id,
Connection::Status status_to_append) {
Connection* item = LookupConnection(endpoint_id);
ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
item->status =
static_cast<Connection::Status>(item->status | status_to_append);
item->first.status =
static_cast<Connection::Status>(item->first.status | status_to_append);
}
}
@@ -870,10 +879,10 @@ std::string ClientProxy::Dump() {
for (auto it = connections_.begin(); it != connections_.end(); ++it) {
// TODO(deling): write Connection.ToString()
sstream << " " << it->first << " :(connection token) "
<< it->second.connection_token << ", (remote os type) "
<< (it->second.os_info.has_value()
<< it->second.first.connection_token << ", (remote os type) "
<< (it->second.first.os_info.has_value()
? location::nearby::connections::OsInfo::OsType_Name(
it->second.os_info->type())
it->second.first.os_info->type())
: "unknown")
<< std::endl;
}
connections/implementation/client_proxy.h
+5 -5
View File
@@ -157,7 +157,7 @@ class ClientProxy final {
bool HasRemoteEndpointResponded(const std::string& endpoint_id) const;
// Marks the local endpoint as having accepted the connection.
void LocalEndpointAcceptedConnection(const std::string& endpoint_id,
const PayloadListener& listener);
PayloadListener listener);
// Marks the local endpoint as having rejected the connection.
void LocalEndpointRejectedConnection(const std::string& endpoint_id);
// Marks the remote endpoint as having accepted the connection.
@@ -236,13 +236,13 @@ class ClientProxy final {
bool is_incoming{false};
Status status{kPending};
ConnectionListener connection_listener;
PayloadListener payload_listener;
ConnectionOptions connection_options;
DiscoveryOptions discovery_options;
AdvertisingOptions advertising_options;
std::string connection_token;
std::optional<location::nearby::connections::OsInfo> os_info;
};
using ConnectionPair = std::pair<Connection, PayloadListener>;
struct AdvertisingInfo {
std::string service_id;
@@ -265,8 +265,8 @@ class ClientProxy final {
void AppendConnectionStatus(const std::string& endpoint_id,
Connection::Status status_to_append);
const Connection* LookupConnection(absl::string_view endpoint_id) const;
Connection* LookupConnection(absl::string_view endpoint_id);
const ConnectionPair* LookupConnection(absl::string_view endpoint_id) const;
ConnectionPair* LookupConnection(absl::string_view endpoint_id);
bool ConnectionStatusMatches(const std::string& endpoint_id,
Connection::Status status) const;
std::vector<std::string> GetMatchingEndpoints(
@@ -323,7 +323,7 @@ class ClientProxy final {
DiscoveryOptions discovery_options_;
// Maps endpoint_id to endpoint connection state.
absl::flat_hash_map<std::string, Connection> connections_;
absl::flat_hash_map<std::string, ConnectionPair> connections_;
// A cache of endpoint ids that we've already notified the discoverer of. We
// check this cache before calling onEndpointFound() so that we don't notify
connections/implementation/client_proxy_test.cc
+9 -8
View File
@@ -85,9 +85,9 @@ class ClientProxyTest : public ::testing::TestWithParam<FeatureFlags::Flags> {
};
struct MockPayloadListener {
StrictMock<
MockFunction<void(const std::string& endpoint_id, Payload payload)>>
MockFunction<void(absl::string_view endpoint_id, Payload payload)>>
payload_cb;
StrictMock<MockFunction<void(const std::string& endpoint_id,
StrictMock<MockFunction<void(absl::string_view endpoint_id,
const PayloadProgressInfo& info)>>
payload_progress_cb;
};
@@ -178,7 +178,13 @@ class ClientProxyTest : public ::testing::TestWithParam<FeatureFlags::Flags> {
const Endpoint& endpoint) {
EXPECT_TRUE(client->HasPendingConnectionToEndpoint(endpoint.id));
EXPECT_FALSE(client->HasLocalEndpointResponded(endpoint.id));
client->LocalEndpointAcceptedConnection(endpoint.id, payload_listener_);
client->LocalEndpointAcceptedConnection(
endpoint.id,
{
.payload_cb = mock_discovery_payload_.payload_cb.AsStdFunction(),
.payload_progress_cb =
mock_discovery_payload_.payload_progress_cb.AsStdFunction(),
});
EXPECT_TRUE(client->HasLocalEndpointResponded(endpoint.id));
EXPECT_TRUE(client->LocalConnectionIsAccepted(endpoint.id));
}
@@ -292,11 +298,6 @@ class ClientProxyTest : public ::testing::TestWithParam<FeatureFlags::Flags> {
.endpoint_found_cb = mock_discovery_.endpoint_found_cb.AsStdFunction(),
.endpoint_lost_cb = mock_discovery_.endpoint_lost_cb.AsStdFunction(),
};
PayloadListener payload_listener_{
.payload_cb = mock_discovery_payload_.payload_cb.AsStdFunction(),
.payload_progress_cb =
mock_discovery_payload_.payload_progress_cb.AsStdFunction(),
};
ConnectionOptions connection_options_;
AdvertisingOptions advertising_options_;
DiscoveryOptions discovery_options_;
connections/implementation/mock_service_controller.h
+1 -1
View File
@@ -60,7 +60,7 @@ class MockServiceController : public ServiceController {
MOCK_METHOD(Status, AcceptConnection,
(ClientProxy * client, const std::string& endpoint_id,
const PayloadListener& listener),
PayloadListener listener),
(override));
MOCK_METHOD(Status, RejectConnection,
connections/implementation/mock_service_controller_router.h
+1 -1
View File
@@ -60,7 +60,7 @@ class MockServiceControllerRouter : public ServiceControllerRouter {
MOCK_METHOD(void, AcceptConnection,
(ClientProxy * client, absl::string_view endpoint_id,
const PayloadListener& listener, const ResultCallback& callback),
PayloadListener listener, const ResultCallback& callback),
(override));
MOCK_METHOD(void, RejectConnection,
connections/implementation/offline_service_controller.cc
+3 -2
View File
@@ -89,12 +89,13 @@ Status OfflineServiceController::RequestConnection(
Status OfflineServiceController::AcceptConnection(
ClientProxy* client, const std::string& endpoint_id,
const PayloadListener& listener) {
PayloadListener listener) {
if (stop_) return {Status::kOutOfOrderApiCall};
NEARBY_LOGS(INFO) << "Client " << client->GetClientId()
<< " accepted the connection with endpoint_id="
<< endpoint_id;
return pcp_manager_.AcceptConnection(client, endpoint_id, listener);
return pcp_manager_.AcceptConnection(client, endpoint_id,
std::move(listener));
}
Status OfflineServiceController::RejectConnection(
connections/implementation/offline_service_controller.h
+1 -1
View File
@@ -59,7 +59,7 @@ class OfflineServiceController : public ServiceController {
const ConnectionRequestInfo& info,
const ConnectionOptions& connection_options) override;
Status AcceptConnection(ClientProxy* client, const std::string& endpoint_id,
const PayloadListener& listener) override;
PayloadListener listener) override;
Status RejectConnection(ClientProxy* client,
const std::string& endpoint_id) override;
connections/implementation/offline_simulation_user.cc
+2 -2
View File
@@ -72,13 +72,13 @@ void OfflineSimulationUser::OnEndpointLost(const std::string& endpoint_id) {
if (lost_latch_) lost_latch_->CountDown();
}
void OfflineSimulationUser::OnPayload(const std::string& endpoint_id,
void OfflineSimulationUser::OnPayload(absl::string_view endpoint_id,
Payload payload) {
payload_ = std::move(payload);
if (payload_latch_) payload_latch_->CountDown();
}
void OfflineSimulationUser::OnPayloadProgress(const std::string& endpoint_id,
void OfflineSimulationUser::OnPayloadProgress(absl::string_view endpoint_id,
const PayloadProgressInfo& info) {
MutexLock lock(&progress_mutex_);
progress_info_ = info;
connections/implementation/offline_simulation_user.h
+2 -2
View File
@@ -172,8 +172,8 @@ class OfflineSimulationUser {
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,
void OnPayload(absl::string_view endpoint_id, Payload payload);
void OnPayloadProgress(absl::string_view endpoint_id,
const PayloadProgressInfo& info);
std::string service_id_;
connections/implementation/pcp_handler.h
+1 -1
View File
@@ -105,7 +105,7 @@ class PcpHandler {
// Update state in ClientProxy.
virtual Status AcceptConnection(ClientProxy* client,
const std::string& endpoint_id,
const PayloadListener& payload_listener) = 0;
PayloadListener payload_listener) = 0;
// Either party may call this to reject connection on their part before
// connection reaches data phase. If either party does call it, connection
connections/implementation/pcp_manager.cc
+3 -2
View File
@@ -109,12 +109,13 @@ Status PcpManager::RequestConnection(
Status PcpManager::AcceptConnection(ClientProxy* client,
const string& endpoint_id,
const PayloadListener& payload_listener) {
PayloadListener payload_listener) {
if (!current_) {
return {Status::kOutOfOrderApiCall};
}
return current_->AcceptConnection(client, endpoint_id, payload_listener);
return current_->AcceptConnection(client, endpoint_id,
std::move(payload_listener));
}
Status PcpManager::RejectConnection(ClientProxy* client,
connections/implementation/pcp_manager.h
+1 -1
View File
@@ -64,7 +64,7 @@ class PcpManager {
const ConnectionRequestInfo& info,
const ConnectionOptions& connection_options);
Status AcceptConnection(ClientProxy* client, const string& endpoint_id,
const PayloadListener& payload_listener);
PayloadListener payload_listener);
Status RejectConnection(ClientProxy* client, const string& endpoint_id);
location::nearby::proto::connections::Medium GetBandwidthUpgradeMedium();
connections/implementation/service_controller.h
+1 -1
View File
@@ -83,7 +83,7 @@ class ServiceController {
const ConnectionOptions& connection_options) = 0;
virtual Status AcceptConnection(ClientProxy* client,
const std::string& endpoint_id,
const PayloadListener& listener) = 0;
PayloadListener listener) = 0;
virtual Status RejectConnection(ClientProxy* client,
const std::string& endpoint_id) = 0;
connections/implementation/service_controller_router.cc
+4 -4
View File
@@ -191,12 +191,12 @@ void ServiceControllerRouter::RequestConnection(
void ServiceControllerRouter::AcceptConnection(ClientProxy* client,
absl::string_view endpoint_id,
const PayloadListener& listener,
PayloadListener listener,
const ResultCallback& callback) {
RouteToServiceController(
"scr-accept-connection",
[this, client, endpoint_id = std::string(endpoint_id), listener,
callback]() {
[this, client, endpoint_id = std::string(endpoint_id),
listener = std::move(listener), callback]() mutable {
if (client->IsConnectedToEndpoint(endpoint_id)) {
callback.result_cb({Status::kAlreadyConnectedToEndpoint});
return;
@@ -213,7 +213,7 @@ void ServiceControllerRouter::AcceptConnection(ClientProxy* client,
}
callback.result_cb(GetServiceController()->AcceptConnection(
client, endpoint_id, listener));
client, endpoint_id, std::move(listener)));
});
}
connections/implementation/service_controller_router.h
+1 -1
View File
@@ -85,7 +85,7 @@ class ServiceControllerRouter {
const ResultCallback& callback);
virtual void AcceptConnection(ClientProxy* client,
absl::string_view endpoint_id,
const PayloadListener& listener,
PayloadListener listener,
const ResultCallback& callback);
virtual void RejectConnection(ClientProxy* client,
absl::string_view endpoint_id,
connections/implementation/service_controller_router_test.cc
+9 -9
View File
@@ -158,7 +158,6 @@ class ServiceControllerRouterTest : public testing::Test {
}
void AcceptConnection(ClientProxy* client, const std::string endpoint_id,
const PayloadListener& listener,
const ResultCallback& callback) {
EXPECT_CALL(*mock_, AcceptConnection)
.WillOnce(Return(Status{Status::kSuccess}));
@@ -167,11 +166,13 @@ class ServiceControllerRouterTest : public testing::Test {
{
MutexLock lock(&mutex_);
complete_ = false;
router_.AcceptConnection(client, endpoint_id, listener, callback);
router_.AcceptConnection(client, endpoint_id, {},
callback);
while (!complete_) cond_.Wait();
EXPECT_EQ(result_, Status{Status::kSuccess});
}
client->LocalEndpointAcceptedConnection(endpoint_id, listener);
client->LocalEndpointAcceptedConnection(endpoint_id,
{});
client->RemoteEndpointAcceptedConnection(endpoint_id);
EXPECT_TRUE(client->IsConnectionAccepted(endpoint_id));
client->OnConnectionAccepted(endpoint_id);
@@ -314,7 +315,6 @@ class ServiceControllerRouterTest : public testing::Test {
};
DiscoveryListener discovery_listener_;
PayloadListener payload_listener_;
Mutex mutex_;
ConditionVariable cond_{&mutex_};
@@ -372,7 +372,7 @@ TEST_F(ServiceControllerRouterTest, AcceptConnectionCalled) {
RequestConnection(&client_, kRemoteEndpointId, kConnectionRequestInfo,
kCallback);
// Now, we can accept connection.
AcceptConnection(&client_, kRemoteEndpointId, payload_listener_, kCallback);
AcceptConnection(&client_, kRemoteEndpointId, kCallback);
}
TEST_F(ServiceControllerRouterTest, RejectConnectionCalled) {
@@ -394,7 +394,7 @@ TEST_F(ServiceControllerRouterTest, InitiateBandwidthUpgradeCalled) {
RequestConnection(&client_, kRemoteEndpointId, kConnectionRequestInfo,
kCallback);
// Now, we can accept connection.
AcceptConnection(&client_, kRemoteEndpointId, payload_listener_, kCallback);
AcceptConnection(&client_, kRemoteEndpointId, kCallback);
// Now we can change connection bandwidth.
InitiateBandwidthUpgrade(&client_, kRemoteEndpointId, kCallback);
}
@@ -407,7 +407,7 @@ TEST_F(ServiceControllerRouterTest, SendPayloadCalled) {
RequestConnection(&client_, kRemoteEndpointId, kConnectionRequestInfo,
kCallback);
// Now, we can accept connection.
AcceptConnection(&client_, kRemoteEndpointId, payload_listener_, kCallback);
AcceptConnection(&client_, kRemoteEndpointId, kCallback);
// Now we can send payload.
SendPayload(&client_, std::vector<std::string>{kRemoteEndpointId},
Payload{ByteArray("data")}, kCallback);
@@ -421,7 +421,7 @@ TEST_F(ServiceControllerRouterTest, CancelPayloadCalled) {
RequestConnection(&client_, kRemoteEndpointId, kConnectionRequestInfo,
kCallback);
// Now, we can accept connection.
AcceptConnection(&client_, kRemoteEndpointId, payload_listener_, kCallback);
AcceptConnection(&client_, kRemoteEndpointId, kCallback);
// We have to know payload id, before we can cancel payload transfer.
// It is either after a call to SendPayload, or after receiving
// PayloadProgress callback. Let's assume we have it, and proceed.
@@ -436,7 +436,7 @@ TEST_F(ServiceControllerRouterTest, DisconnectFromEndpointCalled) {
RequestConnection(&client_, kRemoteEndpointId, kConnectionRequestInfo,
kCallback);
// Now, we can accept connection.
AcceptConnection(&client_, kRemoteEndpointId, payload_listener_, kCallback);
AcceptConnection(&client_, kRemoteEndpointId, kCallback);
// We can disconnect at any time after RequestConnection.
DisconnectFromEndpoint(&client_, kRemoteEndpointId, kCallback);
}
connections/implementation/simulation_user.cc
+2 -3
View File
@@ -63,13 +63,12 @@ void SimulationUser::OnEndpointLost(const std::string& endpoint_id) {
if (lost_latch_) lost_latch_->CountDown();
}
void SimulationUser::OnPayload(const std::string& endpoint_id,
Payload payload) {
void SimulationUser::OnPayload(absl::string_view endpoint_id, Payload payload) {
payload_ = std::move(payload);
if (payload_latch_) payload_latch_->CountDown();
}
void SimulationUser::OnPayloadProgress(const std::string& endpoint_id,
void SimulationUser::OnPayloadProgress(absl::string_view endpoint_id,
const PayloadProgressInfo& info) {
MutexLock lock(&progress_mutex_);
progress_info_ = info;
connections/implementation/simulation_user.h
+2 -2
View File
@@ -141,8 +141,8 @@ class SimulationUser {
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,
void OnPayload(absl::string_view, Payload payload);
void OnPayloadProgress(absl::string_view endpoint_id,
const PayloadProgressInfo& info);
std::string service_id_;
connections/listeners.h
+6 -5
View File
@@ -27,6 +27,7 @@
// default-initialized.
// - callbacks may be initialized with lambdas; lambda definitions are concize.
#include "absl/functional/any_invocable.h"
#include "connections/connection_options.h"
#include "connections/payload.h"
#include "connections/status.h"
@@ -170,8 +171,8 @@ struct PayloadListener {
// endpoint_id - The identifier for the remote endpoint that sent the
// payload.
// payload - The Payload object received.
std::function<void(const std::string& endpoint_id, Payload payload)>
payload_cb = [](const std::string&, Payload) {};
absl::AnyInvocable<void(absl::string_view endpoint_id, Payload payload) const>
payload_cb = [](absl::string_view, Payload) {};
// Called with progress information about an active Payload transfer, either
// incoming or outgoing.
@@ -180,10 +181,10 @@ struct PayloadListener {
// receiving this payload.
// info - The PayloadProgressInfo structure describing the status of
// the transfer.
std::function<void(const std::string& endpoint_id,
const PayloadProgressInfo& info)>
absl::AnyInvocable<void(absl::string_view endpoint_id,
const PayloadProgressInfo& info)>
payload_progress_cb =
[](const std::string&, const PayloadProgressInfo&) {};
[](absl::string_view, const PayloadProgressInfo&) {};
};
} // namespace connections
connections/listeners_test.cc
+1 -1
View File
@@ -62,7 +62,7 @@ TEST(ListenersTest, PayloadListener_PayloadCb_Works) {
PayloadListener listener{
.payload_cb =
[&](const std::string& endpoint_id, Payload payload) {
[&](absl::string_view endpoint_id, Payload payload) {
if (payload.AsBytes().data() == input_bytes) {
payload_content_match = true;
}
connections/swift/NearbyCoreAdapter/Sources/GNCCoreAdapter.mm
+4 -4
View File
@@ -278,14 +278,14 @@ GNCStatus GNCStatusFromCppStatus(Status status) {
std::string endpoint_id = [endpointID cStringUsingEncoding:[NSString defaultCStringEncoding]];
PayloadListener listener;
listener.payload_cb = ^(const std::string &endpoint_id, Payload payload) {
NSString *endpointID = @(endpoint_id.c_str());
listener.payload_cb = [&delegate](absl::string_view endpoint_id, Payload payload) {
NSString *endpointID = @(std::string(endpoint_id).c_str());
GNCPayload *gncPayload = [GNCPayload fromCpp:std::move(payload)];
[delegate receivedPayload:gncPayload fromEndpoint:endpointID];
};
listener.payload_progress_cb =
^(const std::string &endpoint_id, const PayloadProgressInfo &info) {
NSString *endpointID = @(endpoint_id.c_str());
[&delegate](absl::string_view endpoint_id, const PayloadProgressInfo &info) {
NSString *endpointID = @(std::string(endpoint_id).c_str());
GNCPayloadStatus status;
switch (info.status) {
case PayloadProgressInfo::Status::kSuccess: