From a04dc574f4622b3e54a903d53dcbe7e2fb99af14 Mon Sep 17 00:00:00 2001 From: Stephen DeRosa Date: Mon, 6 Jul 2026 15:37:36 -0600 Subject: [PATCH 01/11] SubscriptionThreadDispatcher: guard against duplicate room events, duplicate room events test --- include/livekit/remote_data_track.h | 10 +- .../livekit/subscription_thread_dispatcher.h | 63 ++- src/subscription_thread_dispatcher.cpp | 81 +++- src/tests/integration/test_data_track.cpp | 95 ++++ .../test_room_event_deduplication.cpp | 408 ++++++++++++++++++ .../test_subscription_thread_dispatcher.cpp | 191 ++++++++ 6 files changed, 803 insertions(+), 45 deletions(-) create mode 100644 src/tests/integration/test_room_event_deduplication.cpp diff --git a/include/livekit/remote_data_track.h b/include/livekit/remote_data_track.h index 196cde54..4dafe318 100644 --- a/include/livekit/remote_data_track.h +++ b/include/livekit/remote_data_track.h @@ -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 @@ -106,8 +101,11 @@ class RemoteDataTrack { private: friend class Room; +#ifdef LIVEKIT_TEST_ACCESS + 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. diff --git a/include/livekit/subscription_thread_dispatcher.h b/include/livekit/subscription_thread_dispatcher.h index 73c86684..2890b7bb 100644 --- a/include/livekit/subscription_thread_dispatcher.h +++ b/include/livekit/subscription_thread_dispatcher.h @@ -16,6 +16,8 @@ #pragma once +#include +#include #include #include #include @@ -65,13 +67,15 @@ using DataFrameCallbackId = std::uint64_t; /// /// `SubscriptionThreadDispatcher` is the low-level companion to @ref Room's /// remote track subscription flow. `Room` forwards user-facing callback -/// registration requests here, and then calls @ref handleTrackSubscribed and -/// @ref handleTrackUnsubscribed as room events arrive. +/// registration requests here. For remote audio and video subscriptions it +/// calls @ref handleTrackSubscribed and @ref handleTrackUnsubscribed; for +/// data tracks it calls @ref handleDataTrackPublished and +/// @ref handleDataTrackUnpublished. /// -/// For each registered `(participant identity, track name)` pair, this class -/// may create a dedicated @ref AudioStream or @ref VideoStream and a matching -/// reader thread. That thread blocks on stream reads and invokes the -/// registered callback with decoded frames. +/// For each registered audio or video `(participant identity, track name)` +/// pair, this class may create a dedicated @ref AudioStream or @ref +/// VideoStream and a matching reader thread. That thread blocks on stream +/// reads and invokes the registered callback with decoded frames. /// /// This type is intentionally independent from @ref RoomDelegate. High-level /// room events such as `RoomDelegate::onTrackSubscribed()` remain in @ref Room, @@ -151,27 +155,33 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// @param track_name Track name to clear. void clearOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name); - /// Start or restart reader dispatch for a newly subscribed remote track. + /// Start or restart reader dispatch for a newly subscribed remote audio or + /// video track. /// /// @ref Room calls this after it has processed a track-subscription event and - /// updated its publication state. If a matching callback registration exists, - /// the dispatcher creates the appropriate stream type and launches a reader - /// thread for the `(participant, track_name)` key. + /// updated its publication state. If a matching audio or video callback + /// registration exists, the dispatcher creates the appropriate @ref + /// AudioStream or @ref VideoStream and launches a reader thread for the + /// `(participant, track_name)` key. /// - /// If no matching callback is registered, this is a no-op. + /// Remote data tracks are handled separately via @ref handleDataTrackPublished. + /// If @p track is not audio or video, or no matching callback is registered, + /// this is a no-op. /// /// @param participant_identity Identity of the remote participant. /// @param track_name Track name associated with the subscription. - /// @param track Subscribed remote track to read from. + /// @param track Subscribed remote audio or video track to read from. void handleTrackSubscribed(const std::string& participant_identity, const std::string& track_name, const std::shared_ptr& track); - /// Stop reader dispatch for an unsubscribed remote track. + /// Stop reader dispatch for an unsubscribed remote audio or video track. /// - /// @ref Room calls this when a remote track is unsubscribed. Any active - /// reader stream for the given `(participant, track_name)` key is closed and its - /// thread is joined. Callback registration is preserved so future - /// re-subscription can start dispatch again automatically. + /// @ref Room calls this when a remote audio or video track is unsubscribed. + /// Any active reader stream for the given `(participant, track_name)` key is + /// closed and its thread is joined. Callback registration is preserved so + /// future re-subscription can start dispatch again automatically. + /// + /// Remote data tracks are handled separately via @ref handleDataTrackUnpublished. /// /// @param participant_identity Identity of the remote participant. /// @param source Track source associated with the subscription. @@ -259,6 +269,9 @@ class LIVEKIT_API SubscriptionThreadDispatcher { std::shared_ptr audio_stream; std::shared_ptr video_stream; std::thread thread; + /// SID of the subscribed track backing this reader, used to skip redundant + /// reader restarts when the same publication is re-subscribed. + std::string track_sid; }; /// Compound lookup key for a remote participant identity and data track name. @@ -289,6 +302,9 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// Active read-side resources for one data track stream subscription. struct ActiveDataReader { std::shared_ptr remote_track; + /// Set true when this reader is being replaced or torn down so the reader + /// thread can abort a subscription that is still in flight. + std::atomic cancelled{false}; std::mutex sub_mutex; std::shared_ptr stream; // guarded by sub_mutex std::thread thread; @@ -333,18 +349,21 @@ class LIVEKIT_API SubscriptionThreadDispatcher { const RegisteredVideoCallback& callback); /// Extract and close the data reader for a given callback ID, returning its - /// thread. Must be called with @ref lock_ held. + /// thread. Marks the reader cancelled so a subscription still in flight is + /// aborted. Must be called with @ref lock_ held. std::thread extractDataReaderThreadLocked(DataFrameCallbackId id); - /// Extract and close the data reader for a given (participant, track_name) - /// key, returning its thread. Must be called with @ref lock_ held. - std::thread extractDataReaderThreadLocked(const DataCallbackKey& key); - /// Start a data reader thread for the given callback ID, key, and track. /// Must be called with @ref lock_ held. std::thread startDataReaderLocked(DataFrameCallbackId id, const DataCallbackKey& key, const std::shared_ptr& track, const DataFrameCallback& cb); + /// Remove @p reader from @ref active_data_readers_ if the slot for @p id + /// still refers to it. Called by the reader thread itself when it exits + /// after a failed or cancelled subscription so it does not leave a stale + /// entry behind. Acquires @ref lock_. + void eraseDataReaderIfCurrent(DataFrameCallbackId id, const std::shared_ptr& reader); + /// Protects callback registration maps and active reader state. mutable std::mutex lock_; diff --git a/src/subscription_thread_dispatcher.cpp b/src/subscription_thread_dispatcher.cpp index ed77d0be..2d175c7a 100644 --- a/src/subscription_thread_dispatcher.cpp +++ b/src/subscription_thread_dispatcher.cpp @@ -259,6 +259,8 @@ void SubscriptionThreadDispatcher::handleDataTrackUnpublished(const std::string& for (auto it = active_data_readers_.begin(); it != active_data_readers_.end();) { auto& reader = it->second; if (reader->remote_track && reader->remote_track->info().sid == sid) { + // Mark cancelled before closing to guard in flight subscriptions + reader->cancelled = true; { const std::scoped_lock sub_guard(reader->sub_mutex); if (reader->stream) { @@ -312,6 +314,8 @@ void SubscriptionThreadDispatcher::stopAll() { video_callbacks_.clear(); for (auto& [id, reader] : active_data_readers_) { + // Mark cancelled before closing to guard in flight subscriptions + reader->cancelled = true; { const std::scoped_lock sub_guard(reader->sub_mutex); if (reader->stream) { @@ -397,6 +401,16 @@ std::thread SubscriptionThreadDispatcher::startAudioReaderLocked(const CallbackK const AudioFrameCallback& cb, const AudioStream::Options& opts) { LK_LOG_DEBUG("Starting audio reader for participant={} track_name={}", key.participant_identity, key.track_name); + + auto existing = active_readers_.find(key); + if (existing != active_readers_.end() && existing->second.track_sid == track->sid()) { + LK_LOG_DEBUG( + "Skipping audio reader start for participant={} track_name={} because a " + "reader for sid={} is already active", + key.participant_identity, key.track_name, track->sid()); + return {}; + } + auto old_thread = extractReaderThreadLocked(key); if (static_cast(active_readers_.size()) >= kMaxActiveReaders) { @@ -415,6 +429,7 @@ std::thread SubscriptionThreadDispatcher::startAudioReaderLocked(const CallbackK ActiveReader reader; reader.audio_stream = stream; + reader.track_sid = track->sid(); const std::string participant_identity = key.participant_identity; const std::string track_name = key.track_name; // NOLINTBEGIN(bugprone-lambda-function-name,bugprone-exception-escape) @@ -454,6 +469,16 @@ std::thread SubscriptionThreadDispatcher::startVideoReaderLocked(const CallbackK const std::shared_ptr& track, const RegisteredVideoCallback& callback) { LK_LOG_DEBUG("Starting video reader for participant={} track_name={}", key.participant_identity, key.track_name); + + auto existing = active_readers_.find(key); + if (existing != active_readers_.end() && existing->second.track_sid == track->sid()) { + LK_LOG_DEBUG( + "Skipping video reader start for participant={} track_name={} because a " + "reader for sid={} is already active", + key.participant_identity, key.track_name, track->sid()); + return {}; + } + auto old_thread = extractReaderThreadLocked(key); if (static_cast(active_readers_.size()) >= kMaxActiveReaders) { @@ -472,6 +497,7 @@ std::thread SubscriptionThreadDispatcher::startVideoReaderLocked(const CallbackK ActiveReader reader; reader.video_stream = stream; + reader.track_sid = track->sid(); auto legacy_cb = callback.legacy_callback; auto event_cb = callback.event_callback; const std::string participant_identity = key.participant_identity; @@ -522,6 +548,8 @@ std::thread SubscriptionThreadDispatcher::extractDataReaderThreadLocked(DataFram } auto reader = std::move(it->second); active_data_readers_.erase(it); + // Mark cancelled before closing to guard in flight subscriptions + reader->cancelled = true; { const std::scoped_lock guard(reader->sub_mutex); if (reader->stream) { @@ -531,28 +559,36 @@ std::thread SubscriptionThreadDispatcher::extractDataReaderThreadLocked(DataFram return std::move(reader->thread); } -std::thread SubscriptionThreadDispatcher::extractDataReaderThreadLocked(const DataCallbackKey& key) { - for (auto it = active_data_readers_.begin(); it != active_data_readers_.end(); ++it) { - if (it->second && it->second->remote_track && - it->second->remote_track->publisherIdentity() == key.participant_identity && - it->second->remote_track->info().name == key.track_name) { - auto reader = std::move(it->second); - active_data_readers_.erase(it); - { - const std::scoped_lock guard(reader->sub_mutex); - if (reader->stream) { - reader->stream->close(); - } - } - return std::move(reader->thread); - } +void SubscriptionThreadDispatcher::eraseDataReaderIfCurrent(DataFrameCallbackId id, + const std::shared_ptr& reader) { + const std::scoped_lock lock(lock_); + auto it = active_data_readers_.find(id); + if (it == active_data_readers_.end() || it->second != reader) { + // The slot was already extracted or replaced; the owner joins that thread. + return; } - return {}; + // Detach so the reader can be destroyed by its own thread's final shared_ptr + // release without tripping std::thread's joinable-at-destruction check. The + // thread is already returning when this runs, so nothing is left to join. + if (reader->thread.joinable()) { + reader->thread.detach(); + } + active_data_readers_.erase(it); } std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbackId id, const DataCallbackKey& key, const std::shared_ptr& track, const DataFrameCallback& cb) { + auto existing = active_data_readers_.find(id); + if (existing != active_data_readers_.end() && existing->second->remote_track && + existing->second->remote_track->info().sid == track->info().sid) { + LK_LOG_DEBUG( + "Skipping data reader start for \"{}\" track=\"{}\" because a reader for " + "sid={} is already active", + key.participant_identity, key.track_name, track->info().sid); + return {}; + } + auto old_thread = extractDataReaderThreadLocked(id); const int total_active = static_cast(active_readers_.size()) + static_cast(active_data_readers_.size()); @@ -571,7 +607,7 @@ std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbac auto identity = key.participant_identity; auto track_name = key.track_name; // NOLINTBEGIN(bugprone-lambda-function-name) - reader->thread = std::thread([reader, track, cb, identity, track_name]() { + reader->thread = std::thread([this, id, reader, track, cb, identity, track_name]() { LK_LOG_INFO("Data reader thread: subscribing to \"{}\" track=\"{}\"", identity, track_name); std::shared_ptr stream; auto subscribe_result = track->subscribe(); @@ -581,6 +617,7 @@ std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbac "Failed to subscribe to data track \"{}\" from \"{}\": code={} " "message={}", track_name, identity, static_cast(error.code), error.message); + eraseDataReaderIfCurrent(id, reader); return; } stream = subscribe_result.value(); @@ -588,6 +625,13 @@ std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbac { const std::scoped_lock guard(reader->sub_mutex); + // A replacement or teardown may have cancelled this reader while the + // subscribe was in flight. Close the fresh stream and bail so we do not + // leave a second live subscription behind. + if (reader->cancelled.load()) { + stream->close(); + return; + } reader->stream = stream; } @@ -606,6 +650,9 @@ std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbac "\"{}\": code={} message={}", track_name, identity, static_cast(error->code), error->message); } + // Clean our own slot if the stream ended on its own (server EOS) and no + // extract/teardown already claimed it. A no-op when we were extracted. + eraseDataReaderIfCurrent(id, reader); LK_LOG_INFO("Data reader thread exiting for \"{}\" track=\"{}\"", identity, track_name); }); // NOLINTEND(bugprone-lambda-function-name) diff --git a/src/tests/integration/test_data_track.cpp b/src/tests/integration/test_data_track.cpp index 8548527b..bf7e5bde 100644 --- a/src/tests/integration/test_data_track.cpp +++ b/src/tests/integration/test_data_track.cpp @@ -468,6 +468,101 @@ TEST_F(DataTrackE2ETest, UnpublishUpdatesPublishedStateEndToEnd) { << "Remote track did not report unpublished state"; } +// Verifies that an auto-wired data callback (Room::addOnDataFrameCallback) +// follows a republished track: after unpublish + republish under the same +// (participant, track name) but a new SID, the previous reader is torn down and +// a fresh reader delivers frames from the new publication. +TEST_F(DataTrackE2ETest, RepublishRewiresDataCallbackToNewPublication) { + const auto track_name = makeTrackName("republish"); + + std::vector room_configs(2); + room_configs[0].room_options.single_peer_connection = false; + room_configs[1].room_options.single_peer_connection = false; + + DataTrackPublishedDelegate subscriber_delegate; + room_configs[1].delegate = &subscriber_delegate; + + auto rooms = testRooms(room_configs); + auto& publisher_room = rooms[0]; + auto& subscriber_room = rooms[1]; + const auto publisher_identity = lockLocalParticipant(*publisher_room)->identity(); + + std::atomic frames_received{0}; + std::mutex payload_mutex; + std::vector last_payload; + subscriber_room->addOnDataFrameCallback(publisher_identity, track_name, + [&](const std::vector& payload, std::optional) { + { + const std::scoped_lock lock(payload_mutex); + last_payload = payload; + } + frames_received.fetch_add(1); + }); + + auto publish_with_retry = [&](const std::string& name) -> std::shared_ptr { + std::shared_ptr track; + waitForCondition( + [&]() { + auto result = lockLocalParticipant(*publisher_room)->publishDataTrack(name); + if (result) { + track = result.value(); + return true; + } + return false; + }, + kTrackWaitTimeout); + return track; + }; + + // First publication. + auto first_track = publish_with_retry(track_name); + ASSERT_NE(first_track, nullptr) << "Failed to publish first data track"; + auto first_remote = subscriber_delegate.waitForTrack(kTrackWaitTimeout); + ASSERT_NE(first_remote, nullptr) << "Timed out waiting for first remote data track"; + const std::string first_sid = first_remote->info().sid; + + DataTrackFrame first_frame; + first_frame.payload.assign(64, 0xA1); + ASSERT_TRUE(waitForCondition( + [&]() { + requirePushSuccess(first_track->tryPush(first_frame), "Failed to push first-publication frame"); + return frames_received.load() > 0; + }, + kTransportFrameTimeout)) + << "Auto-wired callback never received a frame from the first publication"; + + // Unpublish: the reader for the first publication must be torn down. + first_track->unpublishDataTrack(); + ASSERT_TRUE(waitForCondition([&]() { return !first_remote->isPublished(); }, kTrackWaitTimeout)) + << "First remote track did not report unpublished state"; + const int frames_before_republish = frames_received.load(); + + // Republish under the same name; the server assigns a new SID. + auto second_track = publish_with_retry(track_name); + ASSERT_NE(second_track, nullptr) << "Failed to republish data track"; + auto remotes = subscriber_delegate.waitForTracks(2, kTrackWaitTimeout); + ASSERT_EQ(remotes.size(), 2u) << "Timed out waiting for republished remote data track"; + const std::string second_sid = remotes.back()->info().sid; + EXPECT_NE(first_sid, second_sid) << "Republish should produce a new SID"; + + DataTrackFrame second_frame; + second_frame.payload.assign(64, 0xB2); + ASSERT_TRUE(waitForCondition( + [&]() { + requirePushSuccess(second_track->tryPush(second_frame), "Failed to push republished frame"); + return frames_received.load() > frames_before_republish; + }, + kTransportFrameTimeout)) + << "Auto-wired callback did not re-wire to the republished track"; + + { + const std::scoped_lock lock(payload_mutex); + EXPECT_EQ(last_payload, second_frame.payload) << "Callback delivered stale payload after republish"; + } + + second_track->unpublishDataTrack(); +} + TEST_F(DataTrackE2ETest, SubscribeAfterUnpublishReportsTerminalError) { const auto track_name = makeTrackName("subscribe_after_unpublish"); diff --git a/src/tests/integration/test_room_event_deduplication.cpp b/src/tests/integration/test_room_event_deduplication.cpp new file mode 100644 index 00000000..a355afa7 --- /dev/null +++ b/src/tests/integration/test_room_event_deduplication.cpp @@ -0,0 +1,408 @@ +/* + * Copyright 2026 LiveKit + * + * 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 + * + * http://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 +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "../common/audio_utils.h" +#include "../common/test_common.h" +#include "../common/video_utils.h" + +namespace livekit::test { + +using namespace std::chrono_literals; + +namespace { + +constexpr auto kEventWaitTimeout = 20s; +constexpr auto kDuplicateGracePeriod = 500ms; + +struct RoomEventCounts { + std::mutex mutex; + std::condition_variable cv; + std::map participant_connected; + std::map participant_disconnected; + std::map track_published; + std::map track_subscribed; + std::map track_unsubscribed; + std::map track_unpublished; + int disconnected = 0; +}; + +struct RoomEventCountsSnapshot { + std::map participant_connected; + std::map participant_disconnected; + std::map track_published; + std::map track_subscribed; + std::map track_unsubscribed; + std::map track_unpublished; + int disconnected = 0; +}; + +RoomEventCountsSnapshot snapshotCounts(RoomEventCounts& counts) { + const std::scoped_lock lock(counts.mutex); + RoomEventCountsSnapshot snapshot; + snapshot.participant_connected = counts.participant_connected; + snapshot.participant_disconnected = counts.participant_disconnected; + snapshot.track_published = counts.track_published; + snapshot.track_subscribed = counts.track_subscribed; + snapshot.track_unsubscribed = counts.track_unsubscribed; + snapshot.track_unpublished = counts.track_unpublished; + snapshot.disconnected = counts.disconnected; + return snapshot; +} + +void incrementMap(std::map& counts, const std::string& key) { ++counts[key]; } + +class RoomEventCounterDelegate : public RoomDelegate { +public: + explicit RoomEventCounterDelegate(RoomEventCounts& counts) : counts_(counts) {} + + void onParticipantConnected(Room&, const ParticipantConnectedEvent& event) override { + if (event.participant == nullptr) { + return; + } + notify([&]() { incrementMap(counts_.participant_connected, event.participant->identity()); }); + } + + void onParticipantDisconnected(Room&, const ParticipantDisconnectedEvent& event) override { + if (event.participant == nullptr) { + return; + } + notify([&]() { incrementMap(counts_.participant_disconnected, event.participant->identity()); }); + } + + void onTrackPublished(Room&, const TrackPublishedEvent& event) override { + if (event.publication == nullptr) { + return; + } + notify([&]() { incrementMap(counts_.track_published, event.publication->name()); }); + } + + void onTrackSubscribed(Room&, const TrackSubscribedEvent& event) override { + if (event.publication == nullptr) { + return; + } + notify([&]() { incrementMap(counts_.track_subscribed, event.publication->name()); }); + } + + void onTrackUnsubscribed(Room&, const TrackUnsubscribedEvent& event) override { + if (event.publication == nullptr) { + return; + } + notify([&]() { incrementMap(counts_.track_unsubscribed, event.publication->name()); }); + } + + void onTrackUnpublished(Room&, const TrackUnpublishedEvent& event) override { + if (event.publication == nullptr) { + return; + } + notify([&]() { incrementMap(counts_.track_unpublished, event.publication->name()); }); + } + + void onDisconnected(Room&, const DisconnectedEvent&) override { + notify([&]() { ++counts_.disconnected; }); + } + +private: + template + void notify(Fn&& update) { + { + const std::scoped_lock lock(counts_.mutex); + update(); + } + counts_.cv.notify_all(); + } + + RoomEventCounts& counts_; +}; + +bool waitForMapCountAtLeast(RoomEventCounts& counts, const std::map& keys, int minimum_count, + std::chrono::milliseconds timeout) { + std::unique_lock lock(counts.mutex); + return counts.cv.wait_for(lock, timeout, [&]() { + for (const auto& [key, _] : keys) { + (void)_; + const auto it = counts.participant_connected.find(key); + if (it == counts.participant_connected.end() || it->second < minimum_count) { + return false; + } + } + return true; + }); +} + +bool waitForMapCountAtLeastTrack(RoomEventCounts& counts, const std::map& expected, + std::map RoomEventCounts::* member, + std::chrono::milliseconds timeout) { + std::unique_lock lock(counts.mutex); + return counts.cv.wait_for(lock, timeout, [&]() { + const auto& actual = counts.*member; + for (const auto& [key, minimum_count] : expected) { + const auto it = actual.find(key); + if (it == actual.end() || it->second < minimum_count) { + return false; + } + } + return true; + }); +} + +void expectMapCountsExact(const std::map& actual, const std::map& expected, + const char* label) { + for (const auto& [key, expected_count] : expected) { + const auto it = actual.find(key); + const int actual_count = it == actual.end() ? 0 : it->second; + EXPECT_EQ(actual_count, expected_count) << label << " count mismatch for key: " << key; + } +} + +void expectCountsUnchangedAfterGrace(RoomEventCounts& counts, const RoomEventCountsSnapshot& before, + const char* phase) { + std::this_thread::sleep_for(kDuplicateGracePeriod); + const RoomEventCountsSnapshot after = snapshotCounts(counts); + + expectMapCountsExact(after.participant_connected, before.participant_connected, + (std::string(phase) + " participant_connected duplicate").c_str()); + expectMapCountsExact(after.participant_disconnected, before.participant_disconnected, + (std::string(phase) + " participant_disconnected duplicate").c_str()); + expectMapCountsExact(after.track_published, before.track_published, + (std::string(phase) + " track_published duplicate").c_str()); + expectMapCountsExact(after.track_subscribed, before.track_subscribed, + (std::string(phase) + " track_subscribed duplicate").c_str()); + expectMapCountsExact(after.track_unsubscribed, before.track_unsubscribed, + (std::string(phase) + " track_unsubscribed duplicate").c_str()); + expectMapCountsExact(after.track_unpublished, before.track_unpublished, + (std::string(phase) + " track_unpublished duplicate").c_str()); + EXPECT_EQ(after.disconnected, before.disconnected) << phase << " onDisconnected duplicate"; +} + +std::string makeUniqueTrackName(const std::string& prefix) { return prefix + "-" + std::to_string(getTimestampUs()); } + +std::string describeCounts(const std::map& counts) { + std::ostringstream out; + bool first = true; + out << "{"; + for (const auto& [key, count] : counts) { + if (!first) { + out << ", "; + } + first = false; + out << key << ": " << count; + } + out << "}"; + return out.str(); +} + +class MediaLoopGuard { +public: + MediaLoopGuard() = default; + MediaLoopGuard(const MediaLoopGuard&) = delete; + MediaLoopGuard& operator=(const MediaLoopGuard&) = delete; + + ~MediaLoopGuard() { stop(); } + + void addAudioSource(const std::shared_ptr& source) { + threads_.emplace_back([this, source]() { + runToneLoop(source, running_, 440.0, false, kDefaultAudioSampleRate, kDefaultAudioChannels); + }); + } + + void addVideoSource(const std::shared_ptr& source) { + threads_.emplace_back([this, source]() { runVideoLoop(source, running_, fillWebcamWrapper); }); + } + + void stop() { + running_.store(false, std::memory_order_relaxed); + for (auto& thread : threads_) { + if (thread.joinable()) { + thread.join(); + } + } + } + +private: + std::atomic running_{true}; + std::vector threads_; +}; + +class PublishedTrackGuard { +public: + explicit PublishedTrackGuard(LocalParticipant* participant) : participant_(participant) {} + PublishedTrackGuard(const PublishedTrackGuard&) = delete; + PublishedTrackGuard& operator=(const PublishedTrackGuard&) = delete; + + ~PublishedTrackGuard() noexcept { + try { + unpublishAll(); + } catch (...) { + } + } + + void addTrackSid(const std::string& sid) { + if (!sid.empty()) { + track_sids_.push_back(sid); + } + } + + void unpublishAll() { + if (participant_ != nullptr) { + for (const auto& sid : track_sids_) { + if (!sid.empty()) { + participant_->unpublishTrack(sid); + } + } + } + track_sids_.clear(); + } + +private: + LocalParticipant* participant_ = nullptr; + std::vector track_sids_; +}; + +} // namespace + +class RoomEventDeduplicationIntegrationTest : public LiveKitTestBase, public ::testing::WithParamInterface { +protected: + void SetUp() override { + LiveKitTestBase::SetUp(); + if (!config_.available) { + GTEST_SKIP() << "LIVEKIT_URL, LIVEKIT_TOKEN_A, and LIVEKIT_TOKEN_B not set"; + } + } +}; + +TEST_P(RoomEventDeduplicationIntegrationTest, RoomLifecycleDelegateCallbacksFireExactlyOnce) { + const bool single_peer_connection = GetParam(); + + RoomOptions options; + options.auto_subscribe = true; + options.single_peer_connection = single_peer_connection; + + RoomEventCounts observer_counts; + RoomEventCounterDelegate observer_delegate(observer_counts); + + Room observer_room; + observer_room.setDelegate(&observer_delegate); + ASSERT_TRUE(observer_room.connect(config_.url, config_.token_b, options)) << "Observer failed to connect"; + ASSERT_FALSE(observer_room.localParticipant().expired()); + + Room peer_room; + ASSERT_TRUE(peer_room.connect(config_.url, config_.token_a, options)) << "Peer failed to connect"; + ASSERT_FALSE(peer_room.localParticipant().expired()); + + const std::string peer_identity = lockLocalParticipant(peer_room)->identity(); + ASSERT_FALSE(peer_identity.empty()); + + const std::map peer_identity_expected{{peer_identity, 1}}; + ASSERT_TRUE(waitForMapCountAtLeast(observer_counts, peer_identity_expected, 1, kEventWaitTimeout)) + << "Timed out waiting for onParticipantConnected"; + ASSERT_TRUE(waitForParticipant(&observer_room, peer_identity, 10s)) << "Peer not visible to observer room"; + { + const RoomEventCountsSnapshot snapshot = snapshotCounts(observer_counts); + expectMapCountsExact(snapshot.participant_connected, peer_identity_expected, "onParticipantConnected"); + expectCountsUnchangedAfterGrace(observer_counts, snapshot, "after participant connected"); + } + + const std::string audio_track_name = makeUniqueTrackName("dedupe-audio"); + const std::string video_track_name = makeUniqueTrackName("dedupe-video"); + + auto audio_source = std::make_shared(kDefaultAudioSampleRate, kDefaultAudioChannels, 0); + auto video_source = std::make_shared(kDefaultVideoWidth, kDefaultVideoHeight); + auto audio_track = LocalAudioTrack::createLocalAudioTrack(audio_track_name, audio_source); + auto video_track = LocalVideoTrack::createLocalVideoTrack(video_track_name, video_source); + + TrackPublishOptions audio_opts; + audio_opts.source = TrackSource::SOURCE_MICROPHONE; + TrackPublishOptions video_opts; + video_opts.source = TrackSource::SOURCE_CAMERA; + + auto peer_participant = lockLocalParticipant(peer_room); + PublishedTrackGuard published_tracks(peer_participant.get()); + MediaLoopGuard media_loops; + + ASSERT_NO_THROW(peer_participant->publishTrack(audio_track, audio_opts)); + ASSERT_NE(audio_track->publication(), nullptr); + published_tracks.addTrackSid(audio_track->publication()->sid()); + media_loops.addAudioSource(audio_source); + + ASSERT_NO_THROW(peer_participant->publishTrack(video_track, video_opts)); + ASSERT_NE(video_track->publication(), nullptr); + published_tracks.addTrackSid(video_track->publication()->sid()); + media_loops.addVideoSource(video_source); + + const std::map expected_subscribed_counts{{audio_track_name, 1}, {video_track_name, 1}}; + + ASSERT_TRUE(waitForMapCountAtLeastTrack(observer_counts, expected_subscribed_counts, + &RoomEventCounts::track_subscribed, kEventWaitTimeout)) + << "Timed out waiting for onTrackSubscribed; observed track_subscribed=" + << describeCounts(snapshotCounts(observer_counts).track_subscribed); + + { + const RoomEventCountsSnapshot snapshot = snapshotCounts(observer_counts); + expectMapCountsExact(snapshot.track_subscribed, expected_subscribed_counts, "onTrackSubscribed"); + expectCountsUnchangedAfterGrace(observer_counts, snapshot, "after track subscribe"); + } + + media_loops.stop(); + published_tracks.unpublishAll(); + + const std::map expected_unsubscribed_counts{{audio_track_name, 1}, {video_track_name, 1}}; + ASSERT_TRUE(waitForMapCountAtLeastTrack(observer_counts, expected_unsubscribed_counts, + &RoomEventCounts::track_unsubscribed, kEventWaitTimeout)) + << "Timed out waiting for onTrackUnsubscribed; observed track_unsubscribed=" + << describeCounts(snapshotCounts(observer_counts).track_unsubscribed); + + { + const RoomEventCountsSnapshot snapshot = snapshotCounts(observer_counts); + expectMapCountsExact(snapshot.track_unsubscribed, expected_unsubscribed_counts, "onTrackUnsubscribed"); + expectCountsUnchangedAfterGrace(observer_counts, snapshot, "after track unsubscribed"); + } + + peer_room.disconnect(); + + ASSERT_TRUE(waitForMapCountAtLeastTrack(observer_counts, peer_identity_expected, + &RoomEventCounts::participant_disconnected, kEventWaitTimeout)) + << "Timed out waiting for onParticipantDisconnected"; + + { + const RoomEventCountsSnapshot snapshot = snapshotCounts(observer_counts); + expectMapCountsExact(snapshot.participant_disconnected, peer_identity_expected, "onParticipantDisconnected"); + expectCountsUnchangedAfterGrace(observer_counts, snapshot, "after participant disconnected"); + } + + ASSERT_TRUE(observer_room.disconnect()) << "Observer disconnect failed"; + EXPECT_EQ(snapshotCounts(observer_counts).disconnected, 1) << "onDisconnected should fire exactly once"; + + const RoomEventCountsSnapshot after_disconnect = snapshotCounts(observer_counts); + expectCountsUnchangedAfterGrace(observer_counts, after_disconnect, "after observer disconnect"); + + EXPECT_FALSE(observer_room.disconnect()) << "Second disconnect should be a no-op"; + EXPECT_EQ(snapshotCounts(observer_counts).disconnected, 1) << "onDisconnected must not double-fire"; +} + +INSTANTIATE_TEST_SUITE_P(SingleAndDualPeerConnection, RoomEventDeduplicationIntegrationTest, ::testing::Bool()); + +} // namespace livekit::test diff --git a/src/tests/unit/test_subscription_thread_dispatcher.cpp b/src/tests/unit/test_subscription_thread_dispatcher.cpp index 80b52120..da0d00e6 100644 --- a/src/tests/unit/test_subscription_thread_dispatcher.cpp +++ b/src/tests/unit/test_subscription_thread_dispatcher.cpp @@ -18,14 +18,47 @@ #include #include +#include #include +#include +#include +#include +#include +#include #include #include #include namespace livekit { +namespace { + +using namespace std::chrono_literals; + +/// Minimal Track used to drive audio/video reader startup decisions without a +/// live FFI handle. The SID-skip check runs before any FFI call, so an invalid +/// handle is sufficient to exercise it deterministically. +class FakeMediaTrack : public Track { +public: + FakeMediaTrack(std::string sid, TrackKind kind) + : Track(FfiHandle(0), std::move(sid), "track", kind, StreamState::STATE_ACTIVE, false, true) {} +}; + +template +bool waitFor(Predicate predicate, std::chrono::milliseconds timeout) { + const auto start = std::chrono::steady_clock::now(); + while (std::chrono::steady_clock::now() - start < timeout) { + if (predicate()) { + return true; + } + std::this_thread::sleep_for(5ms); + } + return predicate(); +} + +} // namespace + class SubscriptionThreadDispatcherTest : public ::testing::Test { protected: void SetUp() override { livekit::initialize(livekit::LogLevel::Info); } @@ -37,6 +70,8 @@ class SubscriptionThreadDispatcherTest : public ::testing::Test { using DataCallbackKey = SubscriptionThreadDispatcher::DataCallbackKey; using DataCallbackKeyHash = SubscriptionThreadDispatcher::DataCallbackKeyHash; + using ActiveDataReader = SubscriptionThreadDispatcher::ActiveDataReader; + static auto& audioCallbacks(SubscriptionThreadDispatcher& dispatcher) { return dispatcher.audio_callbacks_; } static auto& videoCallbacks(SubscriptionThreadDispatcher& dispatcher) { return dispatcher.video_callbacks_; } static auto& activeReaders(SubscriptionThreadDispatcher& dispatcher) { return dispatcher.active_readers_; } @@ -44,6 +79,31 @@ 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 lock(dispatcher.lock_); + return dispatcher.active_readers_.size(); + } + static std::size_t activeDataReaderCount(SubscriptionThreadDispatcher& dispatcher) { + const std::scoped_lock lock(dispatcher.lock_); + return dispatcher.active_data_readers_.size(); + } + + static std::thread extractDataReader(SubscriptionThreadDispatcher& dispatcher, DataFrameCallbackId id) { + const std::scoped_lock lock(dispatcher.lock_); + return dispatcher.extractDataReaderThreadLocked(id); + } + + static void eraseDataReaderIfCurrent(SubscriptionThreadDispatcher& dispatcher, DataFrameCallbackId id, + const std::shared_ptr& reader) { + dispatcher.eraseDataReaderIfCurrent(id, reader); + } + + static std::thread startDataReader(SubscriptionThreadDispatcher& dispatcher, DataFrameCallbackId id, + const DataCallbackKey& key, const std::shared_ptr& track) { + const std::scoped_lock lock(dispatcher.lock_); + return dispatcher.startDataReaderLocked(id, key, track, + [](const std::vector&, std::optional) {}); + } }; // ============================================================================ @@ -478,6 +538,137 @@ TEST_F(SubscriptionThreadDispatcherTest, NoRemoteDataTracksInitially) { EXPECT_TRUE(remoteDataTracks(dispatcher).empty()); } +// ============================================================================ +// Data reader replacement: cancellation and self-erase +// ============================================================================ + +TEST_F(SubscriptionThreadDispatcherTest, ActiveDataReaderNotCancelledByDefault) { + auto reader = std::make_shared(); + EXPECT_FALSE(reader->cancelled.load()); +} + +TEST_F(SubscriptionThreadDispatcherTest, ExtractDataReaderMarksCancelledAndRemovesEntry) { + SubscriptionThreadDispatcher dispatcher; + auto reader = std::make_shared(); + activeDataReaders(dispatcher)[0] = reader; + + auto extracted = extractDataReader(dispatcher, 0); + + EXPECT_TRUE(reader->cancelled.load()) << "Extract must cancel so an in-flight subscribe aborts"; + EXPECT_FALSE(extracted.joinable()) << "No real thread was attached to the seeded reader"; + EXPECT_TRUE(activeDataReaders(dispatcher).empty()); +} + +TEST_F(SubscriptionThreadDispatcherTest, ExtractMissingDataReaderIsNoOp) { + SubscriptionThreadDispatcher dispatcher; + auto extracted = extractDataReader(dispatcher, 42); + EXPECT_FALSE(extracted.joinable()); +} + +TEST_F(SubscriptionThreadDispatcherTest, EraseDataReaderIfCurrentRemovesMatchingEntry) { + SubscriptionThreadDispatcher dispatcher; + auto reader = std::make_shared(); + activeDataReaders(dispatcher)[0] = reader; + + eraseDataReaderIfCurrent(dispatcher, 0, reader); + + EXPECT_TRUE(activeDataReaders(dispatcher).empty()); +} + +TEST_F(SubscriptionThreadDispatcherTest, EraseDataReaderIfCurrentLeavesReplacedEntry) { + SubscriptionThreadDispatcher dispatcher; + auto original = std::make_shared(); + auto replacement = std::make_shared(); + activeDataReaders(dispatcher)[0] = replacement; + + // The original reader exited after being replaced; it must not evict the + // newer reader that now owns the same callback id. + eraseDataReaderIfCurrent(dispatcher, 0, original); + + ASSERT_EQ(activeDataReaders(dispatcher).size(), 1u); + EXPECT_EQ(activeDataReaders(dispatcher)[0], replacement); +} + +// ============================================================================ +// SID deduplication: audio/video reader start is skipped for the same SID +// ============================================================================ + +TEST_F(SubscriptionThreadDispatcherTest, DuplicateSubscribeWithSameAudioSidDoesNotRestartReader) { + SubscriptionThreadDispatcher dispatcher; + dispatcher.setOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {}); + + // Simulate an already-running reader for this subscription. + const CallbackKey key{"alice", "mic"}; + activeReaders(dispatcher)[key].track_sid = "TR_audio_1"; + 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("TR_audio_1", TrackKind::KIND_AUDIO); + dispatcher.handleTrackSubscribed("alice", "mic", track); + + 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"; +} + +TEST_F(SubscriptionThreadDispatcherTest, DuplicateSubscribeWithSameVideoSidDoesNotRestartReader) { + SubscriptionThreadDispatcher dispatcher; + dispatcher.setOnVideoFrameCallback("alice", "cam", [](const VideoFrame&, std::int64_t) {}); + + const CallbackKey key{"alice", "cam"}; + activeReaders(dispatcher)[key].track_sid = "TR_video_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + auto track = std::make_shared("TR_video_1", TrackKind::KIND_VIDEO); + dispatcher.handleTrackSubscribed("alice", "cam", track); + + 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"; +} + +// ============================================================================ +// SID deduplication: data reader start is skipped for the same SID and +// replaced (stopping the previous reader) for a new SID +// ============================================================================ + +TEST_F(SubscriptionThreadDispatcherTest, DuplicateDataPublishWithSameSidDoesNotRestartReader) { + SubscriptionThreadDispatcher dispatcher; + auto reader = std::make_shared(); + reader->remote_track = RemoteDataTrack::createForTest({"foo", "TR_data_1", false}, "alice"); + activeDataReaders(dispatcher)[7] = reader; + + auto incoming = RemoteDataTrack::createForTest({"foo", "TR_data_1", false}, "alice"); + auto old_thread = startDataReader(dispatcher, 7, DataCallbackKey{"alice", "foo"}, incoming); + + EXPECT_FALSE(old_thread.joinable()); + 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"; +} + +TEST_F(SubscriptionThreadDispatcherTest, RepublishWithNewDataSidStopsPreviousReader) { + SubscriptionThreadDispatcher dispatcher; + auto previous = std::make_shared(); + previous->remote_track = RemoteDataTrack::createForTest({"foo", "TR_data_1", false}, "alice"); + activeDataReaders(dispatcher)[7] = previous; + + // A republish under the same (participant, name) but a NEW SID must stop the + // previous reader and start a fresh one. + auto republished = RemoteDataTrack::createForTest({"foo", "TR_data_2", false}, "alice"); + auto old_thread = startDataReader(dispatcher, 7, DataCallbackKey{"alice", "foo"}, republished); + if (old_thread.joinable()) { + old_thread.join(); + } + + EXPECT_TRUE(previous->cancelled.load()) << "Previous reader must be cancelled on republish"; + + // 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 activeDataReaderCount(dispatcher) == 0; }, 2s)); +} + // ============================================================================ // Data track destruction safety // ============================================================================ From f1ff6e6f576769b4b0460e5a5442a683109b1056 Mon Sep 17 00:00:00 2001 From: Stephen DeRosa Date: Tue, 7 Jul 2026 12:16:15 -0600 Subject: [PATCH 02/11] more seperated test artifacts --- .../common/remote_data_track_test_access.h | 46 +++++++++++++++++++ src/tests/integration/test_data_track.cpp | 5 +- .../test_subscription_thread_dispatcher.cpp | 10 ++-- 3 files changed, 55 insertions(+), 6 deletions(-) create mode 100644 src/tests/common/remote_data_track_test_access.h diff --git a/src/tests/common/remote_data_track_test_access.h b/src/tests/common/remote_data_track_test_access.h new file mode 100644 index 00000000..619d1d62 --- /dev/null +++ b/src/tests/common/remote_data_track_test_access.h @@ -0,0 +1,46 @@ +/* + * Copyright 2026 LiveKit + * + * 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 + * + * http://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. + */ + +#pragma once + +#include +#include + +#include +#include +#include +#include + +#include "data_track.pb.h" + +namespace livekit { + +struct RemoteDataTrackTestAccess { + static uintptr_t ffiHandleId(const RemoteDataTrack& track) noexcept { return track.ffiHandleId(); } + + static std::shared_ptr create(DataTrackInfo info, std::string publisher_identity) { + proto::OwnedRemoteDataTrack owned; + owned.mutable_handle()->set_id(0); + auto* proto_info = owned.mutable_info(); + proto_info->set_name(std::move(info.name)); + proto_info->set_sid(std::move(info.sid)); + proto_info->set_uses_e2ee(info.uses_e2ee); + owned.set_publisher_identity(std::move(publisher_identity)); + return std::shared_ptr(new RemoteDataTrack(owned)); + } +}; + +} // namespace livekit diff --git a/src/tests/integration/test_data_track.cpp b/src/tests/integration/test_data_track.cpp index bf7e5bde..8663218f 100644 --- a/src/tests/integration/test_data_track.cpp +++ b/src/tests/integration/test_data_track.cpp @@ -29,6 +29,7 @@ #include #include +#include "../common/remote_data_track_test_access.h" #include "../common/test_common.h" #include "ffi_client.h" @@ -782,8 +783,8 @@ TEST_F(DataTrackE2ETest, FfiClientSubscribeDataTrackReturnsSyncResult) { EXPECT_EQ(remote_track->info().name, expected_name); const auto subscribe_start = std::chrono::steady_clock::now(); - auto subscribe_result = - FfiClient::instance().subscribeDataTrack(static_cast(remote_track->testFfiHandleId())); + auto subscribe_result = FfiClient::instance().subscribeDataTrack( + static_cast(RemoteDataTrackTestAccess::ffiHandleId(*remote_track))); const auto subscribe_elapsed = std::chrono::steady_clock::now() - subscribe_start; const auto subscribe_elapsed_ns = std::chrono::duration_cast(subscribe_elapsed).count(); diff --git a/src/tests/unit/test_subscription_thread_dispatcher.cpp b/src/tests/unit/test_subscription_thread_dispatcher.cpp index da0d00e6..95e46fbd 100644 --- a/src/tests/unit/test_subscription_thread_dispatcher.cpp +++ b/src/tests/unit/test_subscription_thread_dispatcher.cpp @@ -30,6 +30,8 @@ #include #include +#include "../common/remote_data_track_test_access.h" + namespace livekit { namespace { @@ -636,10 +638,10 @@ TEST_F(SubscriptionThreadDispatcherTest, DuplicateSubscribeWithSameVideoSidDoesN TEST_F(SubscriptionThreadDispatcherTest, DuplicateDataPublishWithSameSidDoesNotRestartReader) { SubscriptionThreadDispatcher dispatcher; auto reader = std::make_shared(); - reader->remote_track = RemoteDataTrack::createForTest({"foo", "TR_data_1", false}, "alice"); + reader->remote_track = RemoteDataTrackTestAccess::create({"foo", "TR_data_1", false}, "alice"); activeDataReaders(dispatcher)[7] = reader; - auto incoming = RemoteDataTrack::createForTest({"foo", "TR_data_1", false}, "alice"); + auto incoming = RemoteDataTrackTestAccess::create({"foo", "TR_data_1", false}, "alice"); auto old_thread = startDataReader(dispatcher, 7, DataCallbackKey{"alice", "foo"}, incoming); EXPECT_FALSE(old_thread.joinable()); @@ -651,12 +653,12 @@ TEST_F(SubscriptionThreadDispatcherTest, DuplicateDataPublishWithSameSidDoesNotR TEST_F(SubscriptionThreadDispatcherTest, RepublishWithNewDataSidStopsPreviousReader) { SubscriptionThreadDispatcher dispatcher; auto previous = std::make_shared(); - previous->remote_track = RemoteDataTrack::createForTest({"foo", "TR_data_1", false}, "alice"); + previous->remote_track = RemoteDataTrackTestAccess::create({"foo", "TR_data_1", false}, "alice"); activeDataReaders(dispatcher)[7] = previous; // A republish under the same (participant, name) but a NEW SID must stop the // previous reader and start a fresh one. - auto republished = RemoteDataTrack::createForTest({"foo", "TR_data_2", false}, "alice"); + auto republished = RemoteDataTrackTestAccess::create({"foo", "TR_data_2", false}, "alice"); auto old_thread = startDataReader(dispatcher, 7, DataCallbackKey{"alice", "foo"}, republished); if (old_thread.joinable()) { old_thread.join(); From 338f266dd2cb0ec2b9ef9a34816d7f32547eca06 Mon Sep 17 00:00:00 2001 From: Stephen DeRosa Date: Tue, 7 Jul 2026 13:20:24 -0600 Subject: [PATCH 03/11] doc --- include/livekit/subscription_thread_dispatcher.h | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/include/livekit/subscription_thread_dispatcher.h b/include/livekit/subscription_thread_dispatcher.h index 2890b7bb..b0fa127e 100644 --- a/include/livekit/subscription_thread_dispatcher.h +++ b/include/livekit/subscription_thread_dispatcher.h @@ -329,7 +329,10 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// must be joined after releasing the lock. std::thread extractReaderThreadLocked(const CallbackKey& key); - /// Select the appropriate reader startup path for @p track. + /// Select the appropriate reader startup path for @p media track. + /// + /// This is called by @ref Room when a remote track is subscribed. If a reader is already active for the given key, + /// std::thread default constructed is returned. /// /// Must be called with @ref lock_ held. std::thread startReaderLocked(const CallbackKey& key, const std::shared_ptr& track); From 709a8ab26a27ee3229af23e878d34c50e55c1979 Mon Sep 17 00:00:00 2001 From: Stephen DeRosa Date: Tue, 7 Jul 2026 15:46:22 -0600 Subject: [PATCH 04/11] SubscriptionThreadDispatcher: proper replacing of audio/video callbacks. Deprecate a setOn*Callback(), replace with trySetOn*Callback() --- include/livekit/room.h | 116 +++++++++++++++++- .../livekit/subscription_thread_dispatcher.h | 92 ++++++++++++-- src/room.cpp | 91 ++++++++++++-- src/subscription_thread_dispatcher.cpp | 65 ++++++++-- src/tests/integration/test_platform_audio.cpp | 22 ++-- .../integration/test_video_frame_metadata.cpp | 46 +++---- src/tests/unit/test_room_callbacks.cpp | 22 ++-- .../test_subscription_thread_dispatcher.cpp | 108 ++++++++++++---- 8 files changed, 460 insertions(+), 102 deletions(-) diff --git a/include/livekit/room.h b/include/livekit/room.h index bf4bf68d..8d47733c 100644 --- a/include/livekit/room.h +++ b/include/livekit/room.h @@ -303,16 +303,118 @@ 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, + 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, + 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. + /// Replacing an active callback is not supported through this overload; call + /// @ref clearOnAudioFrameCallback first, then @ref trySetOnAudioFrameCallback. + /// + /// @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. + [[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. + /// Replacing an active callback is not supported through this overload; call + /// @ref clearOnVideoFrameCallback first, then @ref trySetOnVideoFrameCallback. + /// + /// @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. + [[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. + /// Replacing an active callback is not supported through this overload; call + /// @ref clearOnVideoFrameCallback first, then + /// @ref trySetOnVideoFrameEventCallback. + /// + /// @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. + [[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 = {}); @@ -354,6 +456,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 findSubscribedRemoteTrack(const std::string& participant_identity, + const std::string& track_name) const; + void onEvent(const proto::FfiEvent& event); }; } // namespace livekit diff --git a/include/livekit/subscription_thread_dispatcher.h b/include/livekit/subscription_thread_dispatcher.h index b0fa127e..a4fabc15 100644 --- a/include/livekit/subscription_thread_dispatcher.h +++ b/include/livekit/subscription_thread_dispatcher.h @@ -93,54 +93,124 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// Stops all active readers and clears all registered callbacks. ~SubscriptionThreadDispatcher(); - /// Register or replace an audio frame callback for a remote subscription. + /// Register an audio frame callback for a remote subscription. /// /// The callback is keyed by remote participant identity plus @p track_name. /// If the matching remote audio track is already subscribed, @ref Room may /// immediately call @ref handleTrackSubscribed to start a reader. /// + /// Registration only succeeds when no reader is currently active for the + /// key. To replace a callback whose reader is already running, call + /// @ref clearOnAudioFrameCallback 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 audio frame. /// @param opts Options used when creating the backing /// @ref AudioStream. - void setOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name, - AudioFrameCallback callback, const AudioStream::Options& opts = {}); + /// @return @c true if the callback was registered; @c false if a reader is + /// already active for the key (the registration is left unchanged). + [[nodiscard]] bool trySetOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name, + AudioFrameCallback callback, const AudioStream::Options& opts = {}); - /// Register or replace a video frame callback for a remote subscription. + /// Register a video frame callback for a remote subscription. /// /// The callback is keyed by remote participant identity plus @p track_name. /// If the matching remote video track is already subscribed, @ref Room may /// immediately call @ref handleTrackSubscribed to start a reader. /// + /// Registration only succeeds when no reader is currently active for the + /// key. 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. - void setOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name, - VideoFrameCallback callback, const VideoStream::Options& opts = {}); + /// @return @c true if the callback was registered; @c false if a reader is + /// already active for the key (the registration is left unchanged). + [[nodiscard]] bool trySetOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name, + VideoFrameCallback callback, const VideoStream::Options& opts = {}); - /// Register or replace a rich video frame event callback for a remote - /// subscription. + /// Register a rich video frame event callback for a remote subscription. /// /// The callback is keyed by remote participant identity plus @p track_name. /// If the matching remote video track is already subscribed, @ref Room may /// immediately call @ref handleTrackSubscribed to start a reader. /// + /// Registration only succeeds when no reader is currently active for the + /// key. 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 (the registration is left unchanged). + [[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. + /// Replacing an active callback is not supported through this overload; call + /// @ref clearOnAudioFrameCallback first, then @ref trySetOnAudioFrameCallback. + /// + /// @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. + [[deprecated( + "SubscriptionThreadDispatcher::setOnAudioFrameCallback is deprecated; use trySetOnAudioFrameCallback instead")]] + void setOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name, + AudioFrameCallback callback, const AudioStream::Options& opts = {}); + + /// @deprecated Use trySetOnVideoFrameCallback() instead. + /// + /// Forwards to @ref trySetOnVideoFrameCallback and discards the result. + /// Replacing an active callback is not supported through this overload; call + /// @ref clearOnVideoFrameCallback first, then @ref trySetOnVideoFrameCallback. + /// + /// @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. + [[deprecated( + "SubscriptionThreadDispatcher::setOnVideoFrameCallback is deprecated; use trySetOnVideoFrameCallback instead")]] + void setOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name, + VideoFrameCallback callback, const VideoStream::Options& opts = {}); + + /// @deprecated Use trySetOnVideoFrameEventCallback() instead. + /// + /// Forwards to @ref trySetOnVideoFrameEventCallback and discards the result. + /// Replacing an active callback is not supported through this overload; call + /// @ref clearOnVideoFrameCallback first, then + /// @ref trySetOnVideoFrameEventCallback. + /// /// @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. + [[deprecated( + "SubscriptionThreadDispatcher::setOnVideoFrameEventCallback is deprecated; use " + "trySetOnVideoFrameEventCallback instead")]] void setOnVideoFrameEventCallback(const std::string& participant_identity, const std::string& track_name, VideoFrameEventCallback callback, const VideoStream::Options& opts = {}); /// Remove an audio callback registration and stop any active reader. /// /// If an audio reader thread is active for the given key, its stream is - /// closed and the thread is joined before this call returns. + /// closed and the thread is joined before this call returns. Call this + /// before @ref trySetOnAudioFrameCallback to replace an active callback. /// /// @param participant_identity Identity of the remote participant. /// @param track_name Track name to clear. @@ -149,7 +219,9 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// Remove a video callback registration and stop any active reader. /// /// If a video reader thread is active for the given key, its stream is - /// closed and the thread is joined before this call returns. + /// closed and the thread is joined before this call returns. Call this + /// before @ref trySetOnVideoFrameCallback (or + /// @ref trySetOnVideoFrameEventCallback) to replace an active callback. /// /// @param participant_identity Identity of the remote participant. /// @param track_name Track name to clear. diff --git a/src/room.cpp b/src/room.cpp index 32dac5eb..61f45a6e 100644 --- a/src/room.cpp +++ b/src/room.cpp @@ -370,28 +370,95 @@ void Room::unregisterByteStreamHandler(const std::string& topic) { // Frame callback registration // ------------------------------------------------------------------- +std::shared_ptr Room::findSubscribedRemoteTrack(const std::string& participant_identity, + const std::string& track_name) const { + const std::scoped_lock guard(lock_); + auto pit = remote_participants_.find(participant_identity); + if (pit == remote_participants_.end() || !pit->second) { + return nullptr; + } + for (const auto& [sid, publication] : pit->second->trackPublications()) { + (void)sid; + if (publication && publication->subscribed() && publication->name() == track_name) { + return publication->track(); + } + } + return nullptr; +} + +bool Room::trySetOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name, + AudioFrameCallback callback, const AudioStream::Options& opts) { + if (!subscription_thread_dispatcher_) { + LK_LOG_ERROR("Room::trySetOnAudioFrameCallback: subscription_thread_dispatcher_ is nullptr"); + return false; + } + if (!subscription_thread_dispatcher_->trySetOnAudioFrameCallback(participant_identity, track_name, + std::move(callback), opts)) { + return false; + } + auto track = findSubscribedRemoteTrack(participant_identity, track_name); + if (!track) { + LK_LOG_ERROR("Room::trySetOnAudioFrameCallback: track not found for participant={} track_name={}", + participant_identity, track_name); + return false; + } + subscription_thread_dispatcher_->handleTrackSubscribed(participant_identity, track_name, track); + return true; +} + +bool Room::trySetOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name, + VideoFrameCallback callback, const VideoStream::Options& opts) { + if (!subscription_thread_dispatcher_) { + LK_LOG_ERROR("Room::trySetOnVideoFrameCallback: subscription_thread_dispatcher_ is nullptr"); + return false; + } + if (!subscription_thread_dispatcher_->trySetOnVideoFrameCallback(participant_identity, track_name, + std::move(callback), opts)) { + return false; + } + auto track = findSubscribedRemoteTrack(participant_identity, track_name); + if (!track) { + LK_LOG_ERROR("Room::trySetOnVideoFrameCallback: track not found for participant={} track_name={}", + participant_identity, track_name); + return false; + } + subscription_thread_dispatcher_->handleTrackSubscribed(participant_identity, track_name, track); + return true; +} + +bool Room::trySetOnVideoFrameEventCallback(const std::string& participant_identity, const std::string& track_name, + VideoFrameEventCallback callback, const VideoStream::Options& opts) { + if (!subscription_thread_dispatcher_) { + LK_LOG_ERROR("Room::trySetOnVideoFrameEventCallback: subscription_thread_dispatcher_ is nullptr"); + return false; + } + if (!subscription_thread_dispatcher_->trySetOnVideoFrameEventCallback(participant_identity, track_name, + std::move(callback), opts)) { + return false; + } + auto track = findSubscribedRemoteTrack(participant_identity, track_name); + if (!track) { + LK_LOG_ERROR("Room::trySetOnVideoFrameEventCallback: track not found for participant={} track_name={}", + participant_identity, track_name); + return false; + } + subscription_thread_dispatcher_->handleTrackSubscribed(participant_identity, track_name, track); + return true; +} + void Room::setOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name, AudioFrameCallback callback, const AudioStream::Options& opts) { - if (subscription_thread_dispatcher_) { - subscription_thread_dispatcher_->setOnAudioFrameCallback(participant_identity, track_name, std::move(callback), - opts); - } + (void)trySetOnAudioFrameCallback(participant_identity, track_name, std::move(callback), opts); } void Room::setOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name, VideoFrameCallback callback, const VideoStream::Options& opts) { - if (subscription_thread_dispatcher_) { - subscription_thread_dispatcher_->setOnVideoFrameCallback(participant_identity, track_name, std::move(callback), - opts); - } + (void)trySetOnVideoFrameCallback(participant_identity, track_name, std::move(callback), opts); } void Room::setOnVideoFrameEventCallback(const std::string& participant_identity, const std::string& track_name, VideoFrameEventCallback callback, const VideoStream::Options& opts) { - if (subscription_thread_dispatcher_) { - subscription_thread_dispatcher_->setOnVideoFrameEventCallback(participant_identity, track_name, std::move(callback), - opts); - } + (void)trySetOnVideoFrameEventCallback(participant_identity, track_name, std::move(callback), opts); } void Room::clearOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name) { diff --git a/src/subscription_thread_dispatcher.cpp b/src/subscription_thread_dispatcher.cpp index 2d175c7a..651d94c1 100644 --- a/src/subscription_thread_dispatcher.cpp +++ b/src/subscription_thread_dispatcher.cpp @@ -57,25 +57,41 @@ SubscriptionThreadDispatcher::~SubscriptionThreadDispatcher() { } // NOLINTEND(bugprone-exception-escape) -void SubscriptionThreadDispatcher::setOnAudioFrameCallback(const std::string& participant_identity, - const std::string& track_name, AudioFrameCallback callback, - const AudioStream::Options& opts) { +bool SubscriptionThreadDispatcher::trySetOnAudioFrameCallback(const std::string& participant_identity, + const std::string& track_name, + AudioFrameCallback callback, + const AudioStream::Options& opts) { const CallbackKey key{participant_identity, track_name}; const std::scoped_lock lock(lock_); + if (active_readers_.find(key) != active_readers_.end()) { + LK_LOG_WARN( + "Cannot register audio frame callback for participant={} track_name={} " + "because a reader is already active; call clearOnAudioFrameCallback() first", + participant_identity, track_name); + return false; + } const bool replacing = audio_callbacks_.find(key) != audio_callbacks_.end(); audio_callbacks_[key] = RegisteredAudioCallback{std::move(callback), opts}; LK_LOG_DEBUG( "Registered audio frame callback for participant={} track_name={} " "replacing_existing={} total_audio_callbacks={}", participant_identity, track_name, replacing, audio_callbacks_.size()); + return true; } -void SubscriptionThreadDispatcher::setOnVideoFrameEventCallback(const std::string& participant_identity, - const std::string& track_name, - VideoFrameEventCallback callback, - const VideoStream::Options& opts) { +bool SubscriptionThreadDispatcher::trySetOnVideoFrameEventCallback(const std::string& participant_identity, + const std::string& track_name, + VideoFrameEventCallback callback, + const VideoStream::Options& opts) { const CallbackKey key{participant_identity, track_name}; const std::scoped_lock lock(lock_); + if (active_readers_.find(key) != active_readers_.end()) { + LK_LOG_WARN( + "Cannot register video frame event callback for participant={} track_name={} " + "because a reader is already active; call clearOnVideoFrameCallback() first", + participant_identity, track_name); + return false; + } const bool replacing = video_callbacks_.find(key) != video_callbacks_.end(); video_callbacks_[key] = RegisteredVideoCallback{ VideoFrameCallback{}, @@ -86,13 +102,22 @@ void SubscriptionThreadDispatcher::setOnVideoFrameEventCallback(const std::strin "Registered video frame event callback for participant={} track_name={} " "replacing_existing={} total_video_callbacks={}", participant_identity, track_name, replacing, video_callbacks_.size()); + return true; } -void SubscriptionThreadDispatcher::setOnVideoFrameCallback(const std::string& participant_identity, - const std::string& track_name, VideoFrameCallback callback, - const VideoStream::Options& opts) { +bool SubscriptionThreadDispatcher::trySetOnVideoFrameCallback(const std::string& participant_identity, + const std::string& track_name, + VideoFrameCallback callback, + const VideoStream::Options& opts) { const CallbackKey key{participant_identity, track_name}; const std::scoped_lock lock(lock_); + if (active_readers_.find(key) != active_readers_.end()) { + LK_LOG_WARN( + "Cannot register video frame callback for participant={} track_name={} " + "because a reader is already active; call clearOnVideoFrameCallback() first", + participant_identity, track_name); + return false; + } const bool replacing = video_callbacks_.find(key) != video_callbacks_.end(); video_callbacks_[key] = RegisteredVideoCallback{ std::move(callback), @@ -103,6 +128,26 @@ void SubscriptionThreadDispatcher::setOnVideoFrameCallback(const std::string& pa "Registered video frame callback for participant={} track_name={} " "replacing_existing={} total_video_callbacks={}", participant_identity, track_name, replacing, video_callbacks_.size()); + return true; +} + +void SubscriptionThreadDispatcher::setOnAudioFrameCallback(const std::string& participant_identity, + const std::string& track_name, AudioFrameCallback callback, + const AudioStream::Options& opts) { + (void)trySetOnAudioFrameCallback(participant_identity, track_name, std::move(callback), opts); +} + +void SubscriptionThreadDispatcher::setOnVideoFrameEventCallback(const std::string& participant_identity, + const std::string& track_name, + VideoFrameEventCallback callback, + const VideoStream::Options& opts) { + (void)trySetOnVideoFrameEventCallback(participant_identity, track_name, std::move(callback), opts); +} + +void SubscriptionThreadDispatcher::setOnVideoFrameCallback(const std::string& participant_identity, + const std::string& track_name, VideoFrameCallback callback, + const VideoStream::Options& opts) { + (void)trySetOnVideoFrameCallback(participant_identity, track_name, std::move(callback), opts); } void SubscriptionThreadDispatcher::clearOnAudioFrameCallback(const std::string& participant_identity, diff --git a/src/tests/integration/test_platform_audio.cpp b/src/tests/integration/test_platform_audio.cpp index c6e86596..647956c3 100644 --- a/src/tests/integration/test_platform_audio.cpp +++ b/src/tests/integration/test_platform_audio.cpp @@ -287,16 +287,18 @@ TEST_F(PlatformAudioIntegrationTest, PlatformAudioFramesReachRemote) { // The reader thread is only started when the subscription event fires and a // matching callback is already registered, so register before publishing. - receiver_room->setOnAudioFrameCallback(sender_identity, track_name, [&](const AudioFrame& frame) { - if (frame.totalSamples() == 0) { - return; - } - { - std::lock_guard lock(frame_mutex); - ++received_frames; - } - frame_cv.notify_all(); - }); + const bool audio_callback_registered = + receiver_room->trySetOnAudioFrameCallback(sender_identity, track_name, [&](const AudioFrame& frame) { + if (frame.totalSamples() == 0) { + return; + } + { + std::lock_guard lock(frame_mutex); + ++received_frames; + } + frame_cv.notify_all(); + }); + ASSERT_TRUE(audio_callback_registered); TrackPublishOptions publish_options; publish_options.source = TrackSource::SOURCE_MICROPHONE; diff --git a/src/tests/integration/test_video_frame_metadata.cpp b/src/tests/integration/test_video_frame_metadata.cpp index 4c309b05..63b48664 100644 --- a/src/tests/integration/test_video_frame_metadata.cpp +++ b/src/tests/integration/test_video_frame_metadata.cpp @@ -51,18 +51,18 @@ TEST_F(VideoFrameMetadataServerTest, UserTimestampRoundTripsToReceiverEventCallb std::optional received_user_timestamp_us; const std::string track_name = "metadata-track"; - receiver_room.setOnVideoFrameEventCallback(sender_identity, track_name, - [&mutex, &cv, &received_user_timestamp_us](const VideoFrameEvent& event) { - std::lock_guard lock(mutex); - if (!event.metadata) { - return; - } - const auto& user_timestamp_us = event.metadata->user_timestamp_us; - if (user_timestamp_us.has_value() && *user_timestamp_us != 0) { - received_user_timestamp_us = user_timestamp_us; - cv.notify_all(); - } - }); + ASSERT_TRUE(receiver_room.trySetOnVideoFrameEventCallback( + sender_identity, track_name, [&mutex, &cv, &received_user_timestamp_us](const VideoFrameEvent& event) { + std::lock_guard lock(mutex); + if (!event.metadata) { + return; + } + const auto& user_timestamp_us = event.metadata->user_timestamp_us; + if (user_timestamp_us.has_value() && *user_timestamp_us != 0) { + received_user_timestamp_us = user_timestamp_us; + cv.notify_all(); + } + })); auto source = std::make_shared(16, 16); auto track = LocalVideoTrack::createLocalVideoTrack(track_name, source); @@ -180,17 +180,17 @@ TEST_F(VideoFrameMetadataServerTest, UserDataRoundTripsToReceiverEventCallback) const std::string track_name = "userdata-track"; const std::vector expected_user_data{0x01, 0x02, 0xab, 0xcd, 0xef}; - receiver_room.setOnVideoFrameEventCallback(sender_identity, track_name, - [&mutex, &cv, &received_user_data](const VideoFrameEvent& event) { - std::lock_guard lock(mutex); - if (!event.metadata || !event.metadata->user_data.has_value()) { - return; - } - if (!event.metadata->user_data->empty()) { - received_user_data = event.metadata->user_data; - cv.notify_all(); - } - }); + ASSERT_TRUE(receiver_room.trySetOnVideoFrameEventCallback( + sender_identity, track_name, [&mutex, &cv, &received_user_data](const VideoFrameEvent& event) { + std::lock_guard lock(mutex); + if (!event.metadata || !event.metadata->user_data.has_value()) { + return; + } + if (!event.metadata->user_data->empty()) { + received_user_data = event.metadata->user_data; + cv.notify_all(); + } + })); auto source = std::make_shared(16, 16); auto track = LocalVideoTrack::createLocalVideoTrack(track_name, source); diff --git a/src/tests/unit/test_room_callbacks.cpp b/src/tests/unit/test_room_callbacks.cpp index 71349d5b..f74524dc 100644 --- a/src/tests/unit/test_room_callbacks.cpp +++ b/src/tests/unit/test_room_callbacks.cpp @@ -37,12 +37,20 @@ class RoomCallbackTest : public ::testing::Test { TEST_F(RoomCallbackTest, FrameCallbackRegistrationByTrackNameIsAccepted) { Room room; - EXPECT_NO_THROW(room.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {})); - EXPECT_NO_THROW(room.setOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {})); + EXPECT_TRUE(room.trySetOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {})); + EXPECT_TRUE(room.trySetOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {})); EXPECT_NO_THROW(room.clearOnAudioFrameCallback("alice", "mic-main")); EXPECT_NO_THROW(room.clearOnVideoFrameCallback("alice", "cam-main")); } +TEST_F(RoomCallbackTest, TrySetOnAudioReturnsTrueWithoutSubscription) { + // Without a subscribed track, registration succeeds and no reader starts. + Room room; + EXPECT_TRUE(room.trySetOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {})); + // Re-registering the same key while no reader is active is allowed. + EXPECT_TRUE(room.trySetOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {})); +} + TEST_F(RoomCallbackTest, DataCallbackRegistrationReturnsUsableIds) { Room room; @@ -68,8 +76,8 @@ TEST_F(RoomCallbackTest, RemovingUnknownDataCallbackIsNoOp) { TEST_F(RoomCallbackTest, DestroyRoomWithRegisteredCallbacksIsSafe) { EXPECT_NO_THROW({ Room room; - room.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); - room.setOnVideoFrameCallback("bob", "cam-main", [](const VideoFrame&, std::int64_t) {}); + (void)room.trySetOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); + (void)room.trySetOnVideoFrameCallback("bob", "cam-main", [](const VideoFrame&, std::int64_t) {}); room.addOnDataFrameCallback("carol", "track", [](const std::vector&, std::optional) {}); }); @@ -78,7 +86,7 @@ TEST_F(RoomCallbackTest, DestroyRoomWithRegisteredCallbacksIsSafe) { TEST_F(RoomCallbackTest, DestroyRoomAfterClearingCallbacksIsSafe) { EXPECT_NO_THROW({ Room room; - room.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); + (void)room.trySetOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); room.clearOnAudioFrameCallback("alice", "mic-main"); const auto id = room.addOnDataFrameCallback("alice", "track", @@ -95,8 +103,8 @@ TEST_F(RoomCallbackTest, DefaultConnectionStateIsDisconnected) { TEST_F(RoomCallbackTest, ConnectionStateRemainsDisconnectedWithoutConnect) { // Register callbacks, do other operations — state must stay Disconnected. Room room; - room.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); - room.setOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); + (void)room.trySetOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); + (void)room.trySetOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); room.addOnDataFrameCallback("alice", "track", [](const std::vector&, std::optional) {}); room.registerTextStreamHandler("topic", [](const std::shared_ptr&, const std::string&) {}); EXPECT_EQ(room.connectionState(), ConnectionState::Disconnected); diff --git a/src/tests/unit/test_subscription_thread_dispatcher.cpp b/src/tests/unit/test_subscription_thread_dispatcher.cpp index 95e46fbd..dc56f2f5 100644 --- a/src/tests/unit/test_subscription_thread_dispatcher.cpp +++ b/src/tests/unit/test_subscription_thread_dispatcher.cpp @@ -196,21 +196,21 @@ TEST_F(SubscriptionThreadDispatcherTest, MaxActiveReadersIs20) { EXPECT_EQ(maxAc TEST_F(SubscriptionThreadDispatcherTest, SetAudioCallbackStoresRegistration) { SubscriptionThreadDispatcher dispatcher; - dispatcher.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); + EXPECT_TRUE(dispatcher.trySetOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {})); EXPECT_EQ(audioCallbacks(dispatcher).size(), 1u); } TEST_F(SubscriptionThreadDispatcherTest, SetVideoCallbackStoresRegistration) { SubscriptionThreadDispatcher dispatcher; - dispatcher.setOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); + EXPECT_TRUE(dispatcher.trySetOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {})); EXPECT_EQ(videoCallbacks(dispatcher).size(), 1u); } TEST_F(SubscriptionThreadDispatcherTest, ClearAudioCallbackRemovesRegistration) { SubscriptionThreadDispatcher dispatcher; - dispatcher.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); + (void)dispatcher.trySetOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); ASSERT_EQ(audioCallbacks(dispatcher).size(), 1u); dispatcher.clearOnAudioFrameCallback("alice", "mic-main"); @@ -219,7 +219,7 @@ TEST_F(SubscriptionThreadDispatcherTest, ClearAudioCallbackRemovesRegistration) TEST_F(SubscriptionThreadDispatcherTest, ClearVideoCallbackRemovesRegistration) { SubscriptionThreadDispatcher dispatcher; - dispatcher.setOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); + (void)dispatcher.trySetOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); ASSERT_EQ(videoCallbacks(dispatcher).size(), 1u); dispatcher.clearOnVideoFrameCallback("alice", "cam-main"); @@ -237,26 +237,26 @@ TEST_F(SubscriptionThreadDispatcherTest, OverwriteAudioCallbackKeepsSingleEntry) std::atomic counter1{0}; std::atomic counter2{0}; - dispatcher.setOnAudioFrameCallback("alice", "mic-main", [&counter1](const AudioFrame&) { counter1++; }); - dispatcher.setOnAudioFrameCallback("alice", "mic-main", [&counter2](const AudioFrame&) { counter2++; }); + (void)dispatcher.trySetOnAudioFrameCallback("alice", "mic-main", [&counter1](const AudioFrame&) { counter1++; }); + (void)dispatcher.trySetOnAudioFrameCallback("alice", "mic-main", [&counter2](const AudioFrame&) { counter2++; }); EXPECT_EQ(audioCallbacks(dispatcher).size(), 1u) << "Re-registering with the same key should overwrite, not add"; } TEST_F(SubscriptionThreadDispatcherTest, OverwriteVideoCallbackKeepsSingleEntry) { SubscriptionThreadDispatcher dispatcher; - dispatcher.setOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); - dispatcher.setOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); + (void)dispatcher.trySetOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); + (void)dispatcher.trySetOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); EXPECT_EQ(videoCallbacks(dispatcher).size(), 1u); } TEST_F(SubscriptionThreadDispatcherTest, MultipleDistinctCallbacksAreIndependent) { SubscriptionThreadDispatcher dispatcher; - dispatcher.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); - dispatcher.setOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); - dispatcher.setOnAudioFrameCallback("bob", "mic-main", [](const AudioFrame&) {}); - dispatcher.setOnVideoFrameCallback("bob", "cam-main", [](const VideoFrame&, std::int64_t) {}); + (void)dispatcher.trySetOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); + (void)dispatcher.trySetOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); + (void)dispatcher.trySetOnAudioFrameCallback("bob", "mic-main", [](const AudioFrame&) {}); + (void)dispatcher.trySetOnVideoFrameCallback("bob", "cam-main", [](const VideoFrame&, std::int64_t) {}); EXPECT_EQ(audioCallbacks(dispatcher).size(), 2u); EXPECT_EQ(videoCallbacks(dispatcher).size(), 2u); @@ -268,8 +268,8 @@ TEST_F(SubscriptionThreadDispatcherTest, MultipleDistinctCallbacksAreIndependent TEST_F(SubscriptionThreadDispatcherTest, ClearingOneTrackNameDoesNotAffectOther) { SubscriptionThreadDispatcher dispatcher; - dispatcher.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); - dispatcher.setOnAudioFrameCallback("alice", "screenshare-main", [](const AudioFrame&) {}); + (void)dispatcher.trySetOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); + (void)dispatcher.trySetOnAudioFrameCallback("alice", "screenshare-main", [](const AudioFrame&) {}); ASSERT_EQ(audioCallbacks(dispatcher).size(), 2u); dispatcher.clearOnAudioFrameCallback("alice", "mic-main"); @@ -290,7 +290,7 @@ TEST_F(SubscriptionThreadDispatcherTest, NoActiveReadersInitially) { TEST_F(SubscriptionThreadDispatcherTest, ActiveReadersEmptyAfterCallbackRegistration) { SubscriptionThreadDispatcher dispatcher; - dispatcher.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); + (void)dispatcher.trySetOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); EXPECT_TRUE(activeReaders(dispatcher).empty()) << "Registering a callback without a subscribed track should not spawn " "readers"; @@ -303,15 +303,15 @@ TEST_F(SubscriptionThreadDispatcherTest, ActiveReadersEmptyAfterCallbackRegistra TEST_F(SubscriptionThreadDispatcherTest, DestroyDispatcherWithRegisteredCallbacksIsSafe) { EXPECT_NO_THROW({ SubscriptionThreadDispatcher dispatcher; - dispatcher.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); - dispatcher.setOnVideoFrameCallback("bob", "cam-main", [](const VideoFrame&, std::int64_t) {}); + (void)dispatcher.trySetOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); + (void)dispatcher.trySetOnVideoFrameCallback("bob", "cam-main", [](const VideoFrame&, std::int64_t) {}); }); } TEST_F(SubscriptionThreadDispatcherTest, DestroyDispatcherAfterClearingCallbacksIsSafe) { EXPECT_NO_THROW({ SubscriptionThreadDispatcher dispatcher; - dispatcher.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); + (void)dispatcher.trySetOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); dispatcher.clearOnAudioFrameCallback("alice", "mic-main"); }); } @@ -332,7 +332,7 @@ TEST_F(SubscriptionThreadDispatcherTest, ConcurrentRegistrationDoesNotCrash) { threads.emplace_back([&dispatcher, t, kIterations]() { for (int i = 0; i < kIterations; ++i) { const std::string id = "participant-" + std::to_string(t); - dispatcher.setOnAudioFrameCallback(id, "mic-main", [](const AudioFrame&) {}); + (void)dispatcher.trySetOnAudioFrameCallback(id, "mic-main", [](const AudioFrame&) {}); dispatcher.clearOnAudioFrameCallback(id, "mic-main"); } }); @@ -357,8 +357,8 @@ TEST_F(SubscriptionThreadDispatcherTest, ConcurrentMixedAudioVideoRegistration) threads.emplace_back([&dispatcher, t, kIterations]() { const std::string id = "p-" + std::to_string(t); for (int i = 0; i < kIterations; ++i) { - dispatcher.setOnAudioFrameCallback(id, "mic-main", [](const AudioFrame&) {}); - dispatcher.setOnVideoFrameCallback(id, "cam-main", [](const VideoFrame&, std::int64_t) {}); + (void)dispatcher.trySetOnAudioFrameCallback(id, "mic-main", [](const AudioFrame&) {}); + (void)dispatcher.trySetOnVideoFrameCallback(id, "cam-main", [](const VideoFrame&, std::int64_t) {}); } }); } @@ -380,7 +380,8 @@ TEST_F(SubscriptionThreadDispatcherTest, ManyDistinctCallbacksCanBeRegistered) { constexpr int kCount = 50; for (int i = 0; i < kCount; ++i) { - dispatcher.setOnAudioFrameCallback("participant-" + std::to_string(i), "mic-main", [](const AudioFrame&) {}); + (void)dispatcher.trySetOnAudioFrameCallback("participant-" + std::to_string(i), "mic-main", + [](const AudioFrame&) {}); } EXPECT_EQ(audioCallbacks(dispatcher).size(), static_cast(kCount)); @@ -597,7 +598,7 @@ TEST_F(SubscriptionThreadDispatcherTest, EraseDataReaderIfCurrentLeavesReplacedE TEST_F(SubscriptionThreadDispatcherTest, DuplicateSubscribeWithSameAudioSidDoesNotRestartReader) { SubscriptionThreadDispatcher dispatcher; - dispatcher.setOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {}); + (void)dispatcher.trySetOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {}); // Simulate an already-running reader for this subscription. const CallbackKey key{"alice", "mic"}; @@ -616,7 +617,7 @@ TEST_F(SubscriptionThreadDispatcherTest, DuplicateSubscribeWithSameAudioSidDoesN TEST_F(SubscriptionThreadDispatcherTest, DuplicateSubscribeWithSameVideoSidDoesNotRestartReader) { SubscriptionThreadDispatcher dispatcher; - dispatcher.setOnVideoFrameCallback("alice", "cam", [](const VideoFrame&, std::int64_t) {}); + (void)dispatcher.trySetOnVideoFrameCallback("alice", "cam", [](const VideoFrame&, std::int64_t) {}); const CallbackKey key{"alice", "cam"}; activeReaders(dispatcher)[key].track_sid = "TR_video_1"; @@ -630,6 +631,61 @@ TEST_F(SubscriptionThreadDispatcherTest, DuplicateSubscribeWithSameVideoSidDoesN EXPECT_EQ(activeReaders(dispatcher)[key].video_stream, nullptr) << "Reader must not have been rebuilt"; } +// ============================================================================ +// trySetOn* replacement semantics: registration is rejected while a reader is +// active; clearing first allows re-registration. +// ============================================================================ + +TEST_F(SubscriptionThreadDispatcherTest, TrySetOnAudioWhileReaderActiveIsRejected) { + SubscriptionThreadDispatcher dispatcher; + ASSERT_TRUE(dispatcher.trySetOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {})); + + // Simulate an already-running reader for this subscription. + const CallbackKey key{"alice", "mic"}; + activeReaders(dispatcher)[key].track_sid = "TR_audio_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + EXPECT_FALSE(dispatcher.trySetOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {})) + << "Replacing a callback while its reader is active must be rejected"; +} + +TEST_F(SubscriptionThreadDispatcherTest, TrySetOnVideoWhileReaderActiveIsRejected) { + SubscriptionThreadDispatcher dispatcher; + ASSERT_TRUE(dispatcher.trySetOnVideoFrameCallback("alice", "cam", [](const VideoFrame&, std::int64_t) {})); + + const CallbackKey key{"alice", "cam"}; + activeReaders(dispatcher)[key].track_sid = "TR_video_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + EXPECT_FALSE(dispatcher.trySetOnVideoFrameCallback("alice", "cam", [](const VideoFrame&, std::int64_t) {})); + EXPECT_FALSE(dispatcher.trySetOnVideoFrameEventCallback("alice", "cam", [](const VideoFrameEvent&) {})); +} + +TEST_F(SubscriptionThreadDispatcherTest, ClearThenTrySetOnAudioRegistersNewCallback) { + SubscriptionThreadDispatcher dispatcher; + ASSERT_TRUE(dispatcher.trySetOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {})); + + const CallbackKey key{"alice", "mic"}; + activeReaders(dispatcher)[key].track_sid = "TR_audio_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + // Rejected while active, accepted once the reader is cleared. + ASSERT_FALSE(dispatcher.trySetOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {})); + dispatcher.clearOnAudioFrameCallback("alice", "mic"); + EXPECT_EQ(activeReaderCount(dispatcher), 0u); + EXPECT_TRUE(dispatcher.trySetOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {})); + EXPECT_EQ(audioCallbacks(dispatcher).size(), 1u); +} + +TEST_F(SubscriptionThreadDispatcherTest, TrySetOnAudioWithoutActiveReaderOverwritesRegistration) { + SubscriptionThreadDispatcher dispatcher; + EXPECT_TRUE(dispatcher.trySetOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {})); + // No reader is active, so re-registering the same key is allowed and simply + // overwrites the stored callback. + EXPECT_TRUE(dispatcher.trySetOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {})); + EXPECT_EQ(audioCallbacks(dispatcher).size(), 1u); +} + // ============================================================================ // SID deduplication: data reader start is skipped for the same SID and // replaced (stopping the previous reader) for a new SID @@ -700,8 +756,8 @@ TEST_F(SubscriptionThreadDispatcherTest, DestroyDispatcherAfterRemovingDataCallb TEST_F(SubscriptionThreadDispatcherTest, MixedAudioVideoDataCallbacksAreIndependent) { SubscriptionThreadDispatcher dispatcher; - dispatcher.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); - dispatcher.setOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); + (void)dispatcher.trySetOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); + (void)dispatcher.trySetOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); dispatcher.addOnDataFrameCallback("alice", "data-track", [](const std::vector&, std::optional) {}); From 20e5d5b29b3f0025ca56e5d1824445bde7b30e4f Mon Sep 17 00:00:00 2001 From: Stephen DeRosa Date: Tue, 7 Jul 2026 16:06:42 -0600 Subject: [PATCH 05/11] trim --- README.md | 7 +++++ include/livekit/room.h | 26 ------------------- .../livekit/subscription_thread_dispatcher.h | 19 -------------- 3 files changed, 7 insertions(+), 45 deletions(-) diff --git a/README.md b/README.md index 7d4e1ccb..80dfafb8 100644 --- a/README.md +++ b/README.md @@ -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 + callback is already registered rather than silently replacing it. To replace + an active callback, call `clearOn*FrameCallback` first. (The same rename + applies to the corresponding `SubscriptionThreadDispatcher` methods.) ### `v1.0.0` diff --git a/include/livekit/room.h b/include/livekit/room.h index 8d47733c..c699e0b7 100644 --- a/include/livekit/room.h +++ b/include/livekit/room.h @@ -374,14 +374,6 @@ class LIVEKIT_API Room { /// @deprecated Use trySetOnAudioFrameCallback() instead. /// /// Forwards to @ref trySetOnAudioFrameCallback and discards the result. - /// Replacing an active callback is not supported through this overload; call - /// @ref clearOnAudioFrameCallback first, then @ref trySetOnAudioFrameCallback. - /// - /// @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. [[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 = {}); @@ -389,14 +381,6 @@ class LIVEKIT_API Room { /// @deprecated Use trySetOnVideoFrameCallback() instead. /// /// Forwards to @ref trySetOnVideoFrameCallback and discards the result. - /// Replacing an active callback is not supported through this overload; call - /// @ref clearOnVideoFrameCallback first, then @ref trySetOnVideoFrameCallback. - /// - /// @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. [[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 = {}); @@ -404,16 +388,6 @@ class LIVEKIT_API Room { /// @deprecated Use trySetOnVideoFrameEventCallback() instead. /// /// Forwards to @ref trySetOnVideoFrameEventCallback and discards the result. - /// Replacing an active callback is not supported through this overload; call - /// @ref clearOnVideoFrameCallback first, then - /// @ref trySetOnVideoFrameEventCallback. - /// - /// @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. [[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 = {}); diff --git a/include/livekit/subscription_thread_dispatcher.h b/include/livekit/subscription_thread_dispatcher.h index a4fabc15..63e47965 100644 --- a/include/livekit/subscription_thread_dispatcher.h +++ b/include/livekit/subscription_thread_dispatcher.h @@ -160,12 +160,6 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// Forwards to @ref trySetOnAudioFrameCallback and discards the result. /// Replacing an active callback is not supported through this overload; call /// @ref clearOnAudioFrameCallback first, then @ref trySetOnAudioFrameCallback. - /// - /// @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. [[deprecated( "SubscriptionThreadDispatcher::setOnAudioFrameCallback is deprecated; use trySetOnAudioFrameCallback instead")]] void setOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name, @@ -176,12 +170,6 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// Forwards to @ref trySetOnVideoFrameCallback and discards the result. /// Replacing an active callback is not supported through this overload; call /// @ref clearOnVideoFrameCallback first, then @ref trySetOnVideoFrameCallback. - /// - /// @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. [[deprecated( "SubscriptionThreadDispatcher::setOnVideoFrameCallback is deprecated; use trySetOnVideoFrameCallback instead")]] void setOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name, @@ -193,13 +181,6 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// Replacing an active callback is not supported through this overload; call /// @ref clearOnVideoFrameCallback first, then /// @ref trySetOnVideoFrameEventCallback. - /// - /// @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. [[deprecated( "SubscriptionThreadDispatcher::setOnVideoFrameEventCallback is deprecated; use " "trySetOnVideoFrameEventCallback instead")]] From 6053b278e392ec4fdfe81ee48dbd0ca828584793 Mon Sep 17 00:00:00 2001 From: Stephen DeRosa Date: Tue, 7 Jul 2026 16:30:21 -0600 Subject: [PATCH 06/11] fix room logic --- src/room.cpp | 48 +++++++++++++++++++++++++++++++++--------------- 1 file changed, 33 insertions(+), 15 deletions(-) diff --git a/src/room.cpp b/src/room.cpp index 61f45a6e..03373baf 100644 --- a/src/room.cpp +++ b/src/room.cpp @@ -396,13 +396,19 @@ bool Room::trySetOnAudioFrameCallback(const std::string& participant_identity, c std::move(callback), opts)) { return false; } + + // If we've already subscribed to the track, handle it immediately auto track = findSubscribedRemoteTrack(participant_identity, track_name); - if (!track) { - LK_LOG_ERROR("Room::trySetOnAudioFrameCallback: track not found for participant={} track_name={}", - participant_identity, track_name); - return false; + if (track) { + subscription_thread_dispatcher_->handleTrackSubscribed(participant_identity, track_name, track); + } else { + // The track is not subscribed yet. The callback is registered; the reader + // starts when the track is subscribed (see kTrackSubscribed in onEvent). + LK_LOG_DEBUG( + "Room::trySetOnAudioFrameCallback: track not yet subscribed for participant={} track_name={}; " + "callback registered for deferred start", + participant_identity, track_name); } - subscription_thread_dispatcher_->handleTrackSubscribed(participant_identity, track_name, track); return true; } @@ -416,13 +422,19 @@ bool Room::trySetOnVideoFrameCallback(const std::string& participant_identity, c std::move(callback), opts)) { return false; } + + // If we've already subscribed to the track, handle it immediately auto track = findSubscribedRemoteTrack(participant_identity, track_name); - if (!track) { - LK_LOG_ERROR("Room::trySetOnVideoFrameCallback: track not found for participant={} track_name={}", - participant_identity, track_name); - return false; + if (track) { + subscription_thread_dispatcher_->handleTrackSubscribed(participant_identity, track_name, track); + } else { + // The track is not subscribed yet. The callback is registered; the reader + // starts when the track is subscribed (see kTrackSubscribed in onEvent). + LK_LOG_DEBUG( + "Room::trySetOnVideoFrameCallback: track not yet subscribed for participant={} track_name={}; " + "callback registered for deferred start", + participant_identity, track_name); } - subscription_thread_dispatcher_->handleTrackSubscribed(participant_identity, track_name, track); return true; } @@ -436,13 +448,19 @@ bool Room::trySetOnVideoFrameEventCallback(const std::string& participant_identi std::move(callback), opts)) { return false; } + + // If we've already subscribed to the track, handle it immediately auto track = findSubscribedRemoteTrack(participant_identity, track_name); - if (!track) { - LK_LOG_ERROR("Room::trySetOnVideoFrameEventCallback: track not found for participant={} track_name={}", - participant_identity, track_name); - return false; + if (track) { + subscription_thread_dispatcher_->handleTrackSubscribed(participant_identity, track_name, track); + } else { + // The track is not subscribed yet. The callback is registered; the reader + // starts when the track is subscribed (see kTrackSubscribed in onEvent). + LK_LOG_DEBUG( + "Room::trySetOnVideoFrameEventCallback: track not yet subscribed for participant={} track_name={}; " + "callback registered for deferred start", + participant_identity, track_name); } - subscription_thread_dispatcher_->handleTrackSubscribed(participant_identity, track_name, track); return true; } From 22991ec562f245b10beb856b96fda2ac11b69150 Mon Sep 17 00:00:00 2001 From: Stephen DeRosa Date: Tue, 7 Jul 2026 16:36:09 -0600 Subject: [PATCH 07/11] log in old functions --- src/room.cpp | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/src/room.cpp b/src/room.cpp index 03373baf..a6eb8b63 100644 --- a/src/room.cpp +++ b/src/room.cpp @@ -466,17 +466,29 @@ bool Room::trySetOnVideoFrameEventCallback(const std::string& participant_identi void Room::setOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name, AudioFrameCallback callback, const AudioStream::Options& opts) { - (void)trySetOnAudioFrameCallback(participant_identity, track_name, std::move(callback), opts); + bool result = trySetOnAudioFrameCallback(participant_identity, track_name, std::move(callback), opts); + if (!result) { + LK_LOG_ERROR("Room::setOnAudioFrameCallback: failed to set callback for participant={} track_name={}", + participant_identity, track_name); + } } void Room::setOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name, VideoFrameCallback callback, const VideoStream::Options& opts) { - (void)trySetOnVideoFrameCallback(participant_identity, track_name, std::move(callback), opts); + bool result = trySetOnVideoFrameCallback(participant_identity, track_name, std::move(callback), opts); + if (!result) { + LK_LOG_ERROR("Room::setOnVideoFrameCallback: failed to set callback for participant={} track_name={}", + participant_identity, track_name); + } } void Room::setOnVideoFrameEventCallback(const std::string& participant_identity, const std::string& track_name, VideoFrameEventCallback callback, const VideoStream::Options& opts) { - (void)trySetOnVideoFrameEventCallback(participant_identity, track_name, std::move(callback), opts); + bool result = trySetOnVideoFrameEventCallback(participant_identity, track_name, std::move(callback), opts); + if (!result) { + LK_LOG_ERROR("Room::setOnVideoFrameEventCallback: failed to set callback for participant={} track_name={}", + participant_identity, track_name); + } } void Room::clearOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name) { From 1de44c88a1a9faf3137fbebfd6be47afad20c5de Mon Sep 17 00:00:00 2001 From: Stephen DeRosa Date: Tue, 7 Jul 2026 16:47:09 -0600 Subject: [PATCH 08/11] doc clarity --- README.md | 6 +++--- include/livekit/subscription_thread_dispatcher.h | 5 +++-- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 80dfafb8..b040a83d 100644 --- a/README.md +++ b/README.md @@ -263,9 +263,9 @@ The following features are deprecated and will be removed in the next major rele - `Room::setOnAudioFrameCallback`, `Room::setOnVideoFrameCallback`, and `Room::setOnVideoFrameEventCallback` are deprecated. Use the `[[nodiscard]]` variants `trySetOnAudioFrameCallback`, `trySetOnVideoFrameCallback`, and - `trySetOnVideoFrameEventCallback` instead, which return `false` when a - callback is already registered rather than silently replacing it. To replace - an active callback, call `clearOn*FrameCallback` first. (The same rename + `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` diff --git a/include/livekit/subscription_thread_dispatcher.h b/include/livekit/subscription_thread_dispatcher.h index 63e47965..a42a8612 100644 --- a/include/livekit/subscription_thread_dispatcher.h +++ b/include/livekit/subscription_thread_dispatcher.h @@ -384,8 +384,9 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// Select the appropriate reader startup path for @p media track. /// - /// This is called by @ref Room when a remote track is subscribed. If a reader is already active for the given key, - /// std::thread default constructed is returned. + /// This is called by @ref Room when a remote track is subscribed. If a reader for the same track SID is already + /// active, startup is skipped and a default-constructed thread is returned; otherwise any previous reader is + /// extracted and returned to the caller for joining outside the lock. /// /// Must be called with @ref lock_ held. std::thread startReaderLocked(const CallbackKey& key, const std::shared_ptr& track); From 4618ea946bce5f4a633b13aaabd57b9813677a49 Mon Sep 17 00:00:00 2001 From: Stephen DeRosa Date: Tue, 7 Jul 2026 16:52:27 -0600 Subject: [PATCH 09/11] better locking --- src/subscription_thread_dispatcher.cpp | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/src/subscription_thread_dispatcher.cpp b/src/subscription_thread_dispatcher.cpp index 651d94c1..e4dfa656 100644 --- a/src/subscription_thread_dispatcher.cpp +++ b/src/subscription_thread_dispatcher.cpp @@ -668,16 +668,25 @@ std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbac stream = subscribe_result.value(); LK_LOG_INFO("Data reader thread: subscribed to \"{}\" track=\"{}\"", identity, track_name); + bool cancelled = false; { const std::scoped_lock guard(reader->sub_mutex); // A replacement or teardown may have cancelled this reader while the - // subscribe was in flight. Close the fresh stream and bail so we do not - // leave a second live subscription behind. + // subscribe was in flight. Close the fresh stream so we do not leave a + // second live subscription behind. if (reader->cancelled.load()) { + cancelled = true; stream->close(); - return; + } else { + reader->stream = stream; } - reader->stream = stream; + } + if (cancelled) { + // Mirror the normal-exit cleanup below. Done outside sub_mutex to keep the + // lock_ -> sub_mutex order and avoid inversion; a no-op unless this reader + // still owns its slot. + eraseDataReaderIfCurrent(id, reader); + return; } DataTrackFrame frame; From d3ce7a99e92afee51e1d793b921b6ef84f9e8250 Mon Sep 17 00:00:00 2001 From: Stephen DeRosa Date: Tue, 7 Jul 2026 17:13:50 -0600 Subject: [PATCH 10/11] const --- src/room.cpp | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/room.cpp b/src/room.cpp index a6eb8b63..cb08ed8a 100644 --- a/src/room.cpp +++ b/src/room.cpp @@ -466,7 +466,7 @@ bool Room::trySetOnVideoFrameEventCallback(const std::string& participant_identi void Room::setOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name, AudioFrameCallback callback, const AudioStream::Options& opts) { - bool result = trySetOnAudioFrameCallback(participant_identity, track_name, std::move(callback), opts); + bool const result = trySetOnAudioFrameCallback(participant_identity, track_name, std::move(callback), opts); if (!result) { LK_LOG_ERROR("Room::setOnAudioFrameCallback: failed to set callback for participant={} track_name={}", participant_identity, track_name); @@ -475,7 +475,7 @@ void Room::setOnAudioFrameCallback(const std::string& participant_identity, cons void Room::setOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name, VideoFrameCallback callback, const VideoStream::Options& opts) { - bool result = trySetOnVideoFrameCallback(participant_identity, track_name, std::move(callback), opts); + bool const result = trySetOnVideoFrameCallback(participant_identity, track_name, std::move(callback), opts); if (!result) { LK_LOG_ERROR("Room::setOnVideoFrameCallback: failed to set callback for participant={} track_name={}", participant_identity, track_name); @@ -484,7 +484,7 @@ void Room::setOnVideoFrameCallback(const std::string& participant_identity, cons void Room::setOnVideoFrameEventCallback(const std::string& participant_identity, const std::string& track_name, VideoFrameEventCallback callback, const VideoStream::Options& opts) { - bool result = trySetOnVideoFrameEventCallback(participant_identity, track_name, std::move(callback), opts); + bool const result = trySetOnVideoFrameEventCallback(participant_identity, track_name, std::move(callback), opts); if (!result) { LK_LOG_ERROR("Room::setOnVideoFrameEventCallback: failed to set callback for participant={} track_name={}", participant_identity, track_name); From ede1f209bf08baddbf81a7039ec26c8340bdde29 Mon Sep 17 00:00:00 2001 From: Stephen DeRosa Date: Tue, 7 Jul 2026 18:01:23 -0600 Subject: [PATCH 11/11] fix thread detaching --- .../livekit/subscription_thread_dispatcher.h | 13 ++- src/subscription_thread_dispatcher.cpp | 26 ++--- src/tests/integration/test_data_track.cpp | 5 +- .../test_subscription_thread_dispatcher.cpp | 110 ++++++++++++++++-- 4 files changed, 122 insertions(+), 32 deletions(-) diff --git a/include/livekit/subscription_thread_dispatcher.h b/include/livekit/subscription_thread_dispatcher.h index a42a8612..a1cf9db6 100644 --- a/include/livekit/subscription_thread_dispatcher.h +++ b/include/livekit/subscription_thread_dispatcher.h @@ -358,6 +358,9 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// Set true when this reader is being replaced or torn down so the reader /// thread can abort a subscription that is still in flight. std::atomic cancelled{false}; + /// Guarded by lock_. Reader threads may mark themselves finished, but only + /// dispatcher lifecycle paths erase the slot and join the thread. + bool finished = false; std::mutex sub_mutex; std::shared_ptr stream; // guarded by sub_mutex std::thread thread; @@ -415,11 +418,11 @@ class LIVEKIT_API SubscriptionThreadDispatcher { std::thread startDataReaderLocked(DataFrameCallbackId id, const DataCallbackKey& key, const std::shared_ptr& track, const DataFrameCallback& cb); - /// Remove @p reader from @ref active_data_readers_ if the slot for @p id - /// still refers to it. Called by the reader thread itself when it exits - /// after a failed or cancelled subscription so it does not leave a stale - /// entry behind. Acquires @ref lock_. - void eraseDataReaderIfCurrent(DataFrameCallbackId id, const std::shared_ptr& reader); + /// Mark @p reader finished if the slot for @p id still refers to it. + /// Called by the reader thread itself when it exits after a failed, + /// cancelled, or terminal subscription. Acquires @ref lock_. Reader threads + /// must not erase, detach, or join their own @ref std::thread. + void markDataReaderFinishedIfCurrent(DataFrameCallbackId id, const std::shared_ptr& reader); /// Protects callback registration maps and active reader state. mutable std::mutex lock_; diff --git a/src/subscription_thread_dispatcher.cpp b/src/subscription_thread_dispatcher.cpp index e4dfa656..4bf412a9 100644 --- a/src/subscription_thread_dispatcher.cpp +++ b/src/subscription_thread_dispatcher.cpp @@ -604,28 +604,26 @@ std::thread SubscriptionThreadDispatcher::extractDataReaderThreadLocked(DataFram return std::move(reader->thread); } -void SubscriptionThreadDispatcher::eraseDataReaderIfCurrent(DataFrameCallbackId id, - const std::shared_ptr& reader) { +void SubscriptionThreadDispatcher::markDataReaderFinishedIfCurrent(DataFrameCallbackId id, + const std::shared_ptr& reader) { const std::scoped_lock lock(lock_); auto it = active_data_readers_.find(id); if (it == active_data_readers_.end() || it->second != reader) { // The slot was already extracted or replaced; the owner joins that thread. return; } - // Detach so the reader can be destroyed by its own thread's final shared_ptr - // release without tripping std::thread's joinable-at-destruction check. The - // thread is already returning when this runs, so nothing is left to join. - if (reader->thread.joinable()) { - reader->thread.detach(); + reader->finished = true; + { + const std::scoped_lock guard(reader->sub_mutex); + reader->stream.reset(); } - active_data_readers_.erase(it); } std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbackId id, const DataCallbackKey& key, const std::shared_ptr& track, const DataFrameCallback& cb) { auto existing = active_data_readers_.find(id); - if (existing != active_data_readers_.end() && existing->second->remote_track && + if (existing != active_data_readers_.end() && !existing->second->finished && existing->second->remote_track && existing->second->remote_track->info().sid == track->info().sid) { LK_LOG_DEBUG( "Skipping data reader start for \"{}\" track=\"{}\" because a reader for " @@ -662,7 +660,7 @@ std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbac "Failed to subscribe to data track \"{}\" from \"{}\": code={} " "message={}", track_name, identity, static_cast(error.code), error.message); - eraseDataReaderIfCurrent(id, reader); + markDataReaderFinishedIfCurrent(id, reader); return; } stream = subscribe_result.value(); @@ -685,7 +683,7 @@ std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbac // Mirror the normal-exit cleanup below. Done outside sub_mutex to keep the // lock_ -> sub_mutex order and avoid inversion; a no-op unless this reader // still owns its slot. - eraseDataReaderIfCurrent(id, reader); + markDataReaderFinishedIfCurrent(id, reader); return; } @@ -704,9 +702,9 @@ std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbac "\"{}\": code={} message={}", track_name, identity, static_cast(error->code), error->message); } - // Clean our own slot if the stream ended on its own (server EOS) and no - // extract/teardown already claimed it. A no-op when we were extracted. - eraseDataReaderIfCurrent(id, reader); + // Mark our own slot finished if the stream ended on its own (server EOS) + // and no extract/teardown already claimed it. A no-op when we were extracted. + markDataReaderFinishedIfCurrent(id, reader); LK_LOG_INFO("Data reader thread exiting for \"{}\" track=\"{}\"", identity, track_name); }); // NOLINTEND(bugprone-lambda-function-name) diff --git a/src/tests/integration/test_data_track.cpp b/src/tests/integration/test_data_track.cpp index 8663218f..59170727 100644 --- a/src/tests/integration/test_data_track.cpp +++ b/src/tests/integration/test_data_track.cpp @@ -543,7 +543,8 @@ TEST_F(DataTrackE2ETest, RepublishRewiresDataCallbackToNewPublication) { ASSERT_NE(second_track, nullptr) << "Failed to republish data track"; auto remotes = subscriber_delegate.waitForTracks(2, kTrackWaitTimeout); ASSERT_EQ(remotes.size(), 2u) << "Timed out waiting for republished remote data track"; - const std::string second_sid = remotes.back()->info().sid; + auto second_remote = remotes.back(); + const std::string second_sid = second_remote->info().sid; EXPECT_NE(first_sid, second_sid) << "Republish should produce a new SID"; DataTrackFrame second_frame; @@ -562,6 +563,8 @@ TEST_F(DataTrackE2ETest, RepublishRewiresDataCallbackToNewPublication) { } second_track->unpublishDataTrack(); + ASSERT_TRUE(waitForCondition([&]() { return !second_remote->isPublished(); }, kTrackWaitTimeout)) + << "Second remote track did not report unpublished state"; } TEST_F(DataTrackE2ETest, SubscribeAfterUnpublishReportsTerminalError) { diff --git a/src/tests/unit/test_subscription_thread_dispatcher.cpp b/src/tests/unit/test_subscription_thread_dispatcher.cpp index dc56f2f5..d2a34cd1 100644 --- a/src/tests/unit/test_subscription_thread_dispatcher.cpp +++ b/src/tests/unit/test_subscription_thread_dispatcher.cpp @@ -95,9 +95,9 @@ class SubscriptionThreadDispatcherTest : public ::testing::Test { return dispatcher.extractDataReaderThreadLocked(id); } - static void eraseDataReaderIfCurrent(SubscriptionThreadDispatcher& dispatcher, DataFrameCallbackId id, - const std::shared_ptr& reader) { - dispatcher.eraseDataReaderIfCurrent(id, reader); + static void markDataReaderFinishedIfCurrent(SubscriptionThreadDispatcher& dispatcher, DataFrameCallbackId id, + const std::shared_ptr& reader) { + dispatcher.markDataReaderFinishedIfCurrent(id, reader); } static std::thread startDataReader(SubscriptionThreadDispatcher& dispatcher, DataFrameCallbackId id, @@ -542,12 +542,13 @@ TEST_F(SubscriptionThreadDispatcherTest, NoRemoteDataTracksInitially) { } // ============================================================================ -// Data reader replacement: cancellation and self-erase +// Data reader replacement: cancellation and finished-state ownership // ============================================================================ TEST_F(SubscriptionThreadDispatcherTest, ActiveDataReaderNotCancelledByDefault) { auto reader = std::make_shared(); EXPECT_FALSE(reader->cancelled.load()); + EXPECT_FALSE(reader->finished); } TEST_F(SubscriptionThreadDispatcherTest, ExtractDataReaderMarksCancelledAndRemovesEntry) { @@ -568,28 +569,46 @@ TEST_F(SubscriptionThreadDispatcherTest, ExtractMissingDataReaderIsNoOp) { EXPECT_FALSE(extracted.joinable()); } -TEST_F(SubscriptionThreadDispatcherTest, EraseDataReaderIfCurrentRemovesMatchingEntry) { +TEST_F(SubscriptionThreadDispatcherTest, MarkDataReaderFinishedIfCurrentKeepsMatchingEntry) { SubscriptionThreadDispatcher dispatcher; auto reader = std::make_shared(); activeDataReaders(dispatcher)[0] = reader; - eraseDataReaderIfCurrent(dispatcher, 0, reader); + markDataReaderFinishedIfCurrent(dispatcher, 0, reader); - EXPECT_TRUE(activeDataReaders(dispatcher).empty()); + ASSERT_EQ(activeDataReaders(dispatcher).size(), 1u); + EXPECT_EQ(activeDataReaders(dispatcher)[0], reader); + EXPECT_TRUE(reader->finished); } -TEST_F(SubscriptionThreadDispatcherTest, EraseDataReaderIfCurrentLeavesReplacedEntry) { +TEST_F(SubscriptionThreadDispatcherTest, MarkDataReaderFinishedIfCurrentLeavesReplacedEntry) { SubscriptionThreadDispatcher dispatcher; auto original = std::make_shared(); auto replacement = std::make_shared(); activeDataReaders(dispatcher)[0] = replacement; - // The original reader exited after being replaced; it must not evict the + // The original reader exited after being replaced; it must not mark the // newer reader that now owns the same callback id. - eraseDataReaderIfCurrent(dispatcher, 0, original); + markDataReaderFinishedIfCurrent(dispatcher, 0, original); ASSERT_EQ(activeDataReaders(dispatcher).size(), 1u); EXPECT_EQ(activeDataReaders(dispatcher)[0], replacement); + EXPECT_FALSE(replacement->finished); +} + +TEST_F(SubscriptionThreadDispatcherTest, ExtractFinishedDataReaderRemovesEntryAndReturnsJoinableThread) { + SubscriptionThreadDispatcher dispatcher; + auto reader = std::make_shared(); + reader->finished = true; + reader->thread = std::thread([]() {}); + activeDataReaders(dispatcher)[0] = reader; + + auto extracted = extractDataReader(dispatcher, 0); + + EXPECT_TRUE(reader->cancelled.load()); + EXPECT_TRUE(extracted.joinable()); + EXPECT_TRUE(activeDataReaders(dispatcher).empty()); + extracted.join(); } // ============================================================================ @@ -677,6 +696,44 @@ TEST_F(SubscriptionThreadDispatcherTest, ClearThenTrySetOnAudioRegistersNewCallb EXPECT_EQ(audioCallbacks(dispatcher).size(), 1u); } +TEST_F(SubscriptionThreadDispatcherTest, ClearThenDeprecatedSetOnAudioRegistersNewCallback) { + SubscriptionThreadDispatcher dispatcher; +#if defined(__clang__) +#pragma clang diagnostic push +#pragma clang diagnostic ignored "-Wdeprecated-declarations" +#elif defined(__GNUC__) +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wdeprecated-declarations" +#endif + dispatcher.setOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {}); +#if defined(__clang__) +#pragma clang diagnostic pop +#elif defined(__GNUC__) +#pragma GCC diagnostic pop +#endif + + const CallbackKey key{"alice", "mic"}; + activeReaders(dispatcher)[key].track_sid = "TR_audio_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + dispatcher.clearOnAudioFrameCallback("alice", "mic"); + EXPECT_EQ(activeReaderCount(dispatcher), 0u); +#if defined(__clang__) +#pragma clang diagnostic push +#pragma clang diagnostic ignored "-Wdeprecated-declarations" +#elif defined(__GNUC__) +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wdeprecated-declarations" +#endif + dispatcher.setOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {}); +#if defined(__clang__) +#pragma clang diagnostic pop +#elif defined(__GNUC__) +#pragma GCC diagnostic pop +#endif + EXPECT_EQ(audioCallbacks(dispatcher).size(), 1u); +} + TEST_F(SubscriptionThreadDispatcherTest, TrySetOnAudioWithoutActiveReaderOverwritesRegistration) { SubscriptionThreadDispatcher dispatcher; EXPECT_TRUE(dispatcher.trySetOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {})); @@ -706,6 +763,31 @@ TEST_F(SubscriptionThreadDispatcherTest, DuplicateDataPublishWithSameSidDoesNotR EXPECT_FALSE(reader->cancelled.load()) << "A skipped reader must not be cancelled"; } +TEST_F(SubscriptionThreadDispatcherTest, FinishedDataReaderWithSameSidIsReplaced) { + SubscriptionThreadDispatcher dispatcher; + auto reader = std::make_shared(); + reader->remote_track = RemoteDataTrackTestAccess::create({"foo", "TR_data_1", false}, "alice"); + reader->finished = true; + activeDataReaders(dispatcher)[7] = reader; + + auto incoming = RemoteDataTrackTestAccess::create({"foo", "TR_data_1", false}, "alice"); + auto old_thread = startDataReader(dispatcher, 7, DataCallbackKey{"alice", "foo"}, incoming); + if (old_thread.joinable()) { + old_thread.join(); + } + + EXPECT_TRUE(reader->cancelled.load()) << "Finished reader must be extracted before replacement"; + EXPECT_TRUE(waitFor( + [&] { + return activeDataReaderCount(dispatcher) == 1u && activeDataReaders(dispatcher)[7] != reader && + activeDataReaders(dispatcher)[7]->finished; + }, + 2s)); + + dispatcher.stopAll(); + EXPECT_TRUE(activeDataReaders(dispatcher).empty()); +} + TEST_F(SubscriptionThreadDispatcherTest, RepublishWithNewDataSidStopsPreviousReader) { SubscriptionThreadDispatcher dispatcher; auto previous = std::make_shared(); @@ -723,8 +805,12 @@ TEST_F(SubscriptionThreadDispatcherTest, RepublishWithNewDataSidStopsPreviousRea EXPECT_TRUE(previous->cancelled.load()) << "Previous reader must be cancelled on republish"; // 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 activeDataReaderCount(dispatcher) == 0; }, 2s)); + // fast and marks itself finished while the dispatcher keeps ownership. + EXPECT_TRUE(waitFor( + [&] { return activeDataReaderCount(dispatcher) == 1u && activeDataReaders(dispatcher)[7]->finished; }, 2s)); + + dispatcher.stopAll(); + EXPECT_TRUE(activeDataReaders(dispatcher).empty()); } // ============================================================================