feat(nestsClient): MoQ session pump + subscribe API
Phase 3c-3 of the Clubhouse/nests integration. Promotes MoqSession
from a one-shot SETUP runner to a real concurrent session: the
control-stream and datagram pumps run on a caller-supplied scope
after the handshake, and a public subscribe()/unsubscribe() API
delivers per-track OBJECT_DATAGRAMs as a Flow<MoqObject>.
commonMain (nestsclient.moq):
- MoqSession now takes a `pumpScope: CoroutineScope` so callers can
bind the pumps to their lifecycle (test backgroundScope, viewmodel
scope, etc.).
- After setup() succeeds, two background coroutines start:
* Control-stream pump — buffers chunks across multiple writes,
decodes complete frames, dispatches SUBSCRIBE_OK / SUBSCRIBE_ERROR
to the matching CompletableDeferred, drops malformed frames
without killing the session.
* Datagram pump — decodes OBJECT_DATAGRAMs and routes each to the
matching per-track sink keyed by track_alias.
- `subscribe(namespace, trackName, priority, filter)` sends SUBSCRIBE,
awaits SUBSCRIBE_OK (throws MoqProtocolException on SUBSCRIBE_ERROR
with the publisher's reason), and returns a SubscribeHandle whose
`objects` flow emits inbound OBJECT_DATAGRAMs.
- `unsubscribe(subscribeId)` is idempotent; second call is a no-op.
close() cancels pumps + tears down all in-flight subscribes cleanly.
- Per-track sink uses a bounded Channel with DROP_OLDEST overflow
rather than MutableSharedFlow, so objects buffered between
subscribe() returning and the consumer attaching are preserved
(SharedFlow with replay=0 silently drops pre-subscription emissions).
transport (FakeWebTransport):
- Switched from `consumeAsFlow()` to `receiveAsFlow()` so a `.first()`
during setup followed by a long-running pump `.collect{}` works on
the same channel without the first call closing it.
Tests:
- Updated existing setup tests for the new `pumpScope` parameter.
- New `subscribe_completes_on_subscribe_ok_then_delivers_datagrams_in_order`
end-to-end: client subscribes, raw server peer (no MoqSession to
avoid pump-vs-test contention) replies SUBSCRIBE_OK + 3 datagrams,
client's flow yields all three in order with correct payloads.
- `subscribe_throws_MoqProtocolException_when_publisher_replies_with_subscribe_error`.
- `datagrams_for_unknown_track_alias_are_dropped_silently` ensures
the pump doesn't crash on orphan datagrams and the session stays
closeable.
This completes the pure-Kotlin MoQ work. Remaining 3.x phases all
touch real audio (MediaCodec Opus, AudioTrack, AudioRecord) or the
real WebTransport handshake (Phase 3b-2 with Kwik).
https://claude.ai/code/session_013nVLALALKaHVgHm9u5Cg8D
This commit is contained in:
Claude
+292
-34
@@ -22,7 +22,17 @@ package com.vitorpamplona.nestsclient.moq
|
||||
|
||||
import com.vitorpamplona.nestsclient.transport.WebTransportBidiStream
|
||||
import com.vitorpamplona.nestsclient.transport.WebTransportSession
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.channels.BufferOverflow
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.first
|
||||
import kotlinx.coroutines.flow.receiveAsFlow
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import kotlinx.coroutines.withTimeout
|
||||
|
||||
/**
|
||||
@@ -44,22 +54,25 @@ object MoqVersion {
|
||||
/**
|
||||
* Session wrapper over a [WebTransportSession] that speaks MoQ-transport.
|
||||
*
|
||||
* Phase 3c-1 ships only [setup] — the CLIENT_SETUP / SERVER_SETUP handshake.
|
||||
* Phase 3c-2 adds SUBSCRIBE + ANNOUNCE + object-stream multiplexing.
|
||||
*
|
||||
* Usage:
|
||||
* ```
|
||||
* val session = MoqSession.client(webTransport).apply {
|
||||
* setup(listOf(MoqVersion.DRAFT_17))
|
||||
* }
|
||||
* // session.selectedVersion is now the version the server picked.
|
||||
* ```
|
||||
* Lifecycle:
|
||||
* 1. [client] / [server] attaches to a transport. No traffic yet.
|
||||
* 2. [setup] runs the SETUP handshake synchronously. After it returns
|
||||
* successfully, two background pumps start in [pumpScope]:
|
||||
* * a control-stream pump that buffers chunks, decodes complete frames,
|
||||
* and dispatches each (currently SUBSCRIBE_OK / SUBSCRIBE_ERROR);
|
||||
* * a datagram pump that decodes OBJECT_DATAGRAMs and routes them to
|
||||
* the matching per-track flow.
|
||||
* 3. [subscribe] sends a SUBSCRIBE, awaits the OK, returns a
|
||||
* [SubscribeHandle] whose [SubscribeHandle.objects] flow yields every
|
||||
* OBJECT_DATAGRAM tagged with that subscription's track alias.
|
||||
* 4. [close] cancels the pumps and closes the transport.
|
||||
*/
|
||||
class MoqSession private constructor(
|
||||
private val transport: WebTransportSession,
|
||||
/** The single bidi stream every MoQ exchange uses for control. */
|
||||
private val controlStream: WebTransportBidiStream,
|
||||
private val role: Role,
|
||||
private val pumpScope: CoroutineScope,
|
||||
) {
|
||||
enum class Role { Client, Server }
|
||||
|
||||
@@ -71,15 +84,39 @@ class MoqSession private constructor(
|
||||
var serverParameters: List<SetupParameter> = emptyList()
|
||||
private set
|
||||
|
||||
private val stateMutex = Mutex()
|
||||
private val writeMutex = Mutex()
|
||||
private var nextSubscribeId = 0L
|
||||
|
||||
/** subscribe_id → CompletableDeferred awaiting SUBSCRIBE_OK / SUBSCRIBE_ERROR. */
|
||||
private val pendingSubscribes = HashMap<Long, CompletableDeferred<SubscribeOk>>()
|
||||
|
||||
/** track_alias → bounded channel that fans matching OBJECT_DATAGRAMs to the subscriber.
|
||||
*
|
||||
* We use a [Channel] with [BufferOverflow.DROP_OLDEST] (rather than a
|
||||
* [MutableSharedFlow]) because objects must be buffered between
|
||||
* [subscribe] returning and the caller attaching to [SubscribeHandle.objects],
|
||||
* even when no collector exists yet. SharedFlow with `replay=0` drops
|
||||
* pre-subscription emissions, which races real-time audio delivery.
|
||||
*/
|
||||
private val sinks = HashMap<Long, Channel<MoqObject>>()
|
||||
|
||||
/** subscribe_id → track_alias, so [unsubscribe] can clean up the matching sink. */
|
||||
private val aliasBySubscribeId = HashMap<Long, Long>()
|
||||
|
||||
private var controlPumpJob: Job? = null
|
||||
private var datagramPumpJob: Job? = null
|
||||
private var closed = false
|
||||
|
||||
/**
|
||||
* Run the SETUP handshake.
|
||||
*
|
||||
* Client side: writes CLIENT_SETUP with [supportedVersions] + [clientParameters],
|
||||
* then reads exactly one SERVER_SETUP, stores the result, and returns.
|
||||
* then reads exactly one SERVER_SETUP, stores the result, and starts the
|
||||
* background pumps.
|
||||
* Server side: reads exactly one CLIENT_SETUP, selects the first mutually
|
||||
* supported version from [supportedVersions] (where the local list is the
|
||||
* set of versions the server is willing to accept), writes SERVER_SETUP,
|
||||
* and returns.
|
||||
* supported version from [supportedVersions], writes SERVER_SETUP, and
|
||||
* starts the background pumps.
|
||||
*
|
||||
* @throws MoqProtocolException if the peer sent an unexpected message, or
|
||||
* if no version overlap exists (server side).
|
||||
@@ -95,6 +132,7 @@ class MoqSession private constructor(
|
||||
Role.Server -> runServerSetup(supportedVersions, clientParameters)
|
||||
}
|
||||
}
|
||||
startPumps()
|
||||
}
|
||||
|
||||
private suspend fun runClientSetup(
|
||||
@@ -102,7 +140,7 @@ class MoqSession private constructor(
|
||||
clientParameters: List<SetupParameter>,
|
||||
) {
|
||||
controlStream.write(MoqCodec.encode(ClientSetup(supportedVersions, clientParameters)))
|
||||
val reply = readOneMessage()
|
||||
val reply = readOneSetupMessage()
|
||||
val server =
|
||||
reply as? ServerSetup
|
||||
?: throw MoqProtocolException("expected SERVER_SETUP, got ${reply.type.name}")
|
||||
@@ -119,7 +157,7 @@ class MoqSession private constructor(
|
||||
acceptedVersions: List<Long>,
|
||||
serverParameters: List<SetupParameter>,
|
||||
) {
|
||||
val incoming = readOneMessage()
|
||||
val incoming = readOneSetupMessage()
|
||||
val client =
|
||||
incoming as? ClientSetup
|
||||
?: throw MoqProtocolException("expected CLIENT_SETUP, got ${incoming.type.name}")
|
||||
@@ -134,14 +172,12 @@ class MoqSession private constructor(
|
||||
}
|
||||
|
||||
/**
|
||||
* Read exactly one full MoQ message from the control stream.
|
||||
*
|
||||
* Phase 3c-1 assumes a whole message fits in a single transport write
|
||||
* (SETUP frames are only a handful of bytes and the peer always writes
|
||||
* them atomically). Phase 3c-2 will replace this with a buffer-and-retry
|
||||
* loop for messages that may fragment across chunks.
|
||||
* Bootstrap helper: read exactly one SETUP-phase message from the control
|
||||
* stream. Used only during the synchronous handshake (before the pump
|
||||
* starts). SETUP frames always fit in a single transport write, so we can
|
||||
* safely take the first chunk and decode.
|
||||
*/
|
||||
private suspend fun readOneMessage(): MoqMessage {
|
||||
private suspend fun readOneSetupMessage(): MoqMessage {
|
||||
val chunk = controlStream.incoming().first()
|
||||
val decoded =
|
||||
MoqCodec.decode(chunk)
|
||||
@@ -151,38 +187,260 @@ class MoqSession private constructor(
|
||||
return decoded.message
|
||||
}
|
||||
|
||||
/** Close the underlying transport. */
|
||||
// ---------------------------------------------------------------- Subscribe
|
||||
|
||||
/**
|
||||
* Send SUBSCRIBE and suspend until SUBSCRIBE_OK arrives. Returns a
|
||||
* [SubscribeHandle] whose [SubscribeHandle.objects] flow emits every
|
||||
* OBJECT_DATAGRAM tagged with this subscription's track alias.
|
||||
*
|
||||
* @throws MoqProtocolException if the publisher rejects the subscribe with
|
||||
* SUBSCRIBE_ERROR, or if the session is closed mid-flight.
|
||||
*/
|
||||
suspend fun subscribe(
|
||||
namespace: TrackNamespace,
|
||||
trackName: ByteArray,
|
||||
priority: Int = 0x80,
|
||||
filter: SubscribeFilter = SubscribeFilter.LatestGroup,
|
||||
objectBufferCapacity: Int = DEFAULT_OBJECT_BUFFER,
|
||||
): SubscribeHandle {
|
||||
check(!closed) { "session is closed" }
|
||||
|
||||
val subscribeId: Long
|
||||
val trackAlias: Long
|
||||
val deferred = CompletableDeferred<SubscribeOk>()
|
||||
val sink =
|
||||
Channel<MoqObject>(
|
||||
capacity = objectBufferCapacity,
|
||||
onBufferOverflow = BufferOverflow.DROP_OLDEST,
|
||||
)
|
||||
|
||||
stateMutex.withLock {
|
||||
check(!closed) { "session is closed" }
|
||||
subscribeId = nextSubscribeId++
|
||||
trackAlias = subscribeId // 1:1 alias = id keeps things simple
|
||||
pendingSubscribes[subscribeId] = deferred
|
||||
sinks[trackAlias] = sink
|
||||
aliasBySubscribeId[subscribeId] = trackAlias
|
||||
}
|
||||
|
||||
val frame =
|
||||
MoqCodec.encode(
|
||||
Subscribe(
|
||||
subscribeId = subscribeId,
|
||||
trackAlias = trackAlias,
|
||||
namespace = namespace,
|
||||
trackName = trackName,
|
||||
subscriberPriority = priority,
|
||||
filter = filter,
|
||||
),
|
||||
)
|
||||
try {
|
||||
writeMutex.withLock { controlStream.write(frame) }
|
||||
val ok = deferred.await()
|
||||
return SubscribeHandle(
|
||||
subscribeId = subscribeId,
|
||||
trackAlias = trackAlias,
|
||||
ok = ok,
|
||||
objects = sink.receiveAsFlow(),
|
||||
unsubscribeAction = { unsubscribe(subscribeId) },
|
||||
)
|
||||
} catch (t: Throwable) {
|
||||
// Roll back state on any failure (write error, deferred cancellation,
|
||||
// SUBSCRIBE_ERROR, session close).
|
||||
stateMutex.withLock {
|
||||
pendingSubscribes.remove(subscribeId)
|
||||
sinks.remove(trackAlias)?.close()
|
||||
aliasBySubscribeId.remove(subscribeId)
|
||||
}
|
||||
throw t
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Send UNSUBSCRIBE and tear down all local state for [subscribeId]. Safe
|
||||
* to call multiple times — second call is a no-op.
|
||||
*/
|
||||
suspend fun unsubscribe(subscribeId: Long) {
|
||||
val alias =
|
||||
stateMutex.withLock {
|
||||
pendingSubscribes.remove(subscribeId)?.cancel()
|
||||
aliasBySubscribeId.remove(subscribeId)?.also { sinks.remove(it)?.close() }
|
||||
} ?: return
|
||||
|
||||
if (closed) return
|
||||
runCatching {
|
||||
writeMutex.withLock { controlStream.write(MoqCodec.encode(Unsubscribe(subscribeId))) }
|
||||
}
|
||||
// Suppress write errors on unsubscribe — the session is being torn down
|
||||
// and the alias is already gone from local state. `alias` is intentionally
|
||||
// referenced so the compiler keeps the cleanup expression's side effects.
|
||||
@Suppress("UNUSED_VARIABLE")
|
||||
val ignored = alias
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------- Pumps
|
||||
|
||||
private fun startPumps() {
|
||||
controlPumpJob =
|
||||
pumpScope.launch {
|
||||
runControlPump()
|
||||
}
|
||||
datagramPumpJob =
|
||||
pumpScope.launch {
|
||||
runDatagramPump()
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun runControlPump() {
|
||||
var buffer = ByteArray(0)
|
||||
controlStream.incoming().collect { chunk ->
|
||||
buffer =
|
||||
if (buffer.isEmpty()) {
|
||||
chunk
|
||||
} else {
|
||||
val merged = ByteArray(buffer.size + chunk.size)
|
||||
buffer.copyInto(merged, 0)
|
||||
chunk.copyInto(merged, buffer.size)
|
||||
merged
|
||||
}
|
||||
while (true) {
|
||||
val decoded =
|
||||
try {
|
||||
MoqCodec.decode(buffer) ?: break
|
||||
} catch (e: MoqCodecException) {
|
||||
// Drop the corrupted buffer; keep the pump alive so the
|
||||
// next valid frame can recover the session.
|
||||
buffer = ByteArray(0)
|
||||
break
|
||||
}
|
||||
buffer = buffer.copyOfRange(decoded.bytesConsumed, buffer.size)
|
||||
dispatchControlMessage(decoded.message)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun dispatchControlMessage(msg: MoqMessage) {
|
||||
when (msg) {
|
||||
is SubscribeOk -> {
|
||||
val deferred =
|
||||
stateMutex.withLock { pendingSubscribes.remove(msg.subscribeId) }
|
||||
deferred?.complete(msg)
|
||||
}
|
||||
|
||||
is SubscribeError -> {
|
||||
val deferred =
|
||||
stateMutex.withLock {
|
||||
val d = pendingSubscribes.remove(msg.subscribeId)
|
||||
aliasBySubscribeId.remove(msg.subscribeId)?.also { sinks.remove(it)?.close() }
|
||||
d
|
||||
}
|
||||
deferred?.completeExceptionally(
|
||||
MoqProtocolException(
|
||||
"SUBSCRIBE rejected: code=${msg.errorCode} reason=${msg.reasonPhrase}",
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
else -> {
|
||||
// Other control messages (SETUP echoes, future ANNOUNCE/etc.)
|
||||
// are silently dropped at this layer; Phase 3c-3 only needs the
|
||||
// subscribe lifecycle.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun runDatagramPump() {
|
||||
transport.incomingDatagrams().collect { datagram ->
|
||||
val obj =
|
||||
try {
|
||||
MoqObjectDatagram.decode(datagram)
|
||||
} catch (e: MoqCodecException) {
|
||||
// Drop malformed datagrams — UDP-style transport, so partial
|
||||
// / corrupted packets are normal.
|
||||
return@collect
|
||||
}
|
||||
val sink = stateMutex.withLock { sinks[obj.trackAlias] }
|
||||
// trySend on a DROP_OLDEST channel always succeeds (drops oldest on
|
||||
// overflow). Returning value is intentionally ignored.
|
||||
sink?.trySend(obj)
|
||||
}
|
||||
}
|
||||
|
||||
/** Cancel the pumps and close the underlying transport. Idempotent. */
|
||||
suspend fun close(
|
||||
code: Int = 0,
|
||||
reason: String = "",
|
||||
) {
|
||||
controlStream.finish()
|
||||
transport.close(code, reason)
|
||||
stateMutex.withLock {
|
||||
if (closed) return
|
||||
closed = true
|
||||
// Fail any in-flight subscribe waiters cleanly.
|
||||
for ((_, deferred) in pendingSubscribes) {
|
||||
deferred.cancel()
|
||||
}
|
||||
pendingSubscribes.clear()
|
||||
for ((_, ch) in sinks) ch.close()
|
||||
sinks.clear()
|
||||
aliasBySubscribeId.clear()
|
||||
}
|
||||
controlPumpJob?.cancel()
|
||||
datagramPumpJob?.cancel()
|
||||
runCatching { writeMutex.withLock { controlStream.finish() } }
|
||||
runCatching { transport.close(code, reason) }
|
||||
}
|
||||
|
||||
companion object {
|
||||
/**
|
||||
* Attach to a [WebTransportSession] in the client role. Opens the
|
||||
* control stream eagerly so [setup] doesn't need to manage stream
|
||||
* acquisition.
|
||||
* Default per-subscription buffer for unread objects. 64 frames at
|
||||
* Opus 20 ms = ~1.3 s of audio backlog before DROP_OLDEST kicks in,
|
||||
* which matches a typical real-time listener's tolerance.
|
||||
*/
|
||||
suspend fun client(transport: WebTransportSession): MoqSession {
|
||||
const val DEFAULT_OBJECT_BUFFER: Int = 64
|
||||
|
||||
/**
|
||||
* Attach to a [WebTransportSession] in the client role.
|
||||
*
|
||||
* @param pumpScope where the post-handshake pumps live. Tests typically
|
||||
* pass `backgroundScope` from `runTest`; production code passes the
|
||||
* owning ViewModel scope so pumps are cancelled on screen exit.
|
||||
*/
|
||||
suspend fun client(
|
||||
transport: WebTransportSession,
|
||||
pumpScope: CoroutineScope,
|
||||
): MoqSession {
|
||||
val control = transport.openBidiStream()
|
||||
return MoqSession(transport, control, Role.Client)
|
||||
return MoqSession(transport, control, Role.Client, pumpScope)
|
||||
}
|
||||
|
||||
/**
|
||||
* Attach to a [WebTransportSession] in the server role over an
|
||||
* already-accepted control stream (usually the first bidi stream the
|
||||
* peer opened). Used in tests — a real server accepts the first bidi.
|
||||
* already-accepted control stream. Used in tests — a real server
|
||||
* accepts the client's first bidi.
|
||||
*/
|
||||
fun server(
|
||||
transport: WebTransportSession,
|
||||
control: WebTransportBidiStream,
|
||||
): MoqSession = MoqSession(transport, control, Role.Server)
|
||||
pumpScope: CoroutineScope,
|
||||
): MoqSession = MoqSession(transport, control, Role.Server, pumpScope)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle to an active subscription. [objects] emits every OBJECT_DATAGRAM the
|
||||
* publisher delivers for this subscription's track. Call [unsubscribe] when
|
||||
* done — equivalent to [MoqSession.unsubscribe] with this handle's id.
|
||||
*/
|
||||
class SubscribeHandle internal constructor(
|
||||
val subscribeId: Long,
|
||||
val trackAlias: Long,
|
||||
val ok: SubscribeOk,
|
||||
val objects: Flow<MoqObject>,
|
||||
private val unsubscribeAction: suspend () -> Unit,
|
||||
) {
|
||||
suspend fun unsubscribe() = unsubscribeAction()
|
||||
}
|
||||
|
||||
/** Thrown when the peer violates the MoQ-transport state machine. */
|
||||
class MoqProtocolException(
|
||||
message: String,
|
||||
|
||||
+10
-8
@@ -22,7 +22,7 @@ package com.vitorpamplona.nestsclient.transport
|
||||
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.consumeAsFlow
|
||||
import kotlinx.coroutines.flow.receiveAsFlow
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
|
||||
@@ -34,8 +34,10 @@ import kotlinx.coroutines.sync.withLock
|
||||
* only (no packet loss, no congestion); the real Kwik-backed transport will
|
||||
* exercise those codepaths separately.
|
||||
*
|
||||
* [incomingDatagrams] and [FakeBidiStream.incoming] use [consumeAsFlow]
|
||||
* semantics and therefore may only be collected once per channel.
|
||||
* [incomingDatagrams] and [FakeBidiStream.incoming] use [receiveAsFlow]
|
||||
* semantics: a `take(1)` / `first()` followed by a long-running `collect`
|
||||
* works on the same underlying channel, and cancelling a consumer does not
|
||||
* close the channel for future readers.
|
||||
*/
|
||||
class FakeWebTransport private constructor(
|
||||
private val outboundDatagrams: Channel<ByteArray>,
|
||||
@@ -60,7 +62,7 @@ class FakeWebTransport private constructor(
|
||||
return local
|
||||
}
|
||||
|
||||
override fun incomingUniStreams(): Flow<WebTransportReadStream> = inboundUniStreams.consumeAsFlow()
|
||||
override fun incomingUniStreams(): Flow<WebTransportReadStream> = inboundUniStreams.receiveAsFlow()
|
||||
|
||||
override suspend fun sendDatagram(payload: ByteArray): Boolean {
|
||||
if (!open) return false
|
||||
@@ -68,7 +70,7 @@ class FakeWebTransport private constructor(
|
||||
return true
|
||||
}
|
||||
|
||||
override fun incomingDatagrams(): Flow<ByteArray> = inboundDatagrams.consumeAsFlow()
|
||||
override fun incomingDatagrams(): Flow<ByteArray> = inboundDatagrams.receiveAsFlow()
|
||||
|
||||
override suspend fun close(
|
||||
code: Int,
|
||||
@@ -89,7 +91,7 @@ class FakeWebTransport private constructor(
|
||||
* Flow of peer-opened bidi streams (i.e. the local endpoint of a stream the
|
||||
* other side created via [openBidiStream]).
|
||||
*/
|
||||
fun peerOpenedBidiStreams(): Flow<FakeBidiStream> = inboundBidiStreams.consumeAsFlow()
|
||||
fun peerOpenedBidiStreams(): Flow<FakeBidiStream> = inboundBidiStreams.receiveAsFlow()
|
||||
|
||||
companion object {
|
||||
/**
|
||||
@@ -130,7 +132,7 @@ class FakeBidiStream internal constructor(
|
||||
private val write: Channel<ByteArray>,
|
||||
private val read: Channel<ByteArray>,
|
||||
) : WebTransportBidiStream {
|
||||
override fun incoming(): Flow<ByteArray> = read.consumeAsFlow()
|
||||
override fun incoming(): Flow<ByteArray> = read.receiveAsFlow()
|
||||
|
||||
override suspend fun write(chunk: ByteArray) {
|
||||
write.send(chunk)
|
||||
@@ -144,5 +146,5 @@ class FakeBidiStream internal constructor(
|
||||
class FakeReadStream internal constructor(
|
||||
private val read: Channel<ByteArray>,
|
||||
) : WebTransportReadStream {
|
||||
override fun incoming(): Flow<ByteArray> = read.consumeAsFlow()
|
||||
override fun incoming(): Flow<ByteArray> = read.receiveAsFlow()
|
||||
}
|
||||
|
||||
+173
-16
@@ -23,8 +23,11 @@ package com.vitorpamplona.nestsclient.moq
|
||||
import com.vitorpamplona.nestsclient.transport.FakeWebTransport
|
||||
import kotlinx.coroutines.async
|
||||
import kotlinx.coroutines.flow.first
|
||||
import kotlinx.coroutines.flow.take
|
||||
import kotlinx.coroutines.flow.toList
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertContentEquals
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFailsWith
|
||||
|
||||
@@ -36,7 +39,7 @@ class MoqSessionTest {
|
||||
|
||||
val clientJob =
|
||||
async {
|
||||
val session = MoqSession.client(clientSide)
|
||||
val session = MoqSession.client(clientSide, backgroundScope)
|
||||
session.setup(listOf(MoqVersion.DRAFT_17))
|
||||
session.selectedVersion
|
||||
}
|
||||
@@ -44,16 +47,11 @@ class MoqSessionTest {
|
||||
val serverJob =
|
||||
async {
|
||||
val control = serverSide.peerOpenedBidiStreams().first()
|
||||
val session = MoqSession.server(serverSide, control)
|
||||
val session = MoqSession.server(serverSide, control, backgroundScope)
|
||||
session.setup(
|
||||
supportedVersions = listOf(MoqVersion.DRAFT_17, MoqVersion.DRAFT_11),
|
||||
clientParameters =
|
||||
listOf(
|
||||
SetupParameter(
|
||||
SetupParameter.KEY_MAX_SUBSCRIBE_ID,
|
||||
byteArrayOf(0x10),
|
||||
),
|
||||
),
|
||||
listOf(SetupParameter(SetupParameter.KEY_MAX_SUBSCRIBE_ID, byteArrayOf(0x10))),
|
||||
)
|
||||
session.selectedVersion
|
||||
}
|
||||
@@ -69,8 +67,7 @@ class MoqSessionTest {
|
||||
|
||||
val clientJob =
|
||||
async {
|
||||
val session = MoqSession.client(clientSide)
|
||||
// Client prefers 17, falls back to 11.
|
||||
val session = MoqSession.client(clientSide, backgroundScope)
|
||||
session.setup(listOf(MoqVersion.DRAFT_17, MoqVersion.DRAFT_11))
|
||||
session.selectedVersion
|
||||
}
|
||||
@@ -78,8 +75,7 @@ class MoqSessionTest {
|
||||
val serverJob =
|
||||
async {
|
||||
val control = serverSide.peerOpenedBidiStreams().first()
|
||||
val session = MoqSession.server(serverSide, control)
|
||||
// Server only speaks 11 → overlap is 11.
|
||||
val session = MoqSession.server(serverSide, control, backgroundScope)
|
||||
session.setup(supportedVersions = listOf(MoqVersion.DRAFT_11))
|
||||
session.selectedVersion
|
||||
}
|
||||
@@ -95,23 +91,184 @@ class MoqSessionTest {
|
||||
|
||||
val clientJob =
|
||||
async {
|
||||
val session = MoqSession.client(clientSide)
|
||||
val session = MoqSession.client(clientSide, backgroundScope)
|
||||
runCatching { session.setup(listOf(MoqVersion.DRAFT_17)) }
|
||||
}
|
||||
|
||||
val serverJob =
|
||||
async {
|
||||
val control = serverSide.peerOpenedBidiStreams().first()
|
||||
val session = MoqSession.server(serverSide, control)
|
||||
val session = MoqSession.server(serverSide, control, backgroundScope)
|
||||
assertFailsWith<MoqProtocolException> {
|
||||
session.setup(supportedVersions = listOf(MoqVersion.DRAFT_11))
|
||||
}
|
||||
}
|
||||
|
||||
serverJob.await()
|
||||
// Client's result: its control stream closes with no reply, which throws.
|
||||
// We just verify that some throwable propagated.
|
||||
val clientOutcome = clientJob.await()
|
||||
assert(clientOutcome.isFailure) { "client setup should have failed, got: $clientOutcome" }
|
||||
}
|
||||
|
||||
/**
|
||||
* End-to-end: client subscribes; a *raw* server peer (no MoqSession on
|
||||
* the server side, since that would compete with the test's manual reads
|
||||
* for control-stream items) responds with SUBSCRIBE_OK and three
|
||||
* OBJECT_DATAGRAMs. The client's flow should yield them in order.
|
||||
*/
|
||||
@Test
|
||||
fun subscribe_completes_on_subscribe_ok_then_delivers_datagrams_in_order() =
|
||||
runTest {
|
||||
val (clientSide, serverSide) = FakeWebTransport.pair()
|
||||
val clientSession = MoqSession.client(clientSide, backgroundScope)
|
||||
|
||||
// Server-side raw peer: handshake + serve one subscription manually.
|
||||
val serverWork =
|
||||
async {
|
||||
val serverControl = serverSide.peerOpenedBidiStreams().first()
|
||||
// Handshake: read CLIENT_SETUP, write SERVER_SETUP.
|
||||
val clientSetupFrame = serverControl.incoming().first()
|
||||
val clientSetup = MoqCodec.decode(clientSetupFrame)!!.message as ClientSetup
|
||||
val version = clientSetup.supportedVersions.first()
|
||||
serverControl.write(MoqCodec.encode(ServerSetup(selectedVersion = version)))
|
||||
|
||||
// Wait for SUBSCRIBE, reply with SUBSCRIBE_OK + three datagrams.
|
||||
val sub = MoqCodec.decode(serverControl.incoming().first())!!.message as Subscribe
|
||||
serverControl.write(
|
||||
MoqCodec.encode(
|
||||
SubscribeOk(
|
||||
subscribeId = sub.subscribeId,
|
||||
expiresMs = 60_000,
|
||||
groupOrder = 0,
|
||||
contentExists = false,
|
||||
),
|
||||
),
|
||||
)
|
||||
repeat(3) { i ->
|
||||
val obj =
|
||||
MoqObject(
|
||||
trackAlias = sub.trackAlias,
|
||||
groupId = 0,
|
||||
objectId = i.toLong(),
|
||||
publisherPriority = 0x80,
|
||||
payload = byteArrayOf(i.toByte()),
|
||||
)
|
||||
serverSide.sendDatagram(MoqObjectDatagram.encode(obj))
|
||||
}
|
||||
sub
|
||||
}
|
||||
|
||||
clientSession.setup(listOf(MoqVersion.DRAFT_17))
|
||||
|
||||
val handle =
|
||||
clientSession.subscribe(
|
||||
namespace = TrackNamespace.of("nests", "test-room"),
|
||||
trackName = "speaker-1".encodeToByteArray(),
|
||||
)
|
||||
val sub = serverWork.await()
|
||||
|
||||
assertEquals(sub.subscribeId, handle.subscribeId)
|
||||
assertEquals(sub.trackAlias, handle.trackAlias)
|
||||
assertEquals(60_000L, handle.ok.expiresMs)
|
||||
|
||||
val received = handle.objects.take(3).toList()
|
||||
assertEquals(listOf(0L, 1L, 2L), received.map { it.objectId })
|
||||
assertContentEquals(byteArrayOf(0), received[0].payload)
|
||||
assertContentEquals(byteArrayOf(1), received[1].payload)
|
||||
assertContentEquals(byteArrayOf(2), received[2].payload)
|
||||
|
||||
clientSession.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun subscribe_throws_MoqProtocolException_when_publisher_replies_with_subscribe_error() =
|
||||
runTest {
|
||||
val (clientSide, serverSide) = FakeWebTransport.pair()
|
||||
val clientSession = MoqSession.client(clientSide, backgroundScope)
|
||||
|
||||
val rejector =
|
||||
async {
|
||||
val serverControl = serverSide.peerOpenedBidiStreams().first()
|
||||
val clientSetup =
|
||||
MoqCodec.decode(serverControl.incoming().first())!!.message as ClientSetup
|
||||
serverControl.write(MoqCodec.encode(ServerSetup(clientSetup.supportedVersions.first())))
|
||||
|
||||
val sub =
|
||||
MoqCodec.decode(serverControl.incoming().first())!!.message as Subscribe
|
||||
serverControl.write(
|
||||
MoqCodec.encode(
|
||||
SubscribeError(
|
||||
subscribeId = sub.subscribeId,
|
||||
errorCode = 404,
|
||||
reasonPhrase = "track not found",
|
||||
trackAlias = sub.trackAlias,
|
||||
),
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
clientSession.setup(listOf(MoqVersion.DRAFT_17))
|
||||
|
||||
val ex =
|
||||
assertFailsWith<MoqProtocolException> {
|
||||
clientSession.subscribe(
|
||||
namespace = TrackNamespace.of("nests", "test-room"),
|
||||
trackName = "missing".encodeToByteArray(),
|
||||
)
|
||||
}
|
||||
rejector.await()
|
||||
assert("404" in ex.message!!) { "error message should mention the code: ${ex.message}" }
|
||||
|
||||
clientSession.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun datagrams_for_unknown_track_alias_are_dropped_silently() =
|
||||
runTest {
|
||||
val (clientSide, serverSide) = FakeWebTransport.pair()
|
||||
val clientSession = MoqSession.client(clientSide, backgroundScope)
|
||||
|
||||
val handshake =
|
||||
async {
|
||||
val serverControl = serverSide.peerOpenedBidiStreams().first()
|
||||
val cs = MoqCodec.decode(serverControl.incoming().first())!!.message as ClientSetup
|
||||
serverControl.write(MoqCodec.encode(ServerSetup(cs.supportedVersions.first())))
|
||||
}
|
||||
clientSession.setup(listOf(MoqVersion.DRAFT_17))
|
||||
handshake.await()
|
||||
|
||||
// Send a datagram for a track no one subscribed to. The pump should
|
||||
// silently drop it; the session stays usable + closes cleanly.
|
||||
serverSide.sendDatagram(
|
||||
MoqObjectDatagram.encode(
|
||||
MoqObject(
|
||||
trackAlias = 9_999,
|
||||
groupId = 0,
|
||||
objectId = 0,
|
||||
publisherPriority = 0x80,
|
||||
payload = byteArrayOf(0xAA.toByte()),
|
||||
),
|
||||
),
|
||||
)
|
||||
testScheduler.runCurrent()
|
||||
clientSession.close()
|
||||
}
|
||||
|
||||
/**
|
||||
* Existing FakeWebTransport bidi-stream test still passes after the
|
||||
* receiveAsFlow refactor: write + finish + collect should still surface
|
||||
* the chunk and complete cleanly.
|
||||
*/
|
||||
@Test
|
||||
fun fake_bidi_stream_finish_completes_the_consumer_flow() =
|
||||
runTest {
|
||||
val (a, b) = FakeWebTransport.pair()
|
||||
val clientStream = a.openBidiStream()
|
||||
clientStream.write(byteArrayOf(0xAA.toByte()))
|
||||
clientStream.finish()
|
||||
|
||||
val serverStream = b.peerOpenedBidiStreams().first()
|
||||
val received = serverStream.incoming().toList()
|
||||
assertEquals(1, received.size)
|
||||
assertContentEquals(byteArrayOf(0xAA.toByte()), received.single())
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user