Refactor broadcasting code into BroadcastManager

PiperOrigin-RevId: 495021310

presence/implementation/BUILD

@@ -43,6 +43,7 @@ cc_library(
         "advertisement_decoder.cc",
         "advertisement_factory.cc",
         "base_broadcast_request.cc",
+        "broadcast_manager.cc",
         "credential_manager_impl.cc",
         "encryption.cc",
         "ldt.cc",
@@ -169,9 +170,9 @@ cc_test(
 )
 
 cc_test(
-    name = "service_controller_impl_test",
+    name = "broadcast_manager_test",
     size = "small",
-    srcs = ["service_controller_impl_test.cc"],
+    srcs = ["broadcast_manager_test.cc"],
     deps = [
         ":internal",
         "//internal/platform:base",
@@ -179,6 +180,7 @@ cc_test(
         "//internal/platform:types",
         "//internal/platform/implementation/g3",  # build_cleaner: keep
         "//internal/proto:credential_cc_proto",
+        "//presence/implementation/mediums",
         "@com_github_protobuf_matchers//protobuf-matchers",
         "@com_google_googletest//:gtest_main",
     ],

presence/implementation/broadcast_manager.cc

@@ -0,0 +1,185 @@
+// Copyright 2022 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 "presence/implementation/broadcast_manager.h"
+
+#include <memory>
+#include <utility>
+#include <vector>
+
+#include "presence/implementation/advertisement_factory.h"
+
+namespace nearby {
+namespace presence {
+namespace {
+
+using ::location::nearby::api::ble_v2::BleOperationStatus;
+using AdvertisingCallback =
+    ::location::nearby::api::ble_v2::BleMedium::AdvertisingCallback;
+using AdvertisingSession =
+    ::location::nearby::api::ble_v2::BleMedium::AdvertisingSession;
+
+Status ConvertBleStatus(BleOperationStatus status) {
+  return status == BleOperationStatus::kSucceeded
+             ? Status{Status::Value::kSuccess}
+             : Status{Status::Value::kError};
+}
+
+}  // namespace
+
+absl::StatusOr<BroadcastSessionId> BroadcastManager::StartBroadcast(
+    BroadcastRequest broadcast_request, BroadcastCallback callback) {
+  absl::StatusOr<BaseBroadcastRequest> request =
+      BaseBroadcastRequest::Create(broadcast_request);
+  if (!request.ok()) {
+    NEARBY_LOGS(WARNING) << "Invalid broadcast request, reason: "
+                         << request.status();
+    callback.start_broadcast_cb(Status{Status::Value::kError});
+    return request.status();
+  }
+  BroadcastSessionId id = GenerateBroadcastSessionId();
+  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, power_mode)});
+            FetchCredentials(id, std::move(request));
+          });
+  return id;
+}
+
+void BroadcastManager::FetchCredentials(
+    BroadcastSessionId id, BaseBroadcastRequest broadcast_request) {
+  absl::StatusOr<CredentialSelector> credential_selector =
+      AdvertisementFactory::GetCredentialSelector(broadcast_request);
+  if (!credential_selector.ok()) {
+    // Public advertisement, we don't need credential to advertise.
+    Advertise(id, broadcast_request, /*credentials=*/{});
+    return;
+  }
+  credential_manager_->GetPrivateCredentials(
+      *credential_selector,
+      GetPrivateCredentialsResultCallback{
+          .credentials_fetched_cb =
+              [this, id, broadcast_request = std::move(broadcast_request)](
+                  std::vector<::nearby::internal::PrivateCredential>
+                      credentials) {
+                RunOnServiceControllerThread(
+                    "advertise-non-public",
+                    [this, id, broadcast_request = std::move(broadcast_request),
+                     credentials = std::move(credentials)]()
+                        ABSL_EXCLUSIVE_LOCKS_REQUIRED(executor_) {
+                          Advertise(id, broadcast_request, credentials);
+                        });
+              },
+          .get_credentials_failed_cb =
+              [this, id](absl::Status status) {
+                NEARBY_LOGS(WARNING)
+                    << "Failed to fetch credentials, status: " << status;
+                NotifyStartCallbackStatus(id, Status{Status::Value::kError});
+              }});
+}
+
+void BroadcastManager::Advertise(BroadcastSessionId id,
+                                 BaseBroadcastRequest broadcast_request,
+                                 std::vector<PrivateCredential> credentials) {
+  auto it = sessions_.find(id);
+  if (it == sessions_.end()) {
+    NEARBY_LOGS(INFO) << "Broadcast session terminated, id: " << id;
+    return;
+  }
+  absl::StatusOr<AdvertisementData> advertisement =
+      AdvertisementFactory().CreateAdvertisement(broadcast_request,
+                                                 credentials);
+  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, it->second.GetPowerMode(),
+          AdvertisingCallback{.start_advertising_result =
+                                  [this, id](BleOperationStatus status) {
+                                    NotifyStartCallbackStatus(
+                                        id, ConvertBleStatus(status));
+                                  }});
+  if (!session) {
+    NotifyStartCallbackStatus(id, Status{Status::Value::kError});
+    return;
+  }
+  it->second.SetAdvertisingSession(std::move(session));
+}
+
+void BroadcastManager::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 BroadcastManager::StopBroadcast(BroadcastSessionId 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 BroadcastManager::GenerateBroadcastSessionId() {
+  return absl::Uniform<BroadcastSessionId>(bit_gen_);
+}
+
+void BroadcastManager::BroadcastSessionState::SetAdvertisingSession(
+    std::unique_ptr<AdvertisingSession> session) {
+  advertising_session_ = std::move(session);
+}
+
+void BroadcastManager::BroadcastSessionState::CallStartedCallback(
+    Status status) {
+  BroadcastCallback callback = std::move(broadcast_callback_);
+  if (callback.start_broadcast_cb) {
+    callback.start_broadcast_cb(status);
+  }
+}
+
+void BroadcastManager::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/broadcast_manager.h

@@ -15,26 +15,81 @@
 #ifndef THIRD_PARTY_NEARBY_PRESENCE_IMPLEMENTATION_BROADCAST_MANAGER_H_
 #define THIRD_PARTY_NEARBY_PRESENCE_IMPLEMENTATION_BROADCAST_MANAGER_H_
 
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "absl/base/thread_annotations.h"
+#include "absl/random/random.h"
+#include "absl/strings/string_view.h"
+#include "internal/platform/single_thread_executor.h"
+#include "internal/proto/credential.pb.h"
+#include "presence/broadcast_request.h"
+#include "presence/data_types.h"
+#include "presence/implementation/base_broadcast_request.h"
 #include "presence/implementation/credential_manager.h"
 #include "presence/implementation/mediums/mediums.h"
 
 namespace nearby {
 namespace presence {
 
-/*
- * The instance of BroadcastManager is owned by {@code ServiceControllerImpl}.
- * Helping service controller to manage broadcast requests and callbacks.
- */
+// The instance of BroadcastManager is owned by {@code ServiceControllerImpl}.
+// Helping service controller to manage broadcast requests and callbacks.
+
 class BroadcastManager {
  public:
-  BroadcastManager(Mediums& mediums, CredentialManager& credential_manager) {
-    mediums_ = &mediums, credential_manager_ = &credential_manager;
+  using SingleThreadExecutor = ::location::nearby::SingleThreadExecutor;
+  using AdvertisingSession =
+      ::location::nearby::api::ble_v2::BleMedium::AdvertisingSession;
+  using Runnable = ::location::nearby::Runnable;
+  using PrivateCredential = internal::PrivateCredential;
+  BroadcastManager(Mediums& mediums, CredentialManager& credential_manager,
+                   SingleThreadExecutor& executor) {
+    mediums_ = &mediums, credential_manager_ = &credential_manager,
+    executor_ = &executor;
   }
   ~BroadcastManager() = default;
+  absl::StatusOr<BroadcastSessionId> StartBroadcast(
+      BroadcastRequest broadcast_request, BroadcastCallback callback);
+  void StopBroadcast(BroadcastSessionId);
 
  private:
   Mediums* mediums_;
   CredentialManager* credential_manager_;
+  SingleThreadExecutor* executor_;
+  class BroadcastSessionState {
+   public:
+    explicit BroadcastSessionState(BroadcastCallback broadcast_callback,
+                                   PowerMode power_mode)
+        : broadcast_callback_(broadcast_callback), power_mode_(power_mode) {}
+
+    void SetAdvertisingSession(std::unique_ptr<AdvertisingSession> session);
+    void CallStartedCallback(Status status);
+    void StopAdvertising();
+
+    PowerMode GetPowerMode() { return power_mode_; }
+
+   private:
+    BroadcastCallback broadcast_callback_;
+    PowerMode power_mode_;
+    std::unique_ptr<AdvertisingSession> advertising_session_;
+  };
+  BroadcastSessionId GenerateBroadcastSessionId();
+  void NotifyStartCallbackStatus(BroadcastSessionId id, Status status);
+  void RunOnServiceControllerThread(absl::string_view name, Runnable runnable) {
+    executor_->Execute(std::string(name), std::move(runnable));
+  }
+  void FetchCredentials(BroadcastSessionId id,
+                        BaseBroadcastRequest broadcast_request)
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(*executor_);
+
+  void Advertise(BroadcastSessionId id, BaseBroadcastRequest broadcast_request,
+                 std::vector<PrivateCredential> credentials)
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(*executor_);
+  absl::flat_hash_map<BroadcastSessionId, BroadcastSessionState> sessions_
+      ABSL_GUARDED_BY(*executor_);
+  absl::BitGen bit_gen_;
 };
 
 }  // namespace presence

presence/implementation/broadcast_manager_test.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "presence/implementation/service_controller_impl.h"
+#include "presence/implementation/broadcast_manager.h"
 
 #include <memory>
 #include <string>
@@ -25,6 +25,8 @@
 #include "internal/platform/future.h"
 #include "internal/platform/medium_environment.h"
 #include "internal/proto/credential.pb.h"
+#include "presence/implementation/credential_manager_impl.h"
+#include "presence/implementation/mediums/mediums.h"
 
 namespace nearby {
 namespace presence {
@@ -61,15 +63,14 @@ class MediumEnvironmentStarter {
   ~MediumEnvironmentStarter() { MediumEnvironment::Instance().Stop(); }
 };
 
-class ServiceControllerImplTest : public testing::TestWithParam<FeatureFlags> {
+class BroadcastManagerTest : 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())
+        .GetBleV2MediumStatus(*mediums_.GetBle().GetImpl())
         ->is_advertising;
   }
   BroadcastCallback CreateBroadcastCallback() {
@@ -80,8 +81,7 @@ class ServiceControllerImplTest : public testing::TestWithParam<FeatureFlags> {
 
   void WaitForServiceControllerTasks() {
     CountDownLatch latch(1);
-    service_controller_.GetBackgroundExecutor().Execute(
-        [&]() { latch.CountDown(); });
+    executor_.Execute([&]() { latch.CountDown(); });
     latch.Await();
   }
 
@@ -93,16 +93,18 @@ class ServiceControllerImplTest : public testing::TestWithParam<FeatureFlags> {
       .start_broadcast_cb = [this](Status status) {
         start_broadcast_status_.Set(status);
       }};
-  ServiceControllerImpl service_controller_;
+  Mediums mediums_;
+  CredentialManagerImpl credential_manager_;
+  location::nearby::SingleThreadExecutor executor_;
+  BroadcastManager broadcast_manager_{mediums_, credential_manager_, executor_};
 };
 
-INSTANTIATE_TEST_SUITE_P(ParametrisedServiceControllerImplTest,
-                         ServiceControllerImplTest,
+INSTANTIATE_TEST_SUITE_P(ParametrisedBroadcastManagerTest, BroadcastManagerTest,
                          testing::ValuesIn(kTestCases));
 
-TEST_P(ServiceControllerImplTest, StartBroadcastPublicIdentity) {
+TEST_P(BroadcastManagerTest, StartBroadcastPublicIdentity) {
   absl::StatusOr<BroadcastSessionId> session =
-      service_controller_.StartBroadcast(
+      broadcast_manager_.StartBroadcast(
           CreateBroadcastRequest(internal::IDENTITY_TYPE_PUBLIC),
           CreateBroadcastCallback());
 
@@ -113,38 +115,38 @@ TEST_P(ServiceControllerImplTest, StartBroadcastPublicIdentity) {
   EXPECT_TRUE(IsAdvertising());
 }
 
-TEST_P(ServiceControllerImplTest, StartAndStopBroadcast) {
+TEST_P(BroadcastManagerTest, StartAndStopBroadcast) {
   absl::StatusOr<BroadcastSessionId> session =
-      service_controller_.StartBroadcast(
+      broadcast_manager_.StartBroadcast(
           CreateBroadcastRequest(internal::IDENTITY_TYPE_PUBLIC),
           CreateBroadcastCallback());
   ASSERT_OK(session);
   EXPECT_TRUE(IsAdvertising());
 
-  service_controller_.StopBroadcast(*session);
+  broadcast_manager_.StopBroadcast(*session);
   EXPECT_FALSE(IsAdvertising());
 }
 
-TEST_P(ServiceControllerImplTest, StopBroadcastTwiceNoSideEffects) {
+TEST_P(BroadcastManagerTest, StopBroadcastTwiceNoSideEffects) {
   absl::StatusOr<BroadcastSessionId> session =
-      service_controller_.StartBroadcast(
+      broadcast_manager_.StartBroadcast(
           CreateBroadcastRequest(internal::IDENTITY_TYPE_PUBLIC),
           CreateBroadcastCallback());
   ASSERT_OK(session);
   EXPECT_TRUE(IsAdvertising());
 
-  service_controller_.StopBroadcast(*session);
-  service_controller_.StopBroadcast(*session);
+  broadcast_manager_.StopBroadcast(*session);
+  broadcast_manager_.StopBroadcast(*session);
 }
 
-TEST_P(ServiceControllerImplTest, StopBroadcastInvalidSessionNoSideEffects) {
-  service_controller_.StopBroadcast(123456);
+TEST_P(BroadcastManagerTest, StopBroadcastInvalidSessionNoSideEffects) {
+  broadcast_manager_.StopBroadcast(123456);
 }
 
-TEST_P(ServiceControllerImplTest, StartBroadcastInvalidRequestFails) {
+TEST_P(BroadcastManagerTest, StartBroadcastInvalidRequestFails) {
   absl::StatusOr<BroadcastSessionId> session =
-      service_controller_.StartBroadcast(BroadcastRequest{},
-                                         CreateBroadcastCallback());
+      broadcast_manager_.StartBroadcast(BroadcastRequest{},
+                                        CreateBroadcastCallback());
 
   EXPECT_THAT(session, StatusIs(absl::StatusCode::kInvalidArgument));
   EXPECT_TRUE(start_broadcast_status_.Get().ok());
@@ -153,10 +155,10 @@ TEST_P(ServiceControllerImplTest, StartBroadcastInvalidRequestFails) {
   EXPECT_FALSE(IsAdvertising());
 }
 
-TEST_P(ServiceControllerImplTest, StartBroadcastPrivateIdentityFails) {
+TEST_P(BroadcastManagerTest, StartBroadcastPrivateIdentityFails) {
   // TODO(b/256249404): Support private identity.
   absl::StatusOr<BroadcastSessionId> session =
-      service_controller_.StartBroadcast(
+      broadcast_manager_.StartBroadcast(
           CreateBroadcastRequest(internal::IDENTITY_TYPE_PRIVATE),
           CreateBroadcastCallback());
 

presence/implementation/service_controller_impl.cc

@@ -14,38 +14,11 @@
 
 #include "presence/implementation/service_controller_impl.h"
 
-#include <memory>
-#include <string>
-#include <utility>
-#include <vector>
-
-#include "absl/random/random.h"
-#include "absl/status/status.h"
 #include "absl/status/statusor.h"
-#include "internal/platform/implementation/ble_v2.h"
 #include "presence/data_types.h"
-#include "presence/implementation/advertisement_factory.h"
-#include "presence/implementation/base_broadcast_request.h"
-#include "presence/implementation/mediums/advertisement_data.h"
-#include "presence/status.h"
 
 namespace nearby {
 namespace presence {
-namespace {
-
-using ::location::nearby::api::ble_v2::BleOperationStatus;
-using AdvertisingCallback =
-    ::location::nearby::api::ble_v2::BleMedium::AdvertisingCallback;
-using AdvertisingSession =
-    ::location::nearby::api::ble_v2::BleMedium::AdvertisingSession;
-
-Status ConvertBleStatus(BleOperationStatus status) {
-  return status == BleOperationStatus::kSucceeded
-             ? Status{Status::Value::kSuccess}
-             : Status{Status::Value::kError};
-}
-
-}  // namespace
 
 absl::StatusOr<ScanSessionId> ServiceControllerImpl::StartScan(
     ScanRequest scan_request, ScanCallback callback) {
@@ -57,146 +30,11 @@ void ServiceControllerImpl::StopScan(ScanSessionId id) {
 
 absl::StatusOr<BroadcastSessionId> ServiceControllerImpl::StartBroadcast(
     BroadcastRequest broadcast_request, BroadcastCallback callback) {
-  absl::StatusOr<BaseBroadcastRequest> request =
-      BaseBroadcastRequest::Create(broadcast_request);
-  if (!request.ok()) {
-    NEARBY_LOGS(WARNING) << "Invalid broadcast request, reason: "
-                         << request.status();
-    callback.start_broadcast_cb(Status{Status::Value::kError});
-    return request.status();
-  }
-  BroadcastSessionId id = GenerateBroadcastSessionId();
-  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, power_mode)});
-            FetchCredentials(id, std::move(request));
-          });
-
-  return id;
-}
-
-void ServiceControllerImpl::FetchCredentials(
-    BroadcastSessionId id, BaseBroadcastRequest broadcast_request) {
-  absl::StatusOr<CredentialSelector> credential_selector =
-      AdvertisementFactory::GetCredentialSelector(broadcast_request);
-  if (!credential_selector.ok()) {
-    // Public advertisement, we don't need credential to advertise.
-    Advertise(id, broadcast_request, /*credentials=*/{});
-    return;
-  }
-  credential_manager_.GetPrivateCredentials(
-      *credential_selector,
-      GetPrivateCredentialsResultCallback{
-          .credentials_fetched_cb =
-              [this, id, broadcast_request = std::move(broadcast_request)](
-                  std::vector<::nearby::internal::PrivateCredential>
-                      credentials) {
-                RunOnServiceControllerThread(
-                    "advertise-non-public",
-                    [this, id, broadcast_request = std::move(broadcast_request),
-                     credentials = std::move(credentials)]()
-                        ABSL_EXCLUSIVE_LOCKS_REQUIRED(executor_) {
-                          Advertise(id, broadcast_request, credentials);
-                        });
-              },
-          .get_credentials_failed_cb =
-              [this, id](absl::Status status) {
-                NEARBY_LOGS(WARNING)
-                    << "Failed to fetch credentials, status: " << status;
-                NotifyStartCallbackStatus(id, Status{Status::Value::kError});
-              }});
-}
-
-void ServiceControllerImpl::Advertise(
-    BroadcastSessionId id, BaseBroadcastRequest broadcast_request,
-    std::vector<PrivateCredential> credentials) {
-  auto it = sessions_.find(id);
-  if (it == sessions_.end()) {
-    NEARBY_LOGS(INFO) << "Broadcast session terminated, id: " << id;
-    return;
-  }
-  absl::StatusOr<AdvertisementData> advertisement =
-      AdvertisementFactory().CreateAdvertisement(broadcast_request,
-                                                 credentials);
-  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, it->second.GetPowerMode(),
-          AdvertisingCallback{.start_advertising_result =
-                                  [this, id](BleOperationStatus status) {
-                                    NotifyStartCallbackStatus(
-                                        id, ConvertBleStatus(status));
-                                  }});
-  if (!session) {
-    NotifyStartCallbackStatus(id, Status{Status::Value::kError});
-    return;
-  }
-  it->second.SetAdvertisingSession(std::move(session));
-}
-
-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);
-                                     }
-                                   });
+  return broadcast_manager_.StartBroadcast(broadcast_request, callback);
 }
 
 void ServiceControllerImpl::StopBroadcast(BroadcastSessionId 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();
-  }
+  broadcast_manager_.StopBroadcast(id);
 }
 
 }  // namespace presence

presence/implementation/service_controller_impl.h

@@ -19,15 +19,8 @@
 #include <string>
 #include <utility>
 
-#include "absl/base/thread_annotations.h"
-#include "absl/random/random.h"
 #include "absl/status/statusor.h"
-#include "absl/strings/string_view.h"
-#include "internal/platform/single_thread_executor.h"
-#include "internal/proto/credential.pb.h"
-#include "presence/broadcast_request.h"
-#include "presence/data_types.h"
-#include "presence/implementation/base_broadcast_request.h"
+#include "presence/implementation/broadcast_manager.h"
 #include "presence/implementation/credential_manager_impl.h"
 #include "presence/implementation/mediums/mediums.h"
 #include "presence/implementation/scan_manager.h"
@@ -44,10 +37,6 @@ namespace presence {
 class ServiceControllerImpl : public ServiceController {
  public:
   using SingleThreadExecutor = ::location::nearby::SingleThreadExecutor;
-  using AdvertisingSession =
-      ::location::nearby::api::ble_v2::BleMedium::AdvertisingSession;
-  using Runnable = ::location::nearby::Runnable;
-  using PrivateCredential = internal::PrivateCredential;
 
   ServiceControllerImpl() = default;
   ~ServiceControllerImpl() override { executor_.Shutdown(); }
@@ -65,47 +54,11 @@ class ServiceControllerImpl : public ServiceController {
   Mediums& GetMediums() { return mediums_; }
 
  private:
-  class BroadcastSessionState {
-   public:
-    explicit BroadcastSessionState(BroadcastCallback broadcast_callback,
-                                   PowerMode power_mode)
-        : broadcast_callback_(broadcast_callback), power_mode_(power_mode) {}
-
-    void SetAdvertisingSession(std::unique_ptr<AdvertisingSession> session);
-
-    void CallStartedCallback(Status status);
-
-    void StopAdvertising();
-
-    PowerMode GetPowerMode() { return power_mode_; }
-
-   private:
-    BroadcastCallback broadcast_callback_;
-    PowerMode power_mode_;
-    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));
-  }
-  void FetchCredentials(BroadcastSessionId id,
-                        BaseBroadcastRequest broadcast_request)
-      ABSL_EXCLUSIVE_LOCKS_REQUIRED(executor_);
-
-  void Advertise(BroadcastSessionId id, BaseBroadcastRequest broadcast_request,
-                 std::vector<PrivateCredential> credentials)
-      ABSL_EXCLUSIVE_LOCKS_REQUIRED(executor_);
-
-  Mediums mediums_;  // NOLINT: further impl will use it.
-  CredentialManagerImpl
-      credential_manager_;  // NOLINT: further impl will use it.
-  ScanManager scan_manager_{mediums_, credential_manager_,
-                            executor_};  // NOLINT: further impl will use it.
-  absl::flat_hash_map<BroadcastSessionId, BroadcastSessionState> sessions_
-      ABSL_GUARDED_BY(executor_);
-  absl::BitGen bit_gen_;
+  Mediums mediums_;
+  CredentialManagerImpl credential_manager_;
+  ScanManager scan_manager_{mediums_, credential_manager_, executor_};
+  BroadcastManager broadcast_manager_{mediums_, credential_manager_, executor_};
 };
 
 }  // namespace presence