From b2dd32709df59d59a26aa9adb730e7c2cbac9b15 Mon Sep 17 00:00:00 2001 From: Adrian Niculescu <15037449+adrian-niculescu@users.noreply.github.com> Date: Wed, 16 Sep 2026 14:13:26 +0300 Subject: [PATCH] Fixed data track streams dropping frames that arrive just before the end A collector that subscribed just before a short drain finished read the drained flag as set and emitted its own terminator ahead of the frames and terminator already buffered for it, so collection ended without them. The drain now shares its items with a replay of one: a collector receives the replayed item atomically with its subscription, which tells it whether a frame predates it or the stream had already ended. --- .changeset/fix-data-track-stream-end-race.md | 5 ++ .../android/room/datatrack/DataTrackStream.kt | 51 +++++++++++-------- 2 files changed, 35 insertions(+), 21 deletions(-) create mode 100644 .changeset/fix-data-track-stream-end-race.md diff --git a/.changeset/fix-data-track-stream-end-race.md b/.changeset/fix-data-track-stream-end-race.md new file mode 100644 index 00000000..3a4756eb --- /dev/null +++ b/.changeset/fix-data-track-stream-end-race.md @@ -0,0 +1,5 @@ +--- +"client-sdk-android": patch +--- + +Fixed `DataTrackStream.flow` sometimes completing without delivering the frames that arrived just before the stream ended. diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/datatrack/DataTrackStream.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/datatrack/DataTrackStream.kt index 6951d044..0ae1ad33 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/datatrack/DataTrackStream.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/datatrack/DataTrackStream.kt @@ -29,9 +29,10 @@ import kotlinx.coroutines.flow.filterNotNull import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.merge -import kotlinx.coroutines.flow.onSubscription import kotlinx.coroutines.flow.shareIn import kotlinx.coroutines.flow.takeWhile +import kotlinx.coroutines.flow.transform +import kotlinx.coroutines.flow.withIndex import io.livekit.uniffi.DataTrackStreamInterface as FfiDataTrackStream /** @@ -57,17 +58,8 @@ class DataTrackStream internal constructor( private val coroutineScope = CloseableCoroutineScope(dispatcher + SupervisorJob()) /** - * Set once the drain has run the underlying stream to exhaustion. - * - * This exists only for collectors that arrive too late to see the terminator [sharedFrames] - * emits. - */ - private val drained = MutableStateFlow(false) - - /** - * Set by [close]. Unlike [drained] this *is* merged into [flow], because closing cancels the - * drain mid-[next], so no terminator is ever emitted and nothing in the frame stream would - * complete collectors. + * Set by [close]. Merged into [flow] because closing cancels the drain mid-[next], so it never + * emits [DrainItem.Ended] and nothing in the frame stream would complete collectors. * * Racing undelivered frames is correct here, and is the documented difference between the two * paths: the caller asked to stop, so collection ends promptly rather than draining first. @@ -85,16 +77,21 @@ class DataTrackStream internal constructor( /** * Drains the underlying stream while anyone is collecting [flow], so every collector sees * every frame. + * + * The last item is replayed, and a collector receives it as part of subscribing, ahead of + * everything emitted afterwards. That is how a collector learns where it joined: a replayed + * [DrainItem.Frame] was emitted before it subscribed, and a replayed [DrainItem.Ended] means + * the stream had already ended. Every drain run starts with [DrainItem.Started], so a + * collector that finds nothing to replay receives that first rather than a frame. */ - private val sharedFrames: SharedFlow = flow { + private val drainItems: SharedFlow = flow { + emit(DrainItem.Started) while (true) { val frame = next() ?: break - emit(frame) + emit(DrainItem.Frame(frame)) } - // Before the terminator, so a collector that misses the terminator sees this instead. - drained.value = true - emit(null) - }.shareIn(coroutineScope, SharingStarted.WhileSubscribed(), replay = 0) + emit(DrainItem.Ended) + }.shareIn(coroutineScope, SharingStarted.WhileSubscribed(), replay = 1) /** * A [Flow] of incoming frames. Completes normally when the stream ends. @@ -105,9 +102,15 @@ class DataTrackStream internal constructor( * Completes when the stream is unpublished, or when [close] is called. */ val flow: Flow = merge( - // onSubscription runs after this collector holds a subscription but before it takes any - // value, so a `false` reading here means the terminator is still coming to it. - sharedFrames.onSubscription { if (drained.value) emit(null) }, + drainItems.withIndex().transform, DataTrackFrame?> { (index, item) -> + when (item) { + DrainItem.Started -> Unit + // A frame at index 0 is the replayed one, emitted before this collector subscribed. + is DrainItem.Frame -> if (index > 0) emit(item.frame) + // Arrives behind every frame emitted before it, so those are delivered first. + DrainItem.Ended -> emit(null) + } + }, closed.filter { it }.map { null }, ).takeWhile { it != null }.filterNotNull() @@ -127,4 +130,10 @@ class DataTrackStream internal constructor( coroutineScope.close() (impl as? AutoCloseable)?.close() } + + private sealed interface DrainItem { + object Started : DrainItem + class Frame(val frame: DataTrackFrame) : DrainItem + object Ended : DrainItem + } }