import "pkg:/source/api/baseRequest.bs"
import "pkg:/source/remotecontrol/remoteCommand.bs"
import "pkg:/source/remotecontrol/remoteProtocol.bs"
import "pkg:/source/roku_modules/log/LogMixin.brs"
import "pkg:/source/utils/misc.bs"
' Remote-control receiver (#666) — the ws:// "Cast to JellyRock" listener. See RemoteControlTask.xml
' for the lifecycle/threading overview. This task owns the vendored WebSocketClient node and runs the
' Jellyfin session socket off the render thread: open -> parse frames -> answer keepalive -> marshal
' commands to the main thread -> reconnect with backoff. It NEVER dispatches (that's remoteDispatch on
' the main thread) and NEVER logs the socket url (it carries the auth token).
sub init()
m.log = new log.Logger("RemoteControl")
m.top.functionName = "runReceiver"
end sub
' Post-login entry point (control="RUN" from Home.isFirstRun, once the session token exists). Selects
' the transport by server scheme: an http:// server uses the ws:// session socket (#666); an https://
' server can't ws:// (Roku has no socket TLS) so it probes for the companion plugin and, if present,
' runs the long-poll command channel (#667). SignOut sets control="STOP", terminating this thread.
sub runReceiver()
serverUrl = m.global.server.serverUrl
' NOTE: the cold-launch cast pairing report (#668) is NOT sent here — it fires as a fire-and-forget
' SubmitSideEffect from Home.isFirstRun (alongside the capabilities POST), so it can never delay this
' command channel. See remoteProtocol.buildPairRequest + docs/architecture/remote-control.md.
if remoteProtocol.isHttpServer(serverUrl)
runWsReceiver(serverUrl)
else
runLongPollReceiver(serverUrl)
end if
end sub
' ws:// transport (#666): build the session-socket url and run the connect/reconnect loop until the
' session ends (token rotated, or control="STOP" kills the thread).
sub runWsReceiver(serverUrl as string)
url = remoteProtocol.buildSocketUrl(serverUrl, m.global.user.authToken, m.global.device.serverDeviceName)
if url = ""
' Missing credential (scheme already gated by runReceiver) -> stay dark.
m.log.info("ws receiver disabled: missing session")
return
end if
' Snapshot the token so a reconnect can detect a server-side rotation and stop cleanly.
m.openedWithToken = m.global.user.authToken
m.port = CreateObject("roMessagePort")
attempt = 0
while true
wasStable = connectAndPump(url)
' The socket closed. Reconnect unless the session is gone. SignOut(false) (logout AND server
' switch) sets control="STOP", killing this thread; this guard only catches a token rotation.
currentToken = m.global.user.authToken
if not isValidAndNotEmpty(currentToken) or currentToken <> m.openedWithToken
m.log.info("session ended; receiver stopping")
return
end if
' Backoff progression. A connection that was ESTABLISHED and STABLE (connectAndPump returned true:
' it opened and stayed up past the stable threshold) then dropped resets the counter so we
' reconnect promptly — a real session that blips shouldn't wait out a stale 30s backoff. A
' never-opened or quickly-flapping connection leaves the counter climbing, so the exponential
' backoff (1s, 2s, 4s, ... capped 30s) is UNCHANGED for a persistently-down/flapping server —
' that path never resets, so it can't hot-loop. increment AFTER the sleep so the 0th tier (1s) is
' actually used rather than skipped straight to 2s.
if wasStable then attempt = 0
backoffMs = remoteProtocol.nextBackoffMs(attempt)
m.log.info("socket closed; reconnecting", "inMs", backoffMs)
sleep(backoffMs)
attempt++
end while
end sub
' Open the socket and pump events until it closes or errors. Sends a KeepAlive whenever the wait
' times out at the interval Jellyfin's ForceKeepAlive requested (0 = block until an event, before any
' ForceKeepAlive arrives). Marshals each dispatchable command to the main thread. Returns on close.
' Returns TRUE when the connection was ESTABLISHED and STABLE — it opened AND stayed up at least
' STABLE_SESSION_MS — so runReceiver can reset its reconnect backoff. A never-opened socket, or one
' that opened but dropped sooner than that (a flap), returns FALSE so the backoff keeps climbing.
' Threshold = the backoff cap (30s): a session that outlived the max backoff interval was clearly real.
function connectAndPump(url as string) as boolean
STABLE_SESSION_MS = 30000
m.ws = CreateObject("roSGNode", "WebSocketClient")
' Publish the child so SignOut can stop it — an external STOP on this task kills the thread
' before closeSocket() can run. Published on the line after CreateObject so the window in which a
' STOP could miss this child is as small as it can be (and bounded anyway — the vendored socket
' loop self-exits once its connection reaches CLOSED).
m.top.socketNode = m.ws
m.ws.observeField("on_open", m.port)
m.ws.observeField("on_message", m.port)
m.ws.observeField("on_close", m.port)
m.ws.observeField("on_error", m.port)
' Terminal backstop (#728) — see the ready_state arm of the pump loop below.
m.ws.observeField("ready_state", m.port)
' Bind this socket to the SAME Jellyfin DeviceId the REST API advertises. Jellyfin parses DeviceId
' from the Authorization header ONLY — the `deviceId` query param buildSocketUrl puts on the url is
' never read — so a header-less upgrade falls back to the DeviceId the auth TOKEN was minted under.
' On an install upgraded from a pre-#721 build that is the old suffixed id, which puts the command
' channel on a DIFFERENT session than the REST API, the capabilities POST and the /pair report all
' use: the cast target resolves but commands are delivered to a session this socket isn't on (#743).
' Verified against Jellyfin 10.11.11: header absent -> session keyed on the token binding; header
' present -> session keyed on the header, and a command addressed to it arrives here.
' The api_key query param stays on the url as the auth carrier: if a reverse proxy strips
' Authorization on the upgrade we still connect, degrading to the old behavior instead of failing.
' Must be set BEFORE `open` — the client reads its headers when it builds the handshake.
m.ws.headers = ["Authorization", buildAuthHeader(false)]
m.log.info("opening session socket") ' NEVER log url — it carries the auth token
m.ws.open = url
sessionTimer = invalid ' marks when the socket opened; invalid => never opened (an unstable attempt)
m.keepAliveIntervalMs = 0 ' set when a ForceKeepAlive arrives; 0 => wait() blocks indefinitely
while true
msg = wait(m.keepAliveIntervalMs, m.port)
if type(msg) = "roSGNodeEvent"
field = msg.getField()
if field = "on_message"
handleFrame(msg.getData())
else if field = "on_open"
m.log.info("socket open")
sessionTimer = CreateObject("roTimespan")
else if isTerminalSocketEvent(field, msg, isValid(sessionTimer))
m.log.info("socket", field)
closeSocket()
return isValid(sessionTimer) and sessionTimer.TotalMilliseconds() >= STABLE_SESSION_MS
end if
else
' wait() timed out -> keepalive tick (only reachable once an interval is set).
if m.keepAliveIntervalMs > 0 then m.ws.send = [remoteProtocol.keepAliveFrame()]
end if
end while
return false ' unreachable (the loop only exits via the on_close/on_error return) — satisfies the analyzer
end function
' True when this event means the connection is over, so the pump loop should tear down and let
' runWsReceiver reconnect. Two signals, because the receiver and the vendored socket loop key off
' DIFFERENT ones:
'
' - on_close / on_error — the fast path, and the only one that fires for an error raised while the
' socket is still OPEN or CLOSING (a failed read, a rejected handshake). That lands here at once,
' ~30s before _CLOSING_DELAY would force the client to CLOSED.
' - ready_state = CLOSED — the backstop. The vendored task loop releases its Task thread on that
' transition (#728), but the receiver would otherwise only ever wake on on_close/on_error. Those
' coincide in the client today — every _close() path in WebSocketClient.brs posts one of the two —
' yet nothing enforces it, and no lint can see it (it is a property of the vendored CLIENT, not of
' the files lint:socket-thread-release reads). Without this arm, an upstream re-sync that broke the
' coincidence would leave the pump blocked forever on a socket whose thread is already gone: no
' crash, no failing test, cast silently dead.
'
' `opened` is load-bearing on the second arm. ready_state is CLOSED before open() is ever called AND
' the task seeds m.top.ready_state from it at startup, so an ungated test would race that seed and
' tear down every connection before it could open. Whichever signal arrives first wins; because
' _close() posts ready_state before on_close, the backstop usually wins the normal server-close path
' — the outcome is identical either way.
function isTerminalSocketEvent(field as string, msg as object, opened as boolean) as boolean
if field = "on_close" or field = "on_error" then return true
return field = "ready_state" and opened and msg.getData() = m.ws.STATE_CLOSED
end function
' Parse one inbound frame and either answer a keepalive or marshal a dispatchable command.
sub handleFrame(data as object)
if not isValid(data) or not isValidAndNotEmpty(data.message) then return
envelope = parseJson(data.message)
if not isValid(envelope) then return
command = remoteCommand.parseMessage(envelope)
kind = command.kind
if kind = "keepalive"
' Jellyfin asks us to keep the session alive; send one now and poll on HALF the requested
' interval (a safety margin against the server's reap window), floored at 1s.
intervalMs = command.intervalSeconds * 1000 / 2
if intervalMs < 1000 then intervalMs = 1000
m.keepAliveIntervalMs = intervalMs
m.ws.send = [remoteProtocol.keepAliveFrame()]
return
end if
' Marshal every dispatchable command to the main thread. keepalive already returned above; only
' "ignore" is dropped here — so a new dispatchable kind (route/goback/message/…) is forwarded
' without needing this gate updated again.
if kind <> "ignore"
marshalCommand(command)
end if
' kind = "ignore" -> dropped (KeepAlive echo, Sessions, volume, unknown types).
end sub
' Marshal a normalized command to the main thread. Drops the bulky original 'raw' Data first (a whole
' ItemIds list for a series/album can be large) — the normalized fields carry everything the main
' thread needs. Shared by both transports; callers gate out keepalive/ignore before calling.
sub marshalCommand(command as object)
command.delete("raw")
m.top.dispatchCommand = command
end sub
' Best-effort teardown of the socket child + its observers.
sub closeSocket()
if not isValid(m.ws) then return
m.ws.unobserveField("on_open")
m.ws.unobserveField("on_message")
m.ws.unobserveField("on_close")
m.ws.unobserveField("on_error")
m.ws.unobserveField("ready_state")
' STOP releases the child's thread — dropping the reference alone leaves its loop running.
' No close frame is requested here. This only runs after the connection is already terminal (the
' client has left OPEN in every path that gets us here), so `close = [1000]` had nothing to send
' even before the STOP — and a STOP on the next line kills the thread that would deliver it
' anyway. Teardown is abrupt by design; see `remotecontrol-socket-abrupt-teardown` in tech-debt.md
' for the graceful-close direction and why it is not worth the sequencing today.
m.ws.control = "STOP"
m.top.socketNode = invalid
m.ws = invalid
end sub
' HTTPS long-poll transport (#667). Roku can't ws:// a secure server, so on https we consume the
' companion plugin's authenticated long-poll command channel over TLS (roUrlTransfer). The poll
' request ITSELF is the keepalive (it refreshes server-side session activity), so — unlike the ws
' path — there's no client KeepAlive to send. See docs/architecture/remote-control-longpoll-contract.md.
sub runLongPollReceiver(serverUrl as string)
PROBE_TIMEOUT_MS = 8000 ' presence probe (matches ServerReachableTask's pre-flight budget)
POLL_WAIT_MS = 25000 ' server holds each poll up to this before returning 204
POLL_TIMEOUT_MS = 35000 ' client transfer timeout — longer than POLL_WAIT_MS so a 204 always lands first
probeUrl = remoteProtocol.buildProbeUrl(serverUrl)
if probeUrl = ""
m.log.info("long-poll receiver disabled: no server url")
return
end if
' Presence probe: 200 => the plugin is installed (long-poll available); anything else => the
' secure server has no cast plugin, so JellyRock stays dark (unchanged pre-#667 https behavior).
probe = httpGet(probeUrl, PROBE_TIMEOUT_MS)
if probe.code <> 200
m.log.info("long-poll receiver disabled: plugin probe returned", probe.code)
return
end if
' Contract-version gate. The plugin advertises its wire-contract version in the probe body; we run
' the loop only for a version we implement, else stay dark rather than risk acting on a command
' shape we might misread. See remote-control-longpoll-contract.md (Versioning).
version = remoteProtocol.parseContractVersion(probe.body)
if version <> remoteProtocol.CONTRACT_VERSION
m.log.info("long-poll receiver disabled: unsupported contract version", version)
return
end if
m.log.info("long-poll plugin present; starting poll loop")
if remoteProtocol.buildLongPollUrl(serverUrl, POLL_WAIT_MS) = "" then return
' Snapshot the token so a poll can detect a server-side rotation and stop cleanly (mirrors the ws path).
m.openedWithToken = m.global.user.authToken
' At-least-once delivery state (contract v1). m.lastAckId is our cumulative ack — the newest MessageId
' we've received — sent on every poll so the plugin can drop confirmed commands and redeliver the rest.
' m.dispatchedIds/order is a small dedupe ring so a REDELIVERED command (a poll whose response we lost,
' resent by the plugin) can't double-apply a relative verb like NextTrack. Both reset per receiver start;
' a fresh receiver pairs with a fresh server-side session, so there's nothing to carry over.
m.lastAckId = ""
m.dispatchedIds = {}
m.dispatchedOrder = []
attempt = 0
while true
' Token rotation / logout guard. control="STOP" (SignOut / server switch) terminates this thread
' outright; this catches a server-side token rotation between polls.
currentToken = m.global.user.authToken
if not isValidAndNotEmpty(currentToken) or currentToken <> m.openedWithToken
m.log.info("session ended; long-poll receiver stopping")
return
end if
' Rebuilt each poll: the URL carries our current cumulative ackId, which advances as we receive.
pollUrl = remoteProtocol.buildLongPollUrl(serverUrl, POLL_WAIT_MS, m.lastAckId)
' Time the poll so an abnormally fast EMPTY response (a plugin that doesn't honor the hold) can be
' floored below — a compliant plugin holds ~POLL_WAIT_MS, so the floor never fires on it.
pollTimer = CreateObject("roTimespan")
resp = httpGet(pollUrl, POLL_TIMEOUT_MS)
if resp.code = 200
' Queued command(s). Dispatch in order, then re-poll immediately (no backoff on success). A 200
' that dispatched nothing (empty array / garbled body) is abnormal — floor it like a fast 204 so
' a misbehaving plugin can't spin this loop with no backoff. A real command 200 returns fast and
' is NOT floored, so command latency is untouched.
attempt = 0
if dispatchBatch(resp.body) = 0 then throttleEmptyPoll(pollTimer)
else if resp.code = 204
' Healthy empty hold — re-poll immediately (floored only if the server returned it near-instantly
' instead of holding, which a compliant plugin never does).
attempt = 0
throttleEmptyPoll(pollTimer)
else if resp.code = 401 or resp.code = 404
' 401 = token no longer valid; 404 = plugin uninstalled mid-session. Either way, stop.
m.log.info("long-poll receiver stopping; response", resp.code)
return
else
' Transient error / timeout (code 0) -> exponential backoff (1s..30s), then retry. A persistently
' unreachable server keeps the counter climbing so it can't hot-loop.
backoffMs = remoteProtocol.nextBackoffMs(attempt)
m.log.info("long-poll error; retrying", "code", resp.code, "inMs", backoffMs)
sleep(backoffMs)
attempt++
end if
end while
end sub
' Authenticated async GET on this Task thread. Blocks up to timeoutMs for the response. Returns
' { code, body }; code = 0 on a failed initiation or a timeout. A fresh roUrlTransfer + roMessagePort
' per call avoids cross-call event bleed. NEVER logs the url — the Authorization header carries the
' token. Mirrors the raw-transfer pattern in captionTask.bs / ServerReachableTask.bs.
function httpGet(url as string, timeoutMs as integer) as object
req = CreateObject("roUrlTransfer")
port = CreateObject("roMessagePort")
req.SetMessagePort(port)
req.SetUrl(url)
if LCase(url).Left(8) = "https://" then req.SetCertificatesFile("common:/certs/ca-bundle.crt")
req.AddHeader("Authorization", buildAuthHeader())
if not req.AsyncGetToString() then return { code: 0, body: "" }
event = wait(timeoutMs, port)
if type(event) = "roUrlEvent"
return { code: event.GetResponseCode(), body: event.GetString() }
end if
req.AsyncCancel() ' timed out -> abandon this transfer
return { code: 0, body: "" }
end function
' Parse a long-poll 200 body (a JSON array of { MessageType, Data, MessageId } envelopes) and marshal
' each dispatchable command to the main thread, in array order. Returns how many commands were actually
' dispatched (0 for an empty array, a garbled body, or an all-ignore batch) so the caller can floor a
' pointless fast 200 against a hot-loop.
function dispatchBatch(body as string) as integer
if not isValidAndNotEmpty(body) then return 0
envelopes = parseJson(body)
if not isValid(envelopes) or type(envelopes) <> "roArray" then return 0
dispatched = 0
for each envelope in envelopes
if dispatchEnvelope(envelope) then dispatched++
end for
return dispatched
end function
' Normalize one long-poll envelope and marshal it if dispatchable; returns true iff it was dispatched.
' The long-poll contract never carries KeepAlive/ForceKeepAlive (the poll request is the keepalive), so
' keepalive is treated like ignore here — there's no socket to answer on. Reuses the same pure
' parseMessage as the ws path.
function dispatchEnvelope(envelope as object) as boolean
if not isValid(envelope) then return false
command = remoteCommand.parseMessage(envelope)
' Advance the cumulative ack to the newest id we've received — of ANY kind, including ignore/keepalive —
' so the plugin can drop it too. The body arrived atomically (roUrlTransfer is all-or-nothing), so the
' last id in the array acks the whole batch on our next poll.
if isValidAndNotEmpty(command.messageId) then m.lastAckId = command.messageId
kind = command.kind
if kind = "ignore" or kind = "keepalive" then return false
' Dedupe: at-least-once redelivers a batch whose response we lost, so a command we've ALREADY enacted
' can arrive again — drop it, or a relative verb (NextTrack/Rewind/FastForward) would double-apply.
if hasDispatched(command.messageId) then return false
marshalCommand(command)
recordDispatched(command.messageId)
return true
end function
' Dedupe ring for at-least-once redelivery. Records ids we've actually dispatched and evicts the oldest
' past a bounded window — a duplicate only ever arrives back-to-back (a redelivery on the very next poll),
' so a small window is ample. An empty id (the ws path, or a malformed envelope) is never tracked, so it's
' never deduped — those transports/paths don't redeliver.
function hasDispatched(messageId as dynamic) as boolean
if not isValidAndNotEmpty(messageId) then return false
return m.dispatchedIds.doesExist(messageId)
end function
sub recordDispatched(messageId as dynamic)
DEDUPE_WINDOW = 64
if not isValidAndNotEmpty(messageId) or m.dispatchedIds.doesExist(messageId) then return
m.dispatchedIds[messageId] = true
m.dispatchedOrder.push(messageId)
if m.dispatchedOrder.count() > DEDUPE_WINDOW
oldest = m.dispatchedOrder.shift()
m.dispatchedIds.delete(oldest)
end if
end sub
' Hot-loop floor for an abnormally fast EMPTY poll. A compliant plugin holds an empty poll for
' ~POLL_WAIT_MS, so this never fires against it; a misbehaving plugin that returns an empty 200/204
' near-instantly would otherwise spin the poll loop with no backoff. Sleep out the remainder of a
' minimum interval. Only ever called on the empty path, so it never delays a real command.
sub throttleEmptyPoll(pollTimer as object)
POLL_MIN_EMPTY_MS = 1000
elapsed = pollTimer.TotalMilliseconds()
if elapsed < POLL_MIN_EMPTY_MS then sleep(POLL_MIN_EMPTY_MS - elapsed)
end sub