move functions to test

Author: stephen-derosa

include/livekit/subscription_thread_dispatcher.h

@@ -234,14 +234,6 @@ class LIVEKIT_API SubscriptionThreadDispatcher {
   /// thread survives beyond the lifetime of the owning @ref Room.
   void stopAll();
 
-  /// @brief Number of active audio/video reader threads. Intended for tests.
-  /// @return Count of currently active audio and video reader slots.
-  std::size_t activeReaderCount() const;
-
-  /// @brief Number of active data track reader threads. Intended for tests.
-  /// @return Count of currently active data reader slots.
-  std::size_t activeDataReaderCount() const;
-
 private:
   friend class SubscriptionThreadDispatcherTest;
 

src/subscription_thread_dispatcher.cpp

@@ -339,16 +339,6 @@ void SubscriptionThreadDispatcher::stopAll() {
   LK_LOG_DEBUG("Stopped {} subscription reader threads", threads.size());
 }
 
-std::size_t SubscriptionThreadDispatcher::activeReaderCount() const {
-  const std::scoped_lock<std::mutex> lock(lock_);
-  return active_readers_.size();
-}
-
-std::size_t SubscriptionThreadDispatcher::activeDataReaderCount() const {
-  const std::scoped_lock<std::mutex> lock(lock_);
-  return active_data_readers_.size();
-}
-
 std::thread SubscriptionThreadDispatcher::extractReaderThreadLocked(const CallbackKey& key) {
   auto it = active_readers_.find(key);
   if (it == active_readers_.end()) {

src/tests/unit/test_subscription_thread_dispatcher.cpp

@@ -79,6 +79,14 @@ class SubscriptionThreadDispatcherTest : public ::testing::Test {
   static auto& activeDataReaders(SubscriptionThreadDispatcher& dispatcher) { return dispatcher.active_data_readers_; }
   static auto& remoteDataTracks(SubscriptionThreadDispatcher& dispatcher) { return dispatcher.remote_data_tracks_; }
   static int maxActiveReaders() { return SubscriptionThreadDispatcher::kMaxActiveReaders; }
+  static std::size_t activeReaderCount(SubscriptionThreadDispatcher& dispatcher) {
+    const std::scoped_lock<std::mutex> lock(dispatcher.lock_);
+    return dispatcher.active_readers_.size();
+  }
+  static std::size_t activeDataReaderCount(SubscriptionThreadDispatcher& dispatcher) {
+    const std::scoped_lock<std::mutex> lock(dispatcher.lock_);
+    return dispatcher.active_data_readers_.size();
+  }
 
   static std::thread extractDataReader(SubscriptionThreadDispatcher& dispatcher, DataFrameCallbackId id) {
     const std::scoped_lock<std::mutex> lock(dispatcher.lock_);
@@ -592,14 +600,14 @@ TEST_F(SubscriptionThreadDispatcherTest, DuplicateSubscribeWithSameAudioSidDoesN
   // Simulate an already-running reader for this subscription.
   const CallbackKey key{"alice", "mic"};
   activeReaders(dispatcher)[key].track_sid = "TR_audio_1";
-  ASSERT_EQ(dispatcher.activeReaderCount(), 1u);
+  ASSERT_EQ(activeReaderCount(dispatcher), 1u);
 
   // A duplicate track_subscribed carrying the same SID must be a no-op: no
   // extract, no new stream/thread.
   auto track = std::make_shared<FakeMediaTrack>("TR_audio_1", TrackKind::KIND_AUDIO);
   dispatcher.handleTrackSubscribed("alice", "mic", track);
 
-  EXPECT_EQ(dispatcher.activeReaderCount(), 1u);
+  EXPECT_EQ(activeReaderCount(dispatcher), 1u);
   EXPECT_EQ(activeReaders(dispatcher)[key].track_sid, "TR_audio_1");
   EXPECT_EQ(activeReaders(dispatcher)[key].audio_stream, nullptr) << "Reader must not have been rebuilt";
 }
@@ -610,12 +618,12 @@ TEST_F(SubscriptionThreadDispatcherTest, DuplicateSubscribeWithSameVideoSidDoesN
 
   const CallbackKey key{"alice", "cam"};
   activeReaders(dispatcher)[key].track_sid = "TR_video_1";
-  ASSERT_EQ(dispatcher.activeReaderCount(), 1u);
+  ASSERT_EQ(activeReaderCount(dispatcher), 1u);
 
   auto track = std::make_shared<FakeMediaTrack>("TR_video_1", TrackKind::KIND_VIDEO);
   dispatcher.handleTrackSubscribed("alice", "cam", track);
 
-  EXPECT_EQ(dispatcher.activeReaderCount(), 1u);
+  EXPECT_EQ(activeReaderCount(dispatcher), 1u);
   EXPECT_EQ(activeReaders(dispatcher)[key].track_sid, "TR_video_1");
   EXPECT_EQ(activeReaders(dispatcher)[key].video_stream, nullptr) << "Reader must not have been rebuilt";
 }
@@ -635,7 +643,7 @@ TEST_F(SubscriptionThreadDispatcherTest, DuplicateDataPublishWithSameSidDoesNotR
   auto old_thread = startDataReader(dispatcher, 7, DataCallbackKey{"alice", "foo"}, incoming);
 
   EXPECT_FALSE(old_thread.joinable());
-  EXPECT_EQ(dispatcher.activeDataReaderCount(), 1u);
+  EXPECT_EQ(activeDataReaderCount(dispatcher), 1u);
   EXPECT_EQ(activeDataReaders(dispatcher)[7], reader) << "Same-SID publish must not replace the reader";
   EXPECT_FALSE(reader->cancelled.load()) << "A skipped reader must not be cancelled";
 }
@@ -658,7 +666,7 @@ TEST_F(SubscriptionThreadDispatcherTest, RepublishWithNewDataSidStopsPreviousRea
 
   // The replacement reader has an invalid FFI handle, so its subscribe fails
   // fast and it removes its own slot. Confirm nothing is left behind.
-  EXPECT_TRUE(waitFor([&] { return dispatcher.activeDataReaderCount() == 0; }, 2s));
+  EXPECT_TRUE(waitFor([&] { return activeDataReaderCount(dispatcher) == 0; }, 2s));
 }
 
 // ============================================================================