EARHORNMUSIC UPnP · Android TV · Phones · Android Auto · GNU/Linux
Release 0.5.0

CastStreamProxy.kt

412 lines · 17.1 KB

Changed in: 0.5.0

1package dev.stave.cast
2
3import android.content.Context
4import android.util.Log
5import dagger.hilt.android.qualifiers.ApplicationContext
6import okhttp3.OkHttpClient
7import okhttp3.Request
8import java.io.BufferedReader
9import java.io.InputStreamReader
10import java.io.OutputStream
11import java.net.Inet4Address
12import java.net.NetworkInterface
13import java.net.ServerSocket
14import java.net.Socket
15import java.net.URLDecoder
16import java.net.URLEncoder
17import java.util.concurrent.TimeUnit
18import javax.inject.Inject
19import javax.inject.Singleton
20
21/**
22 * Lightweight HTTP proxy between UPnP media servers (MinimServer, etc.)
23 * and Chromecast speaker groups. Cast group leaders make HTTP Range
24 * requests to synchronize playback across members. When fetching the
25 * raw UPnP URL directly, the server's responses can cause broken audio
26 * on group members. The proxy normalizes HTTP behavior — proper
27 * Content-Length, Accept-Ranges, Content-Range headers on every
28 * response — so the group leader can seek and sync reliably.
29 *
30 * Single speakers don't need this; they handle server quirks
31 * internally. The proxy is only used when [CastIntegration] detects
32 * the target is a speaker group.
33 *
34 * Same architecture as BubbleUPnP's internal cast-proxy.
35 */
36@Singleton
37class CastStreamProxy @Inject constructor(
38 @ApplicationContext private val context: Context,
39) {
40 private val tag = "EarhornProxy"
41
42 private val client = OkHttpClient.Builder()
43 .connectTimeout(10, TimeUnit.SECONDS)
44 .readTimeout(120, TimeUnit.SECONDS)
45 .build()
46
47 private var server: ServerSocket? = null
48 private var port: Int = 0
49 @Volatile private var running = false
50
51 /** design decision 2026-09-02 (I.32 / review N.4): the proxy used to
52 * relay ANY url= a LAN client named and serve ANY content:// URI the
53 * app could open, from the first cast until the process died. Now every
54 * link the app hands the receiver carries a random token; requests
55 * without a known token get 403; and [stop] is called on cast end. */
56 private val streamTokens = java.util.concurrent.ConcurrentHashMap<String, String>() // token -> upstream URL
57 private val localTokens = java.util.concurrent.ConcurrentHashMap<String, Pair<String, String>>() // token -> (content uri, mime)
58
59 /** The upstream URL or content URI a proxied link stands for, or null if
60 * the link is not one this proxy issued. Lets the resume check compare
61 * the receiver's contentId against the phone's queue when group casts
62 * go through tokened links (review C.3, 2026-09-02). */
63 fun originalFor(proxiedUrl: String): String? {
64 val token = proxiedUrl.substringAfter("t=", "").substringBefore('&')
65 if (token.isEmpty()) return null
66 return streamTokens[token] ?: localTokens[token]?.first
67 }
68 private val rng = java.security.SecureRandom()
69
70 private fun newToken(): String {
71 val b = ByteArray(16); rng.nextBytes(b)
72 return b.joinToString("") { "%02x".format(it) }
73 }
74
75 /** Start the proxy on a random port, bound to the LAN interface only
76 * when it can be determined. Idempotent. */
77 fun start() {
78 if (running) return
79 val bindAddr = runCatching { getLocalIp()?.let { java.net.InetAddress.getByName(it) } }.getOrNull()
80 val ss = if (bindAddr != null) ServerSocket(0, 50, bindAddr) else ServerSocket(0)
81 port = ss.localPort
82 server = ss
83 running = true
84 Log.i(tag, "Started on port $port (bound to ${bindAddr?.hostAddress ?: "all interfaces"})")
85 Thread(null, {
86 while (running && !ss.isClosed) {
87 try {
88 val socket = ss.accept()
89 Thread(null, { handleConnection(socket) }, "cast-proxy-req").start()
90 } catch (e: Exception) {
91 if (running) Log.w(tag, "Accept error", e)
92 }
93 }
94 }, "cast-proxy-accept").start()
95 }
96
97 fun stop() {
98 running = false
99 runCatching { server?.close() }
100 server = null
101 streamTokens.clear()
102 localTokens.clear()
103 Log.i(tag, "Stopped")
104 }
105
106 /** Wrap a UPnP resource URL so the Cast device fetches from this
107 * proxy instead. Starts the proxy if not already running. */
108 fun proxyUrl(originalUrl: String): String {
109 start()
110 val ip = getLocalIp() ?: "127.0.0.1"
111 val token = newToken()
112 streamTokens[token] = originalUrl
113 return "http://$ip:$port/stream?t=$token"
114 }
115
116 /** Create a proxy URL for a local content:// URI so the Chromecast
117 * can stream local files over HTTP. [contentUri] is the MediaStore
118 * content:// URI; [mimeType] is passed through as Content-Type
119 * so the Cast receiver knows how to decode. */
120 fun localProxyUrl(contentUri: String, mimeType: String?): String {
121 start()
122 val ip = getLocalIp() ?: "127.0.0.1"
123 val token = newToken()
124 localTokens[token] = contentUri to (mimeType ?: "audio/mpeg")
125 return "http://$ip:$port/local?t=$token"
126 }
127
128 private fun handleConnection(socket: Socket) {
129 try {
130 socket.soTimeout = 30_000
131 socket.sendBufferSize = 262144 // 256KB — matches the read buffer
132 socket.tcpNoDelay = true // disable Nagle — send chunks immediately
133 val input = BufferedReader(InputStreamReader(socket.getInputStream()))
134 val output = socket.getOutputStream()
135
136 val requestLine = input.readLine() ?: return
137 val parts = requestLine.split(' ')
138 val method = parts.getOrNull(0) ?: "GET"
139 val path = parts.getOrNull(1) ?: ""
140
141 // Read headers
142 val headers = mutableMapOf<String, String>()
143 var line = input.readLine()
144 while (line != null && line.isNotBlank()) {
145 val colon = line.indexOf(':')
146 if (colon > 0) {
147 headers[line.substring(0, colon).trim().lowercase()] =
148 line.substring(colon + 1).trim()
149 }
150 line = input.readLine()
151 }
152
153 // Extract Range header — needed by both local and UPnP paths
154 val rangeHeader = headers["range"]
155
156 // Only links the app itself issued are served: /local?t=… and
157 // /stream?t=…. Anything else is refused.
158 val token = path.substringAfter("t=", "").substringBefore('&').substringBefore(' ')
159
160 // ── Local file path: /local?t=<token> ──
161 if (path.startsWith("/local")) {
162 val entry = localTokens[token]
163 if (entry == null) {
164 sendError(output, 403, "Forbidden")
165 return
166 }
167 val contentUri = android.net.Uri.parse(entry.first)
168 val contentType = entry.second
169 Log.i(tag, "$method LOCAL $contentUri mime=$contentType range=$rangeHeader")
170
171 handleLocalFile(socket, output, method, contentUri, contentType, rangeHeader)
172 return
173 }
174
175 // ── UPnP proxy path: /stream?t=<token> ──
176 val originalUrl = streamTokens[token]
177 if (originalUrl == null) {
178 sendError(output, 403, "Forbidden")
179 return
180 }
181
182 Log.i(tag, "$method $originalUrl range=$rangeHeader")
183
184 // First: HEAD request to get Content-Length if we don't have Range
185 // For Range requests, pass through directly
186 val reqBuilder = Request.Builder().url(originalUrl)
187 if (rangeHeader != null) {
188 reqBuilder.header("Range", rangeHeader)
189 }
190 if (method == "HEAD") {
191 reqBuilder.head()
192 }
193
194 val response = client.newCall(reqBuilder.build()).execute()
195 val body = response.body
196 val contentType = response.header("Content-Type") ?: "application/octet-stream"
197 val contentLength = response.header("Content-Length")
198 val contentRange = response.header("Content-Range")
199 val acceptRanges = response.header("Accept-Ranges") ?: "bytes"
200
201 val statusCode = response.code
202 // Relay the real reason phrase (I.33: everything used to be "OK").
203 val statusText = response.message.ifBlank {
204 when (statusCode) { 200 -> "OK"; 206 -> "Partial Content"; 404 -> "Not Found"; else -> "Error" }
205 }
206
207 // Build response
208 val sb = StringBuilder()
209 sb.append("HTTP/1.1 $statusCode $statusText\r\n")
210 sb.append("Content-Type: $contentType\r\n")
211 sb.append("Accept-Ranges: bytes\r\n")
212 sb.append("Access-Control-Allow-Origin: *\r\n")
213 if (contentLength != null) {
214 sb.append("Content-Length: $contentLength\r\n")
215 }
216 if (contentRange != null) {
217 sb.append("Content-Range: $contentRange\r\n")
218 }
219 sb.append("Connection: close\r\n")
220 sb.append("\r\n")
221
222 output.write(sb.toString().toByteArray())
223 output.flush()
224
225 // Stream body for GET (not HEAD)
226 if (method != "HEAD" && body != null) {
227 val buf = ByteArray(32768)
228 val bodyStream = body.byteStream()
229 try {
230 var read: Int
231 while (bodyStream.read(buf).also { read = it } != -1) {
232 output.write(buf, 0, read)
233 }
234 output.flush()
235 } catch (e: Exception) {
236 // Client disconnected — normal during seek
237 Log.d(tag, "Stream interrupted: ${e.message}")
238 } finally {
239 bodyStream.close()
240 }
241 }
242 body?.close()
243 response.close()
244 } catch (e: Exception) {
245 Log.w(tag, "Connection error", e)
246 } finally {
247 runCatching { socket.close() }
248 }
249 }
250
251 /** Serve a local content:// file to the Chromecast via HTTP.
252 * Supports Range requests so Cast can seek within the track. */
253 private fun handleLocalFile(
254 socket: Socket,
255 output: OutputStream,
256 method: String,
257 contentUri: android.net.Uri,
258 contentType: String,
259 rangeHeader: String?,
260 ) {
261 try {
262 val afd = context.contentResolver.openAssetFileDescriptor(contentUri, "r")
263 if (afd == null) {
264 sendError(output, 404, "File not found")
265 return
266 }
267
268 val totalSize = afd.length // -1 if unknown
269
270 val rangeStart: Long
271 val rangeEnd: Long
272 val statusCode: Int
273 val statusText: String
274
275 if (rangeHeader != null && totalSize > 0) {
276 val rangeParts = rangeHeader.removePrefix("bytes=").split('-')
277 rangeStart = rangeParts.getOrNull(0)?.toLongOrNull() ?: 0L
278 rangeEnd = rangeParts.getOrNull(1)?.toLongOrNull()
279 ?.coerceAtMost(totalSize - 1) ?: (totalSize - 1)
280 statusCode = 206
281 statusText = "Partial Content"
282 } else {
283 rangeStart = 0L
284 rangeEnd = if (totalSize > 0) totalSize - 1 else -1L
285 statusCode = 200
286 statusText = "OK"
287 }
288
289 val contentLength = if (rangeEnd >= 0) rangeEnd - rangeStart + 1 else totalSize
290 val sb = StringBuilder()
291 sb.append("HTTP/1.1 $statusCode $statusText\r\n")
292 sb.append("Content-Type: $contentType\r\n")
293 sb.append("Accept-Ranges: bytes\r\n")
294 sb.append("Access-Control-Allow-Origin: *\r\n")
295 if (contentLength > 0) sb.append("Content-Length: $contentLength\r\n")
296 if (statusCode == 206 && totalSize > 0) {
297 sb.append("Content-Range: bytes $rangeStart-$rangeEnd/$totalSize\r\n")
298 }
299 sb.append("Connection: close\r\n\r\n")
300
301 // Wrap in BufferedOutputStream — raw socket writes of small
302 // chunks thrash the TCP stack and cause Cast buffer underruns
303 // on high-bitrate formats like FLAC.
304 val buffered = java.io.BufferedOutputStream(output, 262144)
305 buffered.write(sb.toString().toByteArray())
306 buffered.flush()
307
308 if (method == "HEAD") {
309 afd.close()
310 return
311 }
312
313 // Stream from the file descriptor directly (no double-open).
314 // FileInputStream from the AFD supports efficient skip via
315 // FileChannel.position() under the hood.
316 val stream = afd.createInputStream()
317 if (stream == null) {
318 Log.w(tag, "Failed to open input stream for $contentUri")
319 afd.close()
320 return
321 }
322
323 stream.use { input ->
324 // Use FileChannel.position() for O(1) seeking instead of
325 // InputStream.skip() which reads and discards bytes — on a
326 // 50MB FLAC file, skip(40MB) reads 40MB of data before sending
327 // anything, causing the Cast receiver to starve.
328 if (rangeStart > 0 && input is java.io.FileInputStream) {
329 input.channel.position(rangeStart)
330 } else if (rangeStart > 0) {
331 // Fallback for non-FileInputStream (shouldn't happen
332 // with ContentResolver but be safe)
333 var toSkip = rangeStart
334 while (toSkip > 0) {
335 val skipped = input.skip(toSkip)
336 if (skipped <= 0) break
337 toSkip -= skipped
338 }
339 }
340
341 // 256KB read buffer. Flush after every write so the Cast
342 // receiver gets data continuously rather than in bursts.
343 // Without per-write flushes, BufferedOutputStream holds
344 // partial chunks until 256KB accumulates, creating gaps
345 // in the TCP stream that the receiver hears as dropouts
346 // on data-heavy FLAC passages.
347 val buf = ByteArray(262144)
348 var remaining = if (contentLength > 0) contentLength else Long.MAX_VALUE
349 try {
350 while (remaining > 0) {
351 val toRead = minOf(buf.size.toLong(), remaining).toInt()
352 val read = input.read(buf, 0, toRead)
353 if (read == -1) break
354 buffered.write(buf, 0, read)
355 buffered.flush()
356 remaining -= read
357 }
358 } catch (e: Exception) {
359 Log.d(tag, "Local stream interrupted: ${e.message}")
360 }
361 }
362 afd.close()
363 } catch (e: Exception) {
364 Log.w(tag, "Local file error for $contentUri", e)
365 runCatching { sendError(output, 500, "Internal error") }
366 } finally {
367 runCatching { socket.close() }
368 }
369 }
370
371 private fun sendError(output: OutputStream, code: Int, message: String) {
372 val body = message.toByteArray()
373 val response = "HTTP/1.1 $code $message\r\n" +
374 "Content-Type: text/plain\r\n" +
375 "Content-Length: ${body.size}\r\n" +
376 "Connection: close\r\n" +
377 "\r\n"
378 output.write(response.toByteArray())
379 output.write(body)
380 output.flush()
381 }
382
383 /** Phone's local IPv4 address on WiFi/ethernet. */
384 private fun getLocalIp(): String? {
385 try {
386 for (iface in NetworkInterface.getNetworkInterfaces()) {
387 if (iface.isLoopback || !iface.isUp) continue
388 // Prefer wlan/eth interfaces
389 val name = iface.name.lowercase()
390 if (!name.startsWith("wlan") && !name.startsWith("eth") &&
391 !name.startsWith("en")) continue
392 for (addr in iface.inetAddresses) {
393 if (addr is Inet4Address && !addr.isLoopbackAddress) {
394 return addr.hostAddress
395 }
396 }
397 }
398 // Fallback: any non-loopback IPv4
399 for (iface in NetworkInterface.getNetworkInterfaces()) {
400 if (iface.isLoopback || !iface.isUp) continue
401 for (addr in iface.inetAddresses) {
402 if (addr is Inet4Address && !addr.isLoopbackAddress) {
403 return addr.hostAddress
404 }
405 }
406 }
407 } catch (e: Exception) {
408 Log.w(tag, "Failed to get local IP", e)
409 }
410 return null
411 }
412}