From 67af4248c97d2e1568e5c22179fa0026dbbc5219 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Sat, 3 Oct 2026 13:36:19 +0200 Subject: fix(android): the cast header is everything before the first moof The node's first chunk can be the 28-byte ftyp alone, the moov in the next; served as the header, the receiver had no moov and gave up. A receiver early for the header now waits for all of it. Co-Authored-By: Claude Opus 5.5 --- .../org/meshbay/client/cast/BoxAccumulator.kt | 11 +++++- .../kotlin/org/meshbay/client/cast/CastRelay.kt | 41 ++++++++++++++++++++-- 2 files changed, 49 insertions(+), 3 deletions(-) (limited to 'packages/meshbay-android/app/src/main') diff --git a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/BoxAccumulator.kt b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/BoxAccumulator.kt index 3aaf8d2..f56d58a 100644 --- a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/BoxAccumulator.kt +++ b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/BoxAccumulator.kt @@ -11,6 +11,14 @@ import java.io.ByteArrayOutputStream class BoxAccumulator { private var buf = ByteArray(0) private var synced = false + private var preambleSeen = false + + /** + * Called once, with every byte before the first moof: the stream's header + * (ftyp, moov), however many pushes it came in. The node's first chunk can + * hold the 28-byte ftyp alone, with the moov in the next one (measured). + */ + var onPreamble: ((ByteArray) -> Unit)? = null fun push(data: ByteArray, offset: Int = 0, length: Int = data.size - offset): List { buf = buf + data.copyOfRange(offset, offset + length) @@ -19,6 +27,7 @@ class BoxAccumulator { if (!synced) { val idx = findMoof() if (idx == -1) return fragments + if (!preambleSeen) { preambleSeen = true; onPreamble?.invoke(buf.copyOfRange(0, idx)) } buf = buf.copyOfRange(idx, buf.size) synced = true } @@ -55,7 +64,7 @@ class BoxAccumulator { return fragments } - fun reset() { buf = ByteArray(0); synced = false } + fun reset() { buf = ByteArray(0); synced = false; preambleSeen = false } private fun findMoof(): Int { for (i in 0..buf.size - 8) { diff --git a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastRelay.kt b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastRelay.kt index 4d52052..21328db 100644 --- a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastRelay.kt +++ b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastRelay.kt @@ -88,6 +88,8 @@ class CastRelay(private val lanAddress: () -> InetAddress?, private val spoolDir @Volatile private var token: String? = null @Volatile private var host: String? = null @Volatile private var initSegment: ByteArray? = null + private val headerLock = Object() + @Volatile private var headerReady = false @Volatile private var subtitle: Subtitle? = null @Volatile private var subtitleVersion = 0 /** How many times a receiver asked for the stream since the relay started. */ @@ -134,9 +136,11 @@ class CastRelay(private val lanAddress: () -> InetAddress?, private val spoolDir fragments = 0 host = address.hostAddress initSegment = init?.let(::headerOnly) + // Nothing to wait for without a first chunk: no header is coming. + headerReady = init == null synchronized(ring) { ring.clear(); ringBytes = 0 } clients.clear() - accum = BoxAccumulator() + accum = BoxAccumulator().also { a -> a.onPreamble = ::preamble } // What a killed process left behind. spoolDir.mkdirs() spoolDir.listFiles()?.forEach { it.delete() } @@ -166,6 +170,37 @@ class CastRelay(private val lanAddress: () -> InetAddress?, private val spoolDir .put("subtitle", subtitleInfo() ?: JSONObject.NULL) } + /** + * The header is everything the page pushed before the first moof. The + * `init` the page hands to start() is only its first chunk, which may be + * the ftyp without the moov: served as the header, the receiver got no + * track description and gave up — the "fails the first time" of a fresh + * start, every time. The pushed stream carries the whole of it; `init` is + * the fallback when nothing came before the first moof. + */ + private fun preamble(bytes: ByteArray) { + val header = headerOnly(bytes) + synchronized(headerLock) { + if (header.size >= 8 && BoxAccumulator.u32(header, 4) == FTYP) initSegment = header + headerReady = true + headerLock.notifyAll() + } + log("header complete: ${initSegment?.size ?: 0} bytes") + } + + /** A receiver that connects before the first moof waits for the whole header, not a piece of it. */ + private fun awaitHeader(): ByteArray? { + val deadline = System.currentTimeMillis() + HEADER_WAIT_MS + synchronized(headerLock) { + while (!headerReady) { + val left = deadline - System.currentTimeMillis() + if (left <= 0) break + headerLock.wait(left) + } + } + return initSegment + } + @Volatile private var pushes = 0 @Volatile private var fragments = 0 @@ -283,7 +318,7 @@ class CastRelay(private val lanAddress: () -> InetAddress?, private val spoolDir "Connection" to "keep-alive", "Transfer-Encoding" to "chunked", )) - initSegment?.let { chunk(out, it) } + awaitHeader()?.let { chunk(out, it) } out.flush() val client = Client(socket, out, File(spoolDir, "client-${spoolSeq.incrementAndGet()}.spool")) val backlog = synchronized(ring) { @@ -360,6 +395,8 @@ class CastRelay(private val lanAddress: () -> InetAddress?, private val spoolDir const val RING_CAP = 64 const val RING_MAX_BYTES = 32L * 1024 * 1024 const val SPOOL_MAX_BYTES = 2L * 1024 * 1024 * 1024 + private const val HEADER_WAIT_MS = 10_000L + private const val FTYP = 0x66747970L private const val BLOCK = 256 * 1024 const val PORT_BASE = 19550 const val PORT_COUNT = 4 -- cgit v1.2.3