subscription_thread_dispatcher for handling creating threads which execute user set callbacks

Author: stephen-derosa

CMakeLists.txt

@@ -335,6 +335,7 @@ add_library(livekit SHARED
   src/room.cpp
   src/room_proto_converter.cpp
   src/room_proto_converter.h
+  src/subscription_thread_dispatcher.cpp
   src/local_participant.cpp
   src/remote_participant.cpp
   src/stats.cpp
@@ -723,4 +724,4 @@ add_custom_target(clean_all
   COMMAND ${CMAKE_COMMAND} -E chdir "${CMAKE_SOURCE_DIR}" ${CMAKE_COMMAND} -E rm -rf "${CMAKE_BINARY_DIR}"
   COMMENT "Full clean: CMake outputs + Rust target + generated protos + delete build/"
   VERBATIM
-)
+)

include/livekit/room.h

@@ -17,38 +17,24 @@
 #ifndef LIVEKIT_ROOM_H
 #define LIVEKIT_ROOM_H
 
-#include "livekit/audio_stream.h"
 #include "livekit/data_stream.h"
 #include "livekit/e2ee.h"
 #include "livekit/ffi_handle.h"
 #include "livekit/room_event_types.h"
-#include "livekit/video_stream.h"
+#include "livekit/subscription_thread_dispatcher.h"
 
 #include <cstdint>
-#include <functional>
 #include <memory>
 #include <mutex>
-#include <thread>
 
 namespace livekit {
 
-class AudioFrame;
-class VideoFrame;
 class RoomDelegate;
 struct RoomInfoData;
 namespace proto {
 class FfiEvent;
 }
 
-/// Callback type for incoming audio frames.
-/// Invoked on a dedicated reader thread per (participant, source) pair.
-using AudioFrameCallback = std::function<void(const AudioFrame &)>;
-
-/// Callback type for incoming video frames.
-/// Invoked on a dedicated reader thread per (participant, source) pair.
-using VideoFrameCallback =
-    std::function<void(const VideoFrame &frame, std::int64_t timestamp_us)>;
-
 struct E2EEOptions;
 class E2EEManager;
 class LocalParticipant;
@@ -311,8 +297,6 @@ class Room {
                                  TrackSource source);
 
 private:
-  friend class RoomCallbackTest;
-
   mutable std::mutex lock_;
   ConnectionState connection_state_ = ConnectionState::Disconnected;
   RoomDelegate *delegate_ = nullptr; // Not owned
@@ -330,75 +314,12 @@ class Room {
       byte_stream_readers_;
   // E2EE
   std::unique_ptr<E2EEManager> e2ee_manager_;
+  std::shared_ptr<SubscriptionThreadDispatcher> subscription_thread_dispatcher_;
 
   // FfiClient listener ID (0 means no listener registered)
   int listener_id_{0};
 
   void OnEvent(const proto::FfiEvent &event);
-
-  // -------------------------------------------------------------------
-  // Frame callback internals
-  // -------------------------------------------------------------------
-
-  struct CallbackKey {
-    std::string participant_identity;
-    TrackSource source;
-    bool operator==(const CallbackKey &o) const {
-      return participant_identity == o.participant_identity &&
-             source == o.source;
-    }
-  };
-
-  struct CallbackKeyHash {
-    std::size_t operator()(const CallbackKey &k) const {
-      auto h1 = std::hash<std::string>{}(k.participant_identity);
-      auto h2 = std::hash<int>{}(static_cast<int>(k.source));
-      return h1 ^ (h2 << 1);
-    }
-  };
-
-  struct ActiveReader {
-    std::shared_ptr<AudioStream> audio_stream;
-    std::shared_ptr<VideoStream> video_stream;
-    std::thread thread;
-  };
-
-  struct RegisteredAudioCallback {
-    AudioFrameCallback callback;
-    AudioStream::Options options;
-  };
-
-  struct RegisteredVideoCallback {
-    VideoFrameCallback callback;
-    VideoStream::Options options;
-  };
-
-  std::unordered_map<CallbackKey, RegisteredAudioCallback, CallbackKeyHash>
-      audio_callbacks_;
-  std::unordered_map<CallbackKey, RegisteredVideoCallback, CallbackKeyHash>
-      video_callbacks_;
-  std::unordered_map<CallbackKey, ActiveReader, CallbackKeyHash>
-      active_readers_;
-
-  static constexpr int kMaxActiveReaders = 20;
-
-  // Must be called with lock_ held. Closes the stream for the given key and
-  // returns the old reader thread (which the caller must join outside the
-  // lock).
-  std::thread extractReaderThread(const CallbackKey &key);
-
-  // Must be called with lock_ held. Returns old reader thread if one existed.
-  std::thread startAudioReader(const CallbackKey &key,
-                               const std::shared_ptr<Track> &track,
-                               AudioFrameCallback cb,
-                               const AudioStream::Options &opts);
-  std::thread startVideoReader(const CallbackKey &key,
-                               const std::shared_ptr<Track> &track,
-                               VideoFrameCallback cb,
-                               const VideoStream::Options &opts);
-
-  // Stops all readers (closes streams, joins threads). Must NOT hold lock_.
-  void stopAllReaders();
 };
 } // namespace livekit