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

Aynchronous start and stop broadcast

Browse Source 
PiperOrigin-RevId: 491772964
This commit is contained in:
Janusz Sobczak
2022-11-29 16:59:41 -08:00
committed by Copybara-Service
parent f0d36f8d8d
commit e44f7eac61
4 changed files with 151 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
presence/implementation/BUILD
+1
View File
@@ -71,6 +71,7 @@ cc_library(
"//presence:__subpackages__",
],
deps = [
"@com_google_absl//absl/base:core_headers",
"@com_google_absl//absl/container:flat_hash_map",
"@com_google_absl//absl/container:flat_hash_set",
"@com_google_absl//absl/log:die_if_null",
presence/implementation/service_controller_impl.cc
+78 -30
View File
@@ -61,47 +61,95 @@ absl::StatusOr<BroadcastSessionId> ServiceControllerImpl::StartBroadcast(
callback.start_broadcast_cb(Status{Status::Value::kError});
return request.status();
}
absl::StatusOr<AdvertisementData> advertisement =
AdvertisementFactory(&credential_manager_).CreateAdvertisement(*request);
if (!advertisement.ok()) {
NEARBY_LOGS(WARNING) << "Can't create advertisement, reason: "
<< advertisement.status();
callback.start_broadcast_cb(Status{Status::Value::kError});
return advertisement.status();
}
std::unique_ptr<AdvertisingSession> session =
mediums_.GetBle().StartAdvertising(
*advertisement, broadcast_request.power_mode,
AdvertisingCallback{.start_advertising_result =
[callback](BleOperationStatus status) {
callback.start_broadcast_cb(
ConvertBleStatus(status));
}});
if (!session) {
callback.start_broadcast_cb(Status{Status::Value::kError});
return absl::UnavailableError("Failed to start broadcasting");
}
BroadcastSessionId id = GenerateBroadcastSessionId();
sessions_.insert({id, Session{.advertising_session = std::move(session)}});
RunOnServiceControllerThread(
"start-broadcast",
[this, id, power_mode = broadcast_request.power_mode, request = *request,
broadcast_callback =
std::move(callback)]() ABSL_EXCLUSIVE_LOCKS_REQUIRED(executor_) {
sessions_.insert({id, BroadcastSessionState(broadcast_callback)});
absl::StatusOr<AdvertisementData> advertisement =
AdvertisementFactory(&credential_manager_)
.CreateAdvertisement(request);
if (!advertisement.ok()) {
NEARBY_LOGS(WARNING) << "Can't create advertisement, reason: "
<< advertisement.status();
NotifyStartCallbackStatus(id, Status{Status::Value::kError});
return;
}
std::unique_ptr<AdvertisingSession> session =
mediums_.GetBle().StartAdvertising(
*advertisement, power_mode,
AdvertisingCallback{.start_advertising_result =
[this, id](BleOperationStatus status) {
NotifyStartCallbackStatus(
id, ConvertBleStatus(status));
}});
if (!session) {
NotifyStartCallbackStatus(id, Status{Status::Value::kError});
return;
}
sessions_.at(id).SetAdvertisingSession(std::move(session));
});
return id;
}
void ServiceControllerImpl::NotifyStartCallbackStatus(BroadcastSessionId id,
Status status) {
RunOnServiceControllerThread("started-broadcast-cb",
[this, id, status]()
ABSL_EXCLUSIVE_LOCKS_REQUIRED(executor_) {
auto it = sessions_.find(id);
if (it == sessions_.end()) {
return;
}
it->second.CallStartedCallback(status);
if (!status.Ok()) {
// Delete failed session.
sessions_.erase(it);
}
});
}
void ServiceControllerImpl::StopBroadcast(BroadcastSessionId id) {
auto it = sessions_.find(id);
if (it != sessions_.end()) {
it->second.advertising_session->stop_advertising();
sessions_.erase(it);
} else {
NEARBY_LOGS(WARNING) << absl::StrFormat(
"BroadcastSessionId(0x%x) not found", id);
}
RunOnServiceControllerThread(
"stop-broadcast", [this, id]() ABSL_EXCLUSIVE_LOCKS_REQUIRED(executor_) {
auto it = sessions_.find(id);
if (it == sessions_.end()) {
NEARBY_LOGS(VERBOSE)
<< absl::StrFormat("BroadcastSession(0x%x) not found", id);
return;
}
it->second.StopAdvertising();
sessions_.erase(it);
});
}
BroadcastSessionId ServiceControllerImpl::GenerateBroadcastSessionId() {
return absl::Uniform<BroadcastSessionId>(bit_gen_);
}
void ServiceControllerImpl::BroadcastSessionState::SetAdvertisingSession(
std::unique_ptr<AdvertisingSession> session) {
advertising_session_ = std::move(session);
}
void ServiceControllerImpl::BroadcastSessionState::CallStartedCallback(
Status status) {
BroadcastCallback callback = std::move(broadcast_callback_);
if (callback.start_broadcast_cb) {
callback.start_broadcast_cb(status);
}
}
void ServiceControllerImpl::BroadcastSessionState::StopAdvertising() {
std::unique_ptr<AdvertisingSession> advertising_session =
std::move(advertising_session_);
if (advertising_session) {
advertising_session->stop_advertising();
}
}
} // namespace presence
} // namespace nearby
presence/implementation/service_controller_impl.h
+34 -5
View File
@@ -16,8 +16,12 @@
#define THIRD_PARTY_NEARBY_PRESENCE_IMPLEMENTATION_SERVICE_CONTROLLER_IMPL_H_
#include <memory>
#include <string>
#include <utility>
#include "absl/base/thread_annotations.h"
#include "absl/random/random.h"
#include "internal/platform/single_thread_executor.h"
#include "presence/broadcast_request.h"
#include "presence/data_types.h"
#include "presence/implementation/credential_manager_impl.h"
@@ -25,6 +29,7 @@
#include "presence/implementation/scan_manager.h"
#include "presence/implementation/service_controller.h"
#include "presence/scan_request.h"
/*
* This class implements {@code ServiceController} functions. Owns mediums and
* other managers instances.
@@ -34,30 +39,54 @@ namespace presence {
class ServiceControllerImpl : public ServiceController {
public:
using SingleThreadExecutor = ::location::nearby::SingleThreadExecutor;
using AdvertisingSession =
location::nearby::api::ble_v2::BleMedium::AdvertisingSession;
::location::nearby::api::ble_v2::BleMedium::AdvertisingSession;
using Runnable = ::location::nearby::Runnable;
ServiceControllerImpl() = default;
~ServiceControllerImpl() override { executor_.Shutdown(); }
std::unique_ptr<ScanSession> StartScan(ScanRequest scan_request,
ScanCallback callback) override;
absl::StatusOr<BroadcastSessionId> StartBroadcast(
BroadcastRequest broadcast_request, BroadcastCallback callback) override;
void StopBroadcast(BroadcastSessionId) override;
SingleThreadExecutor& GetBackgroundExecutor() { return executor_; }
// Gives tests access to mediums.
Mediums& GetMediums() { return mediums_; }
private:
struct Session {
std::unique_ptr<AdvertisingSession> advertising_session;
};
class BroadcastSessionState {
public:
explicit BroadcastSessionState(BroadcastCallback broadcast_callback)
: broadcast_callback_(broadcast_callback) {}
void SetAdvertisingSession(std::unique_ptr<AdvertisingSession> session);
void CallStartedCallback(Status status);
void StopAdvertising();
private:
BroadcastCallback broadcast_callback_;
std::unique_ptr<AdvertisingSession> advertising_session_;
};
SingleThreadExecutor executor_;
BroadcastSessionId GenerateBroadcastSessionId();
void NotifyStartCallbackStatus(BroadcastSessionId id, Status status);
void RunOnServiceControllerThread(absl::string_view name, Runnable runnable) {
executor_.Execute(std::string(name), std::move(runnable));
}
Mediums mediums_; // NOLINT: further impl will use it.
CredentialManagerImpl
credential_manager_; // NOLINT: further impl will use it.
ScanManager scan_manager_{
mediums_, credential_manager_}; // NOLINT: further impl will use it.
absl::flat_hash_map<BroadcastSessionId, Session> sessions_;
absl::flat_hash_map<BroadcastSessionId, BroadcastSessionState> sessions_
ABSL_GUARDED_BY(executor_);
absl::BitGen bit_gen_;
};
presence/implementation/service_controller_impl_test.cc
+38 -18
View File
@@ -20,6 +20,7 @@
#include "gmock/gmock.h"
#include "protobuf-matchers/protocol-buffer-matchers.h"
#include "gtest/gtest.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/future.h"
#include "internal/platform/medium_environment.h"
@@ -31,6 +32,7 @@ namespace {
using FeatureFlags = ::location::nearby::FeatureFlags::Flags;
using internal::IdentityType;
using ::location::nearby::CountDownLatch;
using ::location::nearby::MediumEnvironment;
using ::testing::status::StatusIs;
@@ -61,15 +63,37 @@ class MediumEnvironmentStarter {
class ServiceControllerImplTest : public testing::TestWithParam<FeatureFlags> {
protected:
void TearDown() override { MediumEnvironment::Instance().Sync(); }
bool IsAdvertising() {
WaitForServiceControllerTasks();
MediumEnvironment::Instance().Sync();
return MediumEnvironment::Instance()
.GetBleV2MediumStatus(
*service_controller_.GetMediums().GetBle().GetImpl())
->is_advertising;
}
BroadcastCallback CreateBroadcastCallback() {
return BroadcastCallback{.start_broadcast_cb = [this](Status status) {
start_broadcast_status_.Set(status);
}};
}
void WaitForServiceControllerTasks() {
CountDownLatch latch(1);
service_controller_.GetBackgroundExecutor().Execute(
[&]() { latch.CountDown(); });
latch.Await();
}
// The medium environment must be initialized (started) before the service
// controller.
MediumEnvironmentStarter env_;
ServiceControllerImpl service_controller_;
location::nearby::Future<Status> start_broadcast_status_;
BroadcastCallback broadcast_callback_{
.start_broadcast_cb = [this](Status status) {
start_broadcast_status_.Set(status);
}};
ServiceControllerImpl service_controller_;
};
INSTANTIATE_TEST_SUITE_P(ParametrisedServiceControllerImplTest,
@@ -80,40 +104,34 @@ TEST_P(ServiceControllerImplTest, StartBroadcastPublicIdentity) {
absl::StatusOr<BroadcastSessionId> session =
service_controller_.StartBroadcast(
CreateBroadcastRequest(internal::IDENTITY_TYPE_PUBLIC),
broadcast_callback_);
CreateBroadcastCallback());
EXPECT_OK(session);
EXPECT_TRUE(start_broadcast_status_.Get().ok());
EXPECT_EQ(start_broadcast_status_.Get().GetResult(),
Status{Status::Value::kSuccess});
EXPECT_TRUE(MediumEnvironment::Instance()
.GetBleV2MediumStatus(
*service_controller_.GetMediums().GetBle().GetImpl())
->is_advertising);
EXPECT_TRUE(IsAdvertising());
}
TEST_P(ServiceControllerImplTest, StartAndStopBroadcast) {
absl::StatusOr<BroadcastSessionId> session =
service_controller_.StartBroadcast(
CreateBroadcastRequest(internal::IDENTITY_TYPE_PUBLIC),
broadcast_callback_);
CreateBroadcastCallback());
ASSERT_OK(session);
EXPECT_TRUE(IsAdvertising());
service_controller_.StopBroadcast(*session);
MediumEnvironment::Instance().Sync();
EXPECT_FALSE(MediumEnvironment::Instance()
.GetBleV2MediumStatus(
*service_controller_.GetMediums().GetBle().GetImpl())
->is_advertising);
EXPECT_FALSE(IsAdvertising());
}
TEST_P(ServiceControllerImplTest, StopBroadcastTwiceNoSideEffects) {
absl::StatusOr<BroadcastSessionId> session =
service_controller_.StartBroadcast(
CreateBroadcastRequest(internal::IDENTITY_TYPE_PUBLIC),
broadcast_callback_);
CreateBroadcastCallback());
ASSERT_OK(session);
service_controller_.StopBroadcast(*session);
EXPECT_TRUE(IsAdvertising());
service_controller_.StopBroadcast(*session);
}
@@ -125,12 +143,13 @@ TEST_P(ServiceControllerImplTest, StopBroadcastInvalidSessionNoSideEffects) {
TEST_P(ServiceControllerImplTest, StartBroadcastInvalidRequestFails) {
absl::StatusOr<BroadcastSessionId> session =
service_controller_.StartBroadcast(BroadcastRequest{},
broadcast_callback_);
CreateBroadcastCallback());
EXPECT_THAT(session, StatusIs(absl::StatusCode::kInvalidArgument));
EXPECT_TRUE(start_broadcast_status_.Get().ok());
EXPECT_EQ(start_broadcast_status_.Get().GetResult(),
Status{Status::Value::kError});
EXPECT_FALSE(IsAdvertising());
}
TEST_P(ServiceControllerImplTest, StartBroadcastPrivateIdentityFails) {
@@ -138,12 +157,13 @@ TEST_P(ServiceControllerImplTest, StartBroadcastPrivateIdentityFails) {
absl::StatusOr<BroadcastSessionId> session =
service_controller_.StartBroadcast(
CreateBroadcastRequest(internal::IDENTITY_TYPE_PRIVATE),
broadcast_callback_);
CreateBroadcastCallback());
EXPECT_THAT(session, StatusIs(absl::StatusCode::kUnavailable));
ASSERT_OK(session);
EXPECT_TRUE(start_broadcast_status_.Get().ok());
EXPECT_EQ(start_broadcast_status_.Get().GetResult(),
Status{Status::Value::kError});
EXPECT_FALSE(IsAdvertising());
}
} // namespace