fix(audio-rooms): DROP_OLDEST on the SubscribeHandle pump (audit #8)
The re-issuing pump's MutableSharedFlow used the default onBufferOverflow = SUSPEND. A stalled consumer (decoder, player, or anything downstream) back-pressured the pump → the inner handle.objects.collect → the underlying transport → the relay. Effect: the room caches up to 64 frames (~1.3s) on the consumer side, then the relay slows down sending us audio. For Opus audio the desired behavior is the inverse: drop OLD frames so playback stays live after a UI hiccup or a foreground transition. Switch to BufferOverflow.DROP_OLDEST so the relay keeps streaming at full rate and the consumer's audio stays synchronized with the speaker, at the cost of dropped frames during stalls (which would have been late anyway).
This commit is contained in:
+11
-1
@@ -27,6 +27,7 @@ import com.vitorpamplona.nestsclient.transport.WebTransportFactory
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.channels.BufferOverflow
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||
@@ -250,7 +251,16 @@ private class ReconnectingHandle(
|
||||
activeListener.value
|
||||
?: error("no live session — wait for state == Connected before subscribing")
|
||||
|
||||
val frames = MutableSharedFlow<MoqObject>(extraBufferCapacity = SUBSCRIBE_BUFFER)
|
||||
// DROP_OLDEST so a stalled consumer doesn't back-pressure the
|
||||
// pump → underlying handle → relay. For Opus audio the user
|
||||
// wants playback to stay "live" rather than catch up on
|
||||
// minutes-stale buffered frames after a UI hiccup or
|
||||
// foreground/background transition.
|
||||
val frames =
|
||||
MutableSharedFlow<MoqObject>(
|
||||
extraBufferCapacity = SUBSCRIBE_BUFFER,
|
||||
onBufferOverflow = BufferOverflow.DROP_OLDEST,
|
||||
)
|
||||
val liveHandleRef = AtomicReference<SubscribeHandle?>(null)
|
||||
|
||||
// Re-subscribe pump: every time activeListener changes, drop
|
||||
|
||||
Reference in New Issue
Block a user