feat(nestsClient): WebTransport abstraction + fake + Kwik stub
Phase 3b-1 of the Clubhouse/nests integration. Lands the [WebTransportSession] abstraction the MoQ layer (Phase 3c) will code against, so MoQ framing + tests can develop in parallel with the real Kwik-based transport integration (deferred to Phase 3b-2). commonMain: - `WebTransportSession` — bidi/uni stream access, datagrams, close. - `WebTransportBidiStream` / `WebTransportReadStream` / `WebTransportWriteStream` — minimal read/write surface. - `WebTransportFactory.connect(authority, path, bearerToken)` for opening sessions. - `WebTransportException(kind, ...)` with four canonical failure modes (HandshakeFailed, ConnectRejected, PeerClosed, NotImplemented) so UI code doesn't need to know about library-specific exceptions. - `FakeWebTransport.pair()` — in-memory, fully-wired client/server pair for unit-testing MoQ framing without touching a real QUIC stack. jvmAndroid: - `KwikWebTransportFactory` stub that throws `WebTransportException(NotImplemented)` at `connect()`. Doc spells out the handshake sequence (QUIC dial → H3 SETTINGS → `:method=CONNECT :protocol=webtransport` Extended CONNECT → 2xx) so Phase 3b-2 can drop the real implementation in without touching callers. Tests: - `FakeWebTransportTest` — datagram round-trip both directions, bidi-stream write visible on peer side, close() flips isOpen. - `WebTransportExceptionTest` — kind + message + cause preserved. - `KwikWebTransportFactoryTest` — `connect()` currently fails with the NotImplemented sentinel, so Phase 3b-2 can replace this test when the real handshake lands. https://claude.ai/code/session_013nVLALALKaHVgHm9u5Cg8D
This commit is contained in:
Claude
+148
@@ -0,0 +1,148 @@
|
||||
/*
|
||||
* Copyright (c) 2025 Vitor Pamplona
|
||||
*
|
||||
* Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
* this software and associated documentation files (the "Software"), to deal in
|
||||
* the Software without restriction, including without limitation the rights to use,
|
||||
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
|
||||
* Software, and to permit persons to whom the Software is furnished to do so,
|
||||
* subject to the following conditions:
|
||||
*
|
||||
* The above copyright notice and this permission notice shall be included in all
|
||||
* copies or substantial portions of the Software.
|
||||
*
|
||||
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
|
||||
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
*/
|
||||
package com.vitorpamplona.nestsclient.transport
|
||||
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.consumeAsFlow
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
|
||||
/**
|
||||
* In-memory [WebTransportSession] used by tests for MoQ framing / session logic.
|
||||
*
|
||||
* A pair of fakes is connected via [pair] — anything written on one side is
|
||||
* delivered on the other. This deliberately simulates *success* semantics
|
||||
* 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.
|
||||
*/
|
||||
class FakeWebTransport private constructor(
|
||||
private val outboundDatagrams: Channel<ByteArray>,
|
||||
private val inboundDatagrams: Channel<ByteArray>,
|
||||
private val outboundBidiStreams: Channel<FakeBidiStream>,
|
||||
private val inboundBidiStreams: Channel<FakeBidiStream>,
|
||||
private val inboundUniStreams: Channel<WebTransportReadStream>,
|
||||
) : WebTransportSession {
|
||||
private val stateLock = Mutex()
|
||||
private var open = true
|
||||
|
||||
override val isOpen: Boolean get() = open
|
||||
|
||||
override suspend fun openBidiStream(): WebTransportBidiStream {
|
||||
stateLock.withLock { check(open) { "session closed" } }
|
||||
val localToPeer = Channel<ByteArray>(Channel.BUFFERED)
|
||||
val peerToLocal = Channel<ByteArray>(Channel.BUFFERED)
|
||||
val local = FakeBidiStream(write = localToPeer, read = peerToLocal)
|
||||
val peer = FakeBidiStream(write = peerToLocal, read = localToPeer)
|
||||
// Send *peer* half to the other side's inbound queue.
|
||||
outboundBidiStreams.send(peer)
|
||||
return local
|
||||
}
|
||||
|
||||
override fun incomingUniStreams(): Flow<WebTransportReadStream> = inboundUniStreams.consumeAsFlow()
|
||||
|
||||
override suspend fun sendDatagram(payload: ByteArray): Boolean {
|
||||
if (!open) return false
|
||||
outboundDatagrams.send(payload)
|
||||
return true
|
||||
}
|
||||
|
||||
override fun incomingDatagrams(): Flow<ByteArray> = inboundDatagrams.consumeAsFlow()
|
||||
|
||||
override suspend fun close(
|
||||
code: Int,
|
||||
reason: String,
|
||||
) {
|
||||
stateLock.withLock {
|
||||
if (!open) return
|
||||
open = false
|
||||
}
|
||||
outboundDatagrams.close()
|
||||
inboundDatagrams.close()
|
||||
outboundBidiStreams.close()
|
||||
inboundBidiStreams.close()
|
||||
inboundUniStreams.close()
|
||||
}
|
||||
|
||||
/**
|
||||
* 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()
|
||||
|
||||
companion object {
|
||||
/**
|
||||
* Create two linked fakes that act as "client" and "server" endpoints of
|
||||
* the same virtual WebTransport session. Datagrams and streams opened on
|
||||
* one side appear on the other.
|
||||
*/
|
||||
fun pair(): Pair<FakeWebTransport, FakeWebTransport> {
|
||||
val aToBDatagrams = Channel<ByteArray>(Channel.BUFFERED)
|
||||
val bToADatagrams = Channel<ByteArray>(Channel.BUFFERED)
|
||||
val aToBBidi = Channel<FakeBidiStream>(Channel.BUFFERED)
|
||||
val bToABidi = Channel<FakeBidiStream>(Channel.BUFFERED)
|
||||
val aToBUni = Channel<WebTransportReadStream>(Channel.BUFFERED)
|
||||
val bToAUni = Channel<WebTransportReadStream>(Channel.BUFFERED)
|
||||
|
||||
val a =
|
||||
FakeWebTransport(
|
||||
outboundDatagrams = aToBDatagrams,
|
||||
inboundDatagrams = bToADatagrams,
|
||||
outboundBidiStreams = aToBBidi,
|
||||
inboundBidiStreams = bToABidi,
|
||||
inboundUniStreams = bToAUni,
|
||||
)
|
||||
val b =
|
||||
FakeWebTransport(
|
||||
outboundDatagrams = bToADatagrams,
|
||||
inboundDatagrams = aToBDatagrams,
|
||||
outboundBidiStreams = bToABidi,
|
||||
inboundBidiStreams = aToBBidi,
|
||||
inboundUniStreams = aToBUni,
|
||||
)
|
||||
return a to b
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class FakeBidiStream internal constructor(
|
||||
private val write: Channel<ByteArray>,
|
||||
private val read: Channel<ByteArray>,
|
||||
) : WebTransportBidiStream {
|
||||
override fun incoming(): Flow<ByteArray> = read.consumeAsFlow()
|
||||
|
||||
override suspend fun write(chunk: ByteArray) {
|
||||
write.send(chunk)
|
||||
}
|
||||
|
||||
override suspend fun finish() {
|
||||
write.close()
|
||||
}
|
||||
}
|
||||
|
||||
class FakeReadStream internal constructor(
|
||||
private val read: Channel<ByteArray>,
|
||||
) : WebTransportReadStream {
|
||||
override fun incoming(): Flow<ByteArray> = read.consumeAsFlow()
|
||||
}
|
||||
+126
@@ -0,0 +1,126 @@
|
||||
/*
|
||||
* Copyright (c) 2025 Vitor Pamplona
|
||||
*
|
||||
* Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
* this software and associated documentation files (the "Software"), to deal in
|
||||
* the Software without restriction, including without limitation the rights to use,
|
||||
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
|
||||
* Software, and to permit persons to whom the Software is furnished to do so,
|
||||
* subject to the following conditions:
|
||||
*
|
||||
* The above copyright notice and this permission notice shall be included in all
|
||||
* copies or substantial portions of the Software.
|
||||
*
|
||||
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
|
||||
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
*/
|
||||
package com.vitorpamplona.nestsclient.transport
|
||||
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
|
||||
/**
|
||||
* Platform-agnostic WebTransport session, as produced by a successful Extended
|
||||
* CONNECT (RFC 9220) handshake.
|
||||
*
|
||||
* The MoQ layer (Phase 3c) talks to this interface; the real Kwik-based
|
||||
* implementation sits behind [WebTransportFactory] in jvmAndroid. Keeping this
|
||||
* abstract lets us:
|
||||
* - unit-test the MoQ framing layer with an in-memory fake,
|
||||
* - swap transport implementations (Cronet on Android, a browser-backed
|
||||
* WebView bridge as a contingency, Kwik on JVM desktop) without touching
|
||||
* audio/UI code.
|
||||
*
|
||||
* Lifecycle: the session is opened via [WebTransportFactory.connect] and must
|
||||
* be closed with [close] to release the underlying QUIC connection.
|
||||
*/
|
||||
interface WebTransportSession {
|
||||
/** True once the Extended CONNECT exchange has returned 2xx and before [close] is called. */
|
||||
val isOpen: Boolean
|
||||
|
||||
/**
|
||||
* Open a new bidirectional WebTransport stream. The returned [WebTransportBidiStream]
|
||||
* is writable + readable and closes when either peer half-closes.
|
||||
*/
|
||||
suspend fun openBidiStream(): WebTransportBidiStream
|
||||
|
||||
/**
|
||||
* Flow of inbound unidirectional streams initiated by the peer.
|
||||
*
|
||||
* The flow completes when [close] is called or the peer tears down the session.
|
||||
*/
|
||||
fun incomingUniStreams(): Flow<WebTransportReadStream>
|
||||
|
||||
/** Send a QUIC datagram; returns false if the datagram was dropped by congestion control. */
|
||||
suspend fun sendDatagram(payload: ByteArray): Boolean
|
||||
|
||||
/** Flow of inbound datagrams. */
|
||||
fun incomingDatagrams(): Flow<ByteArray>
|
||||
|
||||
/** Gracefully close the session with an optional application-level code + reason. */
|
||||
suspend fun close(
|
||||
code: Int = 0,
|
||||
reason: String = "",
|
||||
)
|
||||
}
|
||||
|
||||
/** Writable + readable half of a bidirectional WebTransport stream. */
|
||||
interface WebTransportBidiStream :
|
||||
WebTransportReadStream,
|
||||
WebTransportWriteStream
|
||||
|
||||
/** Read-only WebTransport stream (either a received uni stream or the read half of a bidi). */
|
||||
interface WebTransportReadStream {
|
||||
/** Flow of chunks as they arrive. Completes when the peer closes its write side. */
|
||||
fun incoming(): Flow<ByteArray>
|
||||
}
|
||||
|
||||
/** Write-only WebTransport stream. */
|
||||
interface WebTransportWriteStream {
|
||||
suspend fun write(chunk: ByteArray)
|
||||
|
||||
/** Half-close the write side (FIN). No further writes after this call. */
|
||||
suspend fun finish()
|
||||
}
|
||||
|
||||
/**
|
||||
* Factory that opens a [WebTransportSession] against a nests MoQ endpoint.
|
||||
*
|
||||
* [authority] is the `host:port` of the WT server, [path] is the URL path
|
||||
* (nests defaults to `/moq`), and [bearerToken] is typically the token
|
||||
* returned by the `/api/v1/nests/<roomId>` HTTP call in Phase 3a.
|
||||
*/
|
||||
interface WebTransportFactory {
|
||||
suspend fun connect(
|
||||
authority: String,
|
||||
path: String,
|
||||
bearerToken: String? = null,
|
||||
): WebTransportSession
|
||||
}
|
||||
|
||||
/**
|
||||
* Single transport-layer exception type, so UI code only has to differentiate
|
||||
* on [kind] rather than catching library-specific classes.
|
||||
*/
|
||||
class WebTransportException(
|
||||
val kind: Kind,
|
||||
message: String,
|
||||
cause: Throwable? = null,
|
||||
) : RuntimeException(message, cause) {
|
||||
enum class Kind {
|
||||
/** DNS / TCP / TLS / QUIC handshake failed. */
|
||||
HandshakeFailed,
|
||||
|
||||
/** WebTransport Extended CONNECT returned a non-2xx response. */
|
||||
ConnectRejected,
|
||||
|
||||
/** Peer closed the session or a stream unexpectedly. */
|
||||
PeerClosed,
|
||||
|
||||
/** The implementation does not yet support the requested operation. */
|
||||
NotImplemented,
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user