2 files changed, 57 insertions, 3 deletions

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 e2cd58a..8b5d8e1 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
@@ -143,16 +143,45 @@ class CastControl(private val context: Context) {
         val name = deviceName ?: onMain { session.castDevice?.friendlyName }
         Log.i(TAG, "session up on '$name', loading the relay stream")
         val state = load(session, mediaUrl, subtitle)
+        watch(session)
         return JSONObject().put("deviceName", name ?: JSONObject.NULL).put("playerState", state)
     }
 
     private class Pending(val done: CountDownLatch, val session: () -> CastSession?, val cleanup: () -> Unit)
 
+    /**
+     * What the receiver does for the whole film, not only at the start: a
+     * freeze on the television is a BUFFERING the phone never hears about
+     * otherwise. Logged with the position, on the main thread the SDK wants.
+     */
+    private var watched: com.google.android.gms.cast.framework.media.RemoteMediaClient? = null
+    private val watcher = object : com.google.android.gms.cast.framework.media.RemoteMediaClient.Callback() {
+        private var last = -1
+        override fun onStatusUpdated() {
+            val c = watched ?: return
+            val state = c.playerState
+            if (state == last) return
+            last = state
+            Log.i(TAG, "receiver state ${stateName(state)} at ${c.approximateStreamPosition / 1000.0}s" +
+                (if (state == MediaStatus.PLAYER_STATE_IDLE) " (idle: ${idleName(c.idleReason)})" else ""))
+        }
+    }
+
+    private fun watch(session: CastSession) = onMain {
+        val c = session.remoteMediaClient ?: return@onMain
+        if (watched === c) return@onMain
+        watched?.unregisterCallback(watcher)
+        c.registerCallback(watcher)
+        watched = c
+    }
+
     fun reload(mediaUrl: String, subtitle: JSONObject?): JSONObject {
         need()
         val session = onMain { CastContext.getSharedInstance(context).sessionManager.currentCastSession }
             ?: throw Refused("Not connected")
-        return JSONObject().put("playerState", load(session, mediaUrl, subtitle))
+        val state = load(session, mediaUrl, subtitle)
+        watch(session)
+        return JSONObject().put("playerState", state)
     }
 
     private fun load(session: CastSession, mediaUrl: String, subtitle: JSONObject?): String {
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 c56d57d..0d85a17 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
@@ -37,6 +37,9 @@ class CastRelay(private val lanAddress: () -> InetAddress?) {
         val queue = LinkedBlockingDeque<ByteArray>()
         val queued = AtomicLong()
         @Volatile var closed = false
+        val sent = AtomicLong()
+        val dropped = AtomicLong()
+        @Volatile var lastReport = System.currentTimeMillis()
     }
 
     private class Subtitle(val vtt: ByteArray, val language: String, val label: String)
@@ -140,7 +143,17 @@ class CastRelay(private val lanAddress: () -> InetAddress?) {
                 }
             }
             for (c in clients) {
-                if (c.queued.get() > BACKPRESSURE_HIGH) continue   // drop for a slow client
+                if (c.queued.get() > BACKPRESSURE_HIGH) {
+                    // Dropped for a slow client: the picture on the receiver
+                    // freezes until the next fragment it does get. Silent on
+                    // the desktop too; said here, with what was waiting.
+                    val n = c.dropped.incrementAndGet()
+                    if (n <= 3 || n % 20 == 0L) {
+                        log("DROPPED fragment for ${c.socket.inetAddress?.hostAddress}: ${frag.size} bytes, " +
+                            "${c.queued.get() / 1024} KiB already waiting, $n dropped so far")
+                    }
+                    continue
+                }
                 c.queued.addAndGet(frag.size.toLong())
                 c.queue.offer(frag)
             }
@@ -236,9 +249,21 @@ class CastRelay(private val lanAddress: () -> InetAddress?) {
             while (!client.closed) {
                 val frag = client.queue.take()
                 if (frag === END) { out.write("0\r\n\r\n".toByteArray()); out.flush(); break }
-                client.queued.addAndGet(-frag.size.toLong())
+                val t0 = System.currentTimeMillis()
                 chunk(out, frag)
                 out.flush()
+                val took = System.currentTimeMillis() - t0
+                client.queued.addAndGet(-frag.size.toLong())
+                client.sent.incrementAndGet()
+                // A write that blocks is the receiver not reading (or the
+                // Wi-Fi not carrying): the one place a stall downstream shows.
+                if (took > 2000) log("slow write to ${socket.inetAddress?.hostAddress}: ${frag.size} bytes took $took ms")
+                val now = System.currentTimeMillis()
+                if (now - client.lastReport >= 10_000) {
+                    client.lastReport = now
+                    log("client ${socket.inetAddress?.hostAddress}: sent ${client.sent.get()} fragments, " +
+                        "${client.queued.get() / 1024} KiB waiting, ${client.dropped.get()} dropped")
+                }
             }
         } catch (e: Exception) {
             // The receiver went away; nothing to tell anyone.