Skip to content

subscription_thread_dispatcher: guard against duplicate room events #195

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
stephen-derosa wants to merge 11 commits into
base: main
Choose a base branch
from
Open

subscription_thread_dispatcher: guard against duplicate room events #195

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
a04dc57
SubscriptionThreadDispatcher: guard against duplicate room events, du…
stephen-derosa Jul 6, 2026
f1ff6e6
more seperated test artifacts
stephen-derosa Jul 7, 2026
338f266
doc
stephen-derosa Jul 7, 2026
709a8ab
SubscriptionThreadDispatcher: proper replacing of audio/video callbac…
stephen-derosa Jul 7, 2026
20e5d5b
trim
stephen-derosa Jul 7, 2026
6053b27
fix room logic
stephen-derosa Jul 7, 2026
22991ec
log in old functions
stephen-derosa Jul 7, 2026
1de44c8
doc clarity
stephen-derosa Jul 7, 2026
4618ea9
better locking
stephen-derosa Jul 7, 2026
d3ce7a9
const
stephen-derosa Jul 7, 2026
ede1f20
fix thread detaching
stephen-derosa Jul 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,6 +260,13 @@ The following features are deprecated and will be removed in the next major rele

- `PacketTrailerFeatures` is deprecated. Use `FrameMetadataFeatures` via
`TrackPublishOptions::frame_metadata_features` instead.
- `Room::setOnAudioFrameCallback`, `Room::setOnVideoFrameCallback`, and
`Room::setOnVideoFrameEventCallback` are deprecated. Use the `[[nodiscard]]`
variants `trySetOnAudioFrameCallback`, `trySetOnVideoFrameCallback`, and
`trySetOnVideoFrameEventCallback` instead, which return `false` when a reader
is already active for the key (instead of silently replacing a running callback).
To replace an active callback, call `clearOn*FrameCallback` first. (The same rename
applies to the corresponding `SubscriptionThreadDispatcher` methods.)

### `v1.0.0`

Expand Down
10 changes: 4 additions & 6 deletions include/livekit/remote_data_track.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,11 +92,6 @@ class RemoteDataTrack {
/// @param options Pipeline options to apply to this remote data track.
LIVEKIT_API void setPipelineOptions(const DataTrackPipelineOptions& options);

#ifdef LIVEKIT_TEST_ACCESS
/// Test-only accessor for exercising lower-level FFI subscription paths.
uintptr_t testFfiHandleId() const noexcept { return ffiHandleId(); }
#endif

/// Subscribe to this remote data track.
///
/// Returns a DataTrackStream that delivers frames via blocking
Expand All @@ -106,8 +101,11 @@ class RemoteDataTrack {

private:
friend class Room;
#ifdef LIVEKIT_TEST_ACCESS

@alan-george-lk alan-george-lk Jul 8, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure if better or worse, but I typically use GTest's FRIEND_TEST macro so you can just give a single test suite/case access to what it needs. May break down a bit between unit and integration test binaries if they have different suite names (could unify name). Basically decays down to a friend class <blah> anyways, just looks a bit cleaner: https://google.github.io/googletest/reference/testing.html#FRIEND_TEST

friend struct RemoteDataTrackTestAccess;
#endif

explicit RemoteDataTrack(const proto::OwnedRemoteDataTrack& owned);
LIVEKIT_INTERNAL_API explicit RemoteDataTrack(const proto::OwnedRemoteDataTrack& owned);

uintptr_t ffiHandleId() const noexcept { return handle_.get(); }
/// RAII wrapper for the Rust-owned FFI resource.
Expand Down
90 changes: 86 additions & 4 deletions include/livekit/room.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -303,16 +303,92 @@ class LIVEKIT_API Room {
// Frame callbacks
// ---------------------------------------------------------------

/// @brief Sets the audio frame callback via SubscriptionThreadDispatcher.
/// Register an audio frame callback for a remote subscription.
///
/// The callback is keyed by @p participant_identity and @p track_name. If the
/// matching remote audio track is already subscribed, a reader is started
/// immediately; otherwise the reader starts when the track is subscribed.
///
/// To replace a callback whose reader is already running, call
/// @ref clearOnAudioFrameCallback first, then register again:
/// @code
/// room.clearOnAudioFrameCallback(identity, track_name);
/// if (!room.trySetOnAudioFrameCallback(identity, track_name, new_handler)) {
/// // registration was rejected (a reader is still active)
/// }
/// @endcode
///
/// @param participant_identity Identity of the remote participant.
/// @param track_name Track name to match.
/// @param callback Function invoked for each decoded audio frame.
/// @param opts Options used when creating the backing
/// @ref AudioStream.
/// @return @c true if the callback was registered; @c false if a reader is
/// already active for the key (call @ref clearOnAudioFrameCallback
/// first) or the room has no dispatcher.
[[nodiscard]] bool trySetOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name,

@alan-george-lk alan-george-lk Jul 8, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

From what I can tell, these new "try" methods have the same inputs as the deprecated prior ones, just a new return type (and name).

This is actually a situation where I think maintaining the API and using exceptions is cleaner than deprecation/new API just to get the bool return type.

Thoughts on this? Could we not move the new implementations from room.cpp into the older versions and throw instead of returning false?

@stephen-derosa stephen-derosa Jul 8, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

isnt that still changing the public API though? maybe im wrong but i would think adding a throw to a function that previously didnt would require a major bump?

@alan-george-lk alan-george-lk Jul 8, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fair point, technically speaking those methods were never marked noexcept and anything within them could throw (STL container exceptions, runtime exceptions from FFI stuff, etc.) so I think it's the least of the evils?

AudioFrameCallback callback, const AudioStream::Options& opts = {});

/// Register a video frame callback for a remote subscription.
///
/// The callback is keyed by @p participant_identity and @p track_name. If the
/// matching remote video track is already subscribed, a reader is started
/// immediately; otherwise the reader starts when the track is subscribed.
///
/// To replace a callback whose reader is already running, call
/// @ref clearOnVideoFrameCallback first, then register again.
///
/// @param participant_identity Identity of the remote participant.
/// @param track_name Track name to match.
/// @param callback Function invoked for each decoded video frame.
/// @param opts Options used when creating the backing
/// @ref VideoStream.
/// @return @c true if the callback was registered; @c false if a reader is
/// already active for the key (call @ref clearOnVideoFrameCallback
/// first) or the room has no dispatcher.
[[nodiscard]] bool trySetOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name,

@xianshijing-lk xianshijing-lk Jul 9, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess I didn't understand why these callback events need to be in the room ?
couldn't they be in the remote participants?

And does this frame callbacks work on both local / remote participants ?

VideoFrameCallback callback, const VideoStream::Options& opts = {});

/// Register a rich video frame event callback for a remote subscription.
///
/// The callback is keyed by @p participant_identity and @p track_name. If the
/// matching remote video track is already subscribed, a reader is started
/// immediately; otherwise the reader starts when the track is subscribed.
///
/// To replace a callback whose reader is already running, call
/// @ref clearOnVideoFrameCallback first, then register again.
///
/// @param participant_identity Identity of the remote participant.
/// @param track_name Track name to match.
/// @param callback Function invoked for each decoded video frame
/// event, including optional metadata.
/// @param opts Options used when creating the backing
/// @ref VideoStream.
/// @return @c true if the callback was registered; @c false if a reader is
/// already active for the key (call @ref clearOnVideoFrameCallback
/// first) or the room has no dispatcher.
[[nodiscard]] bool trySetOnVideoFrameEventCallback(const std::string& participant_identity,
const std::string& track_name, VideoFrameEventCallback callback,
const VideoStream::Options& opts = {});

/// @deprecated Use trySetOnAudioFrameCallback() instead.
///
/// Forwards to @ref trySetOnAudioFrameCallback and discards the result.
[[deprecated("Room::setOnAudioFrameCallback is deprecated; use trySetOnAudioFrameCallback instead")]]
void setOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name,
AudioFrameCallback callback, const AudioStream::Options& opts = {});

/// @brief Sets the video frame callback via SubscriptionThreadDispatcher.
/// @deprecated Use trySetOnVideoFrameCallback() instead.
///
/// Forwards to @ref trySetOnVideoFrameCallback and discards the result.
[[deprecated("Room::setOnVideoFrameCallback is deprecated; use trySetOnVideoFrameCallback instead")]]
void setOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name,
VideoFrameCallback callback, const VideoStream::Options& opts = {});

/// @brief Sets the video frame event callback via
/// SubscriptionThreadDispatcher.
/// @deprecated Use trySetOnVideoFrameEventCallback() instead.
///
/// Forwards to @ref trySetOnVideoFrameEventCallback and discards the result.
[[deprecated("Room::setOnVideoFrameEventCallback is deprecated; use trySetOnVideoFrameEventCallback instead")]]
void setOnVideoFrameEventCallback(const std::string& participant_identity, const std::string& track_name,
VideoFrameEventCallback callback, const VideoStream::Options& opts = {});

Expand Down Expand Up @@ -354,6 +430,12 @@ class LIVEKIT_API Room {
// FfiClient listener ID (0 means no listener registered)
int listener_id_{0};

/// Find a currently subscribed remote track matching the given participant
/// identity and track name. Returns nullptr if no such subscribed track
/// exists. Acquires @ref lock_.
std::shared_ptr<Track> findSubscribedRemoteTrack(const std::string& participant_identity,
const std::string& track_name) const;

void onEvent(const proto::FfiEvent& event);
};
} // namespace livekit
Loading
Loading