diff options
Diffstat (limited to 'packages/meshbay-android/app/src/main/kotlin')
4 files changed, 135 insertions, 12 deletions
diff --git a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/MainActivity.kt b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/MainActivity.kt index 41f5457..23df907 100644 --- a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/MainActivity.kt +++ b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/MainActivity.kt @@ -80,7 +80,8 @@ class MainActivity : Activity() { startActivity = { intent -> runOnUiThread { startActivity(intent) } }) // A process killed mid-download left unfinished files; nothing else will. Thread { saves.cleanUpAfterAKilledProcess() }.start() - cast = CastChannels(this, onCasting = { on -> runOnUiThread { casting(on) } }) + cast = CastChannels(this, onCasting = { on -> runOnUiThread { casting(on) } }, + tell = { m -> runOnUiThread { android.widget.Toast.makeText(this, m, android.widget.Toast.LENGTH_LONG).show() } }) channels = Channels(hub, onHubChanged = { runOnUiThread { reloadForHub() } }, hasCatalogue = { code -> hasAsset("ui/locales/$code.js") }, keys = keys, saves = saves, cast = cast) diff --git a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastChannels.kt b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastChannels.kt index 0efae9d..1044209 100644 --- a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastChannels.kt +++ b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastChannels.kt @@ -19,6 +19,7 @@ import java.util.Base64 class CastChannels( private val context: Context, private val onCasting: (Boolean) -> Unit, + private val tell: (String) -> Unit = {}, ) { val control = CastControl(context) val relay = CastRelay(::lanAddress) @@ -49,16 +50,29 @@ class CastChannels( "cast:devices" -> control.devices() "cast:chromecast:connect" -> { val o = a.optJSONObject(0) ?: JSONObject() - control.connect(o.optString("deviceId", ""), relayUrl(o), subtitleOf(o)) + reported { control.connect(o.optString("deviceId", ""), relayUrl(o), subtitleOf(o)) } } "cast:chromecast:reload" -> { val o = a.optJSONObject(0) ?: JSONObject() - control.reload(relayUrl(o), subtitleOf(o)) + reported { control.reload(relayUrl(o), subtitleOf(o)) } } "cast:chromecast:disconnect" -> control.disconnect() else -> throw Refused("Refused: no such channel") } + /** + * A failed cast says why, on the screen. The player catches the error and + * stops the relay without a word, which leaves the television on the + * receiver's idle screen and nobody knowing whether it ever reached the + * phone — so the receiver's reason and that fact are shown here. + */ + private fun <T> reported(block: () -> T): T = try { block() } catch (e: Exception) { + val reached = if (relay.streamRequests > 0) "the television did reach this phone" + else "the television never reached this phone at ${relay.url?.substringBefore("/stream") ?: "?"}" + tell("Cast failed: ${e.message} — $reached") + throw e + } + fun push(bytes: ByteArray, offset: Int, length: Int): Boolean { relay.push(bytes, offset, length); return true } /** @@ -77,11 +91,17 @@ class CastChannels( private fun subtitleOf(o: JSONObject): JSONObject? = if (o.has("subtitle") && o.get("subtitle") != JSONObject.NULL) o.optJSONObject("subtitle") else relay.subtitleInfo() - /** The Wi-Fi (or Ethernet) address, never the mobile network's: a TV cannot be reached there. */ + /** + * The Wi-Fi (or Ethernet) address, never the mobile network's: a TV cannot + * be reached there. Nor a VPN's: a VPN network carries the transports of + * the network under it, Wi-Fi included, and its address is a tunnel the + * television cannot route to. + */ private fun lanAddress(): InetAddress? { val cm = context.getSystemService(ConnectivityManager::class.java) for (network in cm.allNetworks) { val caps = cm.getNetworkCapabilities(network) ?: continue + if (caps.hasTransport(NetworkCapabilities.TRANSPORT_VPN)) continue if (!caps.hasTransport(NetworkCapabilities.TRANSPORT_WIFI) && !caps.hasTransport(NetworkCapabilities.TRANSPORT_ETHERNET)) continue val addr = cm.getLinkProperties(network)?.linkAddresses?.map { it.address } diff --git a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastControl.kt b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastControl.kt index 88e3a03..e2cd58a 100644 --- a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastControl.kt +++ b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastControl.kt @@ -111,7 +111,10 @@ class CastControl(private val context: Context) { var result: CastSession? = null val listener = object : SessionManagerListener<CastSession> { override fun onSessionStarted(s: CastSession, id: String) { result = s; started.countDown() } - override fun onSessionStartFailed(s: CastSession, error: Int) { started.countDown() } + override fun onSessionStartFailed(s: CastSession, error: Int) { + Log.w(TAG, "session start failed: error=$error") + started.countDown() + } override fun onSessionStarting(s: CastSession) {} override fun onSessionEnding(s: CastSession) {} override fun onSessionEnded(s: CastSession, error: Int) {} @@ -121,6 +124,7 @@ class CastControl(private val context: Context) { override fun onSessionSuspended(s: CastSession, reason: Int) {} } cast.sessionManager.addSessionManagerListener(listener, CastSession::class.java) + Log.i(TAG, "connect: selecting route '${route.name}'") r.selectRoute(route) deviceName = route.name Pending(started, { result }, { cast.sessionManager.removeSessionManagerListener(listener, CastSession::class.java) }) @@ -134,8 +138,12 @@ class CastControl(private val context: Context) { else -> throw IllegalStateException() } } + // The cast framework's objects answer on the main thread only, the + // device's name included: read it there, never from this worker. + val name = deviceName ?: onMain { session.castDevice?.friendlyName } + Log.i(TAG, "session up on '$name', loading the relay stream") val state = load(session, mediaUrl, subtitle) - return JSONObject().put("deviceName", deviceName ?: session.castDevice?.friendlyName).put("playerState", state) + return JSONObject().put("deviceName", name ?: JSONObject.NULL).put("playerState", state) } private class Pending(val done: CountDownLatch, val session: () -> CastSession?, val cleanup: () -> Unit) @@ -177,12 +185,47 @@ class CastControl(private val context: Context) { .apply { if (subUrl != null) setActiveTrackIds(longArrayOf(TEXT_TRACK_ID)) }.build() val loaded = CountDownLatch(1) var ok = false + var reason = "" onMain { val client = session.remoteMediaClient ?: throw IllegalStateException("The receiver has no media channel") - client.load(request).setResultCallback { r -> ok = r.status.isSuccess; loaded.countDown() } + client.load(request).setResultCallback { r -> + ok = r.status.isSuccess + reason = "code ${r.status.statusCode}" + (r.status.statusMessage?.let { ", $it" } ?: "") + // The receiver's own reason, which is the only one there is: + // the page is told "refused" and nothing else. + Log.i(TAG, "load result: ok=$ok code=${r.status.statusCode} " + + "message=${r.status.statusMessage} state=${stateName(client.playerState)} " + + "idleReason=${client.idleReason} subtitles=${subUrl != null}") + loaded.countDown() + } + } + if (!loaded.await(20, TimeUnit.SECONDS)) { + Log.w(TAG, "load: no answer from the receiver in 20 s") + throw IllegalStateException("The receiver did not load the stream") + } + if (!ok) throw IllegalStateException("The receiver refused the stream ($reason)") + // A load the receiver accepted is not a film on the screen: it answers + // IDLE, then fetches the stream, and only then plays — or gives up. + // So the answer waits for what the receiver actually did (measured + // against this receiver: OK/IDLE, then BUFFERING, then PLAYING). + val deadline = System.currentTimeMillis() + PLAY_WAIT_MS + while (System.currentTimeMillis() < deadline) { + val (state, idle) = onMain { + val c = session.remoteMediaClient + (c?.playerState ?: MediaStatus.PLAYER_STATE_UNKNOWN) to (c?.idleReason ?: MediaStatus.IDLE_REASON_NONE) + } + if (state == MediaStatus.PLAYER_STATE_PLAYING || state == MediaStatus.PLAYER_STATE_BUFFERING || + state == MediaStatus.PLAYER_STATE_PAUSED) { + Log.i(TAG, "receiver is ${stateName(state)}") + return stateName(state) + } + if (state == MediaStatus.PLAYER_STATE_IDLE && idle != MediaStatus.IDLE_REASON_NONE) { + Log.w(TAG, "receiver gave up: idle reason ${idleName(idle)}") + throw IllegalStateException("The receiver gave up on the stream (${idleName(idle)})") + } + Thread.sleep(300) } - if (!loaded.await(20, TimeUnit.SECONDS)) throw IllegalStateException("The receiver did not load the stream") - if (!ok) throw IllegalStateException("The receiver refused the stream") + Log.w(TAG, "receiver still not playing after ${PLAY_WAIT_MS / 1000} s") return onMain { stateName(session.remoteMediaClient?.playerState ?: MediaStatus.PLAYER_STATE_UNKNOWN) } } @@ -207,11 +250,22 @@ class CastControl(private val context: Context) { companion object { const val SCAN_DURATION_MS = 6000L + private const val TAG = "MeshBayCast" // The one track the receiver is ever told about; changing subtitles // reloads with a new URL under this same id. const val TEXT_TRACK_ID = 1L private val SCAN_TOKEN = Any() + private const val PLAY_WAIT_MS = 15000L + + fun idleName(r: Int) = when (r) { + MediaStatus.IDLE_REASON_FINISHED -> "finished" + MediaStatus.IDLE_REASON_CANCELED -> "cancelled" + MediaStatus.IDLE_REASON_INTERRUPTED -> "interrupted" + MediaStatus.IDLE_REASON_ERROR -> "error" + else -> "reason $r" + } + fun stateName(s: Int) = when (s) { MediaStatus.PLAYER_STATE_PLAYING -> "PLAYING" MediaStatus.PLAYER_STATE_PAUSED -> "PAUSED" 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 5cd5f3e..c56d57d 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 @@ -1,5 +1,6 @@ package org.meshbay.client.cast +import android.util.Log import org.json.JSONObject import java.io.BufferedReader import java.io.InputStreamReader @@ -47,6 +48,9 @@ class CastRelay(private val lanAddress: () -> InetAddress?) { @Volatile private var initSegment: ByteArray? = null @Volatile private var subtitle: Subtitle? = null @Volatile private var subtitleVersion = 0 + /** How many times a receiver asked for the stream since the relay started. */ + @Volatile var streamRequests = 0 + private set private val ring = ArrayDeque<ByteArray>() private var ringBytes = 0L private val clients = ConcurrentHashMap.newKeySet<Client>() @@ -83,8 +87,11 @@ class CastRelay(private val lanAddress: () -> InetAddress?) { if (server != null) stop() val address = lanAddress() ?: throw IllegalStateException("Casting needs this device on Wi-Fi") token = randomToken() + streamRequests = 0 + pushes = 0 + fragments = 0 host = address.hostAddress - initSegment = init + initSegment = init?.let(::headerOnly) synchronized(ring) { ring.clear(); ringBytes = 0 } clients.clear() accum = BoxAccumulator() @@ -95,7 +102,11 @@ class CastRelay(private val lanAddress: () -> InetAddress?) { for (i in 0 until PORT_COUNT) { val s = ServerSocket() try { - s.reuseAddress = false + // As Node's server does (and so the desktop relay): a port just + // closed sits in TIME_WAIT, and every seek restarts the relay — + // without this, four quick seeks used all four ports. It does + // not let two live listeners share a port. + s.reuseAddress = true s.bind(InetSocketAddress(address, PORT_BASE + i)) bound = s; port = PORT_BASE + i break @@ -105,13 +116,20 @@ class CastRelay(private val lanAddress: () -> InetAddress?) { } server = bound ?: throw IllegalStateException("All cast relay ports are in use") Thread({ acceptLoop(bound) }, "cast-relay-accept").apply { isDaemon = true }.start() + log("started on ${hostPart()}:$port, init ${init?.size ?: 0} bytes, header ${initSegment?.size ?: 0} bytes") return JSONObject().put("url", url).put("port", port).put("token", token) .put("subtitle", subtitleInfo() ?: JSONObject.NULL) } + @Volatile private var pushes = 0 + @Volatile private var fragments = 0 + fun push(data: ByteArray, offset: Int = 0, length: Int = data.size - offset) { if (server == null) return + pushes++ + if (pushes <= 3 || pushes % 100 == 0) log("push #$pushes: $length bytes, $fragments fragments so far") for (frag in accum.push(data, offset, length)) { + fragments++ synchronized(ring) { ring.addLast(frag); ringBytes += frag.size // Bounded in fragments, as the desktop is, and in bytes too: a @@ -173,13 +191,14 @@ class CastRelay(private val lanAddress: () -> InetAddress?) { if (method == "OPTIONS") return respond(out, socket, 204, CORS_HEADERS) val path = target.substringBefore('?') + log("$method $path from ${socket.inetAddress?.hostAddress}") val query = target.substringAfter('?', "").split('&').associate { it.substringBefore('=') to it.substringAfter('=', "") } if (token == null || query["t"] != token) return respond(out, socket, 403, emptyMap()) when (path) { "/subs.vtt" -> serveSubtitle(out, socket) - "/stream.mp4" -> serveStream(out, socket) + "/stream.mp4" -> { streamRequests++; serveStream(out, socket) } else -> respond(out, socket, 404, emptyMap()) } } catch (e: Exception) { @@ -212,6 +231,7 @@ class CastRelay(private val lanAddress: () -> InetAddress?) { for (frag in backlog) chunk(out, frag) out.flush() clients.add(client) + log("client ${socket.inetAddress?.hostAddress} served init + ${backlog.size} fragments") try { while (!client.closed) { val frag = client.queue.take() @@ -223,12 +243,37 @@ class CastRelay(private val lanAddress: () -> InetAddress?) { } catch (e: Exception) { // The receiver went away; nothing to tell anyone. } finally { + log("client ${socket.inetAddress?.hostAddress} gone") clients.remove(client) try { socket.close() } catch (e: Exception) {} } } companion object { + /** + * The init segment is what comes before the first moof, and only that. + * + * The page hands over the first chunk it decrypted, which is a slice of + * the stream and not a box: on a real film it was 65536 bytes — ftyp, + * moov, then the first moof and the start of its mdat. Served whole, + * the receiver read that partial fragment, then the same fragment again + * from the accumulator (the page pushes that chunk too), and the box + * structure was broken from the first fragment: the receiver gave up + * within three seconds, on the television's idle screen. Whether the + * first chunk carries film depends on when the node read ffmpeg's + * output, so the same film can work once and not the next time. + */ + fun headerOnly(init: ByteArray): ByteArray { + var at = 0 + while (at + 8 <= init.size) { + if (BoxAccumulator.u32(init, at + 4) == BoxAccumulator.MOOF) return init.copyOfRange(0, at) + val size = BoxAccumulator.u32(init, at) + if (size < 8) break + at += size.toInt() + } + return init + } + const val RING_CAP = 64 const val RING_MAX_BYTES = 32L * 1024 * 1024 const val BACKPRESSURE_HIGH = 8L * 1024 * 1024 @@ -236,6 +281,9 @@ class CastRelay(private val lanAddress: () -> InetAddress?) { const val PORT_COUNT = 4 private val END = ByteArray(0) + // Never the URL: it carries the token. + private fun log(m: String) = try { Log.i("MeshBayCast", "[relay] $m") } catch (e: RuntimeException) { /* JVM tests */ } + // A receiver reads a side-loaded subtitle with XHR from its own // origin, so the headers it sends are allowed by name — Range included. val CORS_HEADERS = mapOf( |