feat: add Chromecast playback support
This commit is contained in:
@@ -0,0 +1,806 @@
|
||||
import Foundation
|
||||
import Combine
|
||||
@preconcurrency import Network
|
||||
import Security
|
||||
|
||||
struct CastDevice: Identifiable, Equatable, Sendable {
|
||||
let id: String
|
||||
let name: String
|
||||
let host: String
|
||||
let port: UInt16
|
||||
}
|
||||
|
||||
@MainActor
|
||||
final class CastDiscovery: NSObject, ObservableObject, @preconcurrency NetServiceBrowserDelegate,
|
||||
@preconcurrency NetServiceDelegate {
|
||||
@Published private(set) var devices: [CastDevice] = []
|
||||
@Published private(set) var error: String?
|
||||
private var browser: NetServiceBrowser?
|
||||
private var services: [String: NetService] = [:]
|
||||
private var resolved: [String: CastDevice] = [:]
|
||||
|
||||
func start() {
|
||||
if error != nil { stop() }
|
||||
guard browser == nil else { return }
|
||||
error = nil
|
||||
let browser = NetServiceBrowser()
|
||||
self.browser = browser
|
||||
browser.delegate = self
|
||||
browser.searchForServices(ofType: "_googlecast._tcp.", inDomain: "local.")
|
||||
}
|
||||
|
||||
func stop() {
|
||||
browser?.stop()
|
||||
browser?.delegate = nil
|
||||
browser = nil
|
||||
for service in services.values { service.stop(); service.delegate = nil }
|
||||
services.removeAll()
|
||||
resolved.removeAll()
|
||||
devices.removeAll()
|
||||
}
|
||||
|
||||
private func key(_ service: NetService) -> String { service.domain + service.type + service.name }
|
||||
private func publish() {
|
||||
// Multiple advertised interfaces may resolve to the same receiver ID.
|
||||
var unique: [String: CastDevice] = [:]
|
||||
for device in resolved.values { unique[device.id] = device }
|
||||
devices = unique.values.sorted { $0.name.localizedStandardCompare($1.name) == .orderedAscending }
|
||||
}
|
||||
|
||||
func netServiceBrowser(_ browser: NetServiceBrowser, didFind service: NetService, moreComing: Bool) {
|
||||
guard browser === self.browser, services.count < 128 else { return }
|
||||
let key = self.key(service)
|
||||
guard services[key] == nil else { return }
|
||||
services[key] = service
|
||||
service.delegate = self
|
||||
service.resolve(withTimeout: 5)
|
||||
}
|
||||
|
||||
func netServiceBrowser(_ browser: NetServiceBrowser, didRemove service: NetService, moreComing: Bool) {
|
||||
guard browser === self.browser else { return }
|
||||
let key = self.key(service)
|
||||
services.removeValue(forKey: key)?.stop()
|
||||
resolved.removeValue(forKey: key)
|
||||
publish()
|
||||
}
|
||||
|
||||
func netServiceBrowser(_ browser: NetServiceBrowser, didNotSearch errorDict: [String: NSNumber]) {
|
||||
guard browser === self.browser else { return }
|
||||
error = "Cast discovery failed. Check Local Network permission and your Wi-Fi connection."
|
||||
}
|
||||
|
||||
func netServiceDidResolveAddress(_ sender: NetService) {
|
||||
let key = self.key(sender)
|
||||
guard services[key] === sender, let host = sender.hostName,
|
||||
sender.port > 0, sender.port <= Int(UInt16.max) else { return }
|
||||
let txt = sender.txtRecordData().map(NetService.dictionary(fromTXTRecord:)) ?? [:]
|
||||
func text(_ key: String) -> String? {
|
||||
txt[key].flatMap { String(data: $0, encoding: .utf8) }.flatMap { $0.isEmpty ? nil : $0 }
|
||||
}
|
||||
resolved[key] = CastDevice(id: text("id") ?? key, name: text("fn") ?? sender.name,
|
||||
host: host, port: UInt16(sender.port))
|
||||
publish()
|
||||
}
|
||||
|
||||
func netService(_ sender: NetService, didNotResolve errorDict: [String: NSNumber]) {
|
||||
let key = self.key(sender)
|
||||
guard services[key] === sender else { return }
|
||||
services.removeValue(forKey: key)?.stop()
|
||||
}
|
||||
}
|
||||
|
||||
struct CastStatus: Sendable {
|
||||
let position: Double
|
||||
let duration: Double
|
||||
let playing: Bool
|
||||
let buffering: Bool
|
||||
let ended: Bool
|
||||
let volume: Double
|
||||
let seekable: Bool
|
||||
let subtitle: Int?
|
||||
let seekPending: Bool
|
||||
let playbackPending: Bool
|
||||
}
|
||||
|
||||
enum CastError: Error, LocalizedError, Sendable {
|
||||
case protocolError, disconnected, cancelled, timeout(String), rejected(String), invalidMedia, queueFull
|
||||
var errorDescription: String? {
|
||||
switch self {
|
||||
case .protocolError: "The Cast receiver sent an invalid message."
|
||||
case .disconnected: "The Cast receiver disconnected."
|
||||
case .cancelled: "The Cast operation was cancelled."
|
||||
case .timeout(let operation): "The Cast receiver did not acknowledge \(operation) in time."
|
||||
case .rejected(let operation): "The Cast receiver rejected \(operation)."
|
||||
case .invalidMedia: "The media or subtitle selection cannot be cast."
|
||||
case .queueFull: "Too many Cast commands are waiting for a response."
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Cast v2's original (unchunked) envelope, from Open Screen's cast_channel.proto:
|
||||
// https://chromium.googlesource.com/openscreen/+/refs/heads/main/cast/common/channel/proto/cast_channel.proto
|
||||
// Media transport messages are limited to 64 KiB by https://developers.google.com/cast/docs/media/messages.
|
||||
enum CastWire {
|
||||
static let maximumFrame = 64 * 1024
|
||||
static let connection = "urn:x-cast:com.google.cast.tp.connection"
|
||||
static let heartbeat = "urn:x-cast:com.google.cast.tp.heartbeat"
|
||||
static let receiver = "urn:x-cast:com.google.cast.receiver"
|
||||
static let media = "urn:x-cast:com.google.cast.media"
|
||||
|
||||
struct Message: Sendable {
|
||||
let source: String
|
||||
let destination: String
|
||||
let namespace: String
|
||||
let payload: Data
|
||||
}
|
||||
|
||||
static func varint(_ value: UInt64, into data: inout Data) {
|
||||
var value = value
|
||||
repeat {
|
||||
let byte = UInt8(value & 127)
|
||||
value >>= 7
|
||||
data.append(byte | (value == 0 ? 0 : 128))
|
||||
} while value != 0
|
||||
}
|
||||
|
||||
static func frame(_ message: Message) throws -> Data {
|
||||
var body = Data([8, 0]) // protocol_version CASTV2_1_0
|
||||
for (field, text) in [(2, message.source), (3, message.destination), (4, message.namespace)] {
|
||||
let bytes = Data(text.utf8)
|
||||
varint(UInt64(field << 3 | 2), into: &body)
|
||||
varint(UInt64(bytes.count), into: &body)
|
||||
body.append(bytes)
|
||||
}
|
||||
body.append(contentsOf: [40, 0, 50]) // payload_type STRING; payload_utf8
|
||||
varint(UInt64(message.payload.count), into: &body)
|
||||
body.append(message.payload)
|
||||
guard body.count <= maximumFrame else { throw CastError.protocolError }
|
||||
let count = UInt32(body.count)
|
||||
var frame = Data([UInt8(count >> 24), UInt8((count >> 16) & 255),
|
||||
UInt8((count >> 8) & 255), UInt8(count & 255)])
|
||||
frame.append(body)
|
||||
return frame
|
||||
}
|
||||
|
||||
static func decode(_ data: Data) throws -> Message {
|
||||
guard !data.isEmpty, data.count <= maximumFrame else { throw CastError.protocolError }
|
||||
var offset = 0
|
||||
func integer() throws -> UInt64 {
|
||||
var value: UInt64 = 0
|
||||
for shift in stride(from: 0, through: 63, by: 7) {
|
||||
guard offset < data.count else { throw CastError.protocolError }
|
||||
let byte = data[data.startIndex + offset]
|
||||
offset += 1
|
||||
guard shift != 63 || byte <= 1 else { throw CastError.protocolError }
|
||||
value |= UInt64(byte & 127) << shift
|
||||
if byte & 128 == 0 { return value }
|
||||
}
|
||||
throw CastError.protocolError
|
||||
}
|
||||
var numbers: [Int: UInt64] = [:]
|
||||
var bytes: [Int: Data] = [:]
|
||||
while offset < data.count {
|
||||
let key = try integer()
|
||||
guard key >> 3 > 0, key >> 3 <= UInt64(Int32.max) else { throw CastError.protocolError }
|
||||
let field = Int(key >> 3)
|
||||
switch key & 7 {
|
||||
case 0:
|
||||
let value = try integer()
|
||||
if field == 1 || field == 5 || field == 8 { numbers[field] = value }
|
||||
case 1, 5:
|
||||
let size = key & 7 == 1 ? 8 : 4
|
||||
guard size <= data.count - offset else { throw CastError.protocolError }
|
||||
offset += size
|
||||
case 2:
|
||||
let size = try integer()
|
||||
guard size <= UInt64(data.count - offset) else { throw CastError.protocolError }
|
||||
let end = offset + Int(size)
|
||||
if [2, 3, 4, 6].contains(field) {
|
||||
bytes[field] = data.subdata(in: (data.startIndex + offset)..<(data.startIndex + end))
|
||||
}
|
||||
offset = end
|
||||
default: throw CastError.protocolError
|
||||
}
|
||||
}
|
||||
guard numbers[1] == 0, numbers[5] == 0, numbers[8, default: 0] == 0,
|
||||
let sourceData = bytes[2], let source = String(data: sourceData, encoding: .utf8), !source.isEmpty,
|
||||
let destinationData = bytes[3], let destination = String(data: destinationData, encoding: .utf8), !destination.isEmpty,
|
||||
let namespaceData = bytes[4], let namespace = String(data: namespaceData, encoding: .utf8), !namespace.isEmpty,
|
||||
let payload = bytes[6], String(data: payload, encoding: .utf8) != nil else {
|
||||
throw CastError.protocolError
|
||||
}
|
||||
return Message(source: source, destination: destination, namespace: namespace, payload: payload)
|
||||
}
|
||||
|
||||
struct Stream {
|
||||
private var buffer = Data()
|
||||
mutating func append(_ data: Data) throws -> [Message] {
|
||||
// Network reads are <= 16 KiB. Bound incomplete input even before its length is known.
|
||||
guard data.count <= maximumFrame, buffer.count + data.count <= maximumFrame * 2 + 4 else {
|
||||
throw CastError.protocolError
|
||||
}
|
||||
buffer.append(data)
|
||||
var messages: [Message] = []
|
||||
while buffer.count >= 4 {
|
||||
let i = buffer.startIndex
|
||||
let length = buffer[i..<i + 4].reduce(0) { ($0 << 8) | Int($1) }
|
||||
guard length > 0, length <= maximumFrame else { throw CastError.protocolError }
|
||||
guard buffer.count >= length + 4 else { break }
|
||||
messages.append(try decode(buffer.subdata(in: (i + 4)..<(i + 4 + length))))
|
||||
buffer.removeFirst(length + 4)
|
||||
}
|
||||
return messages
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@MainActor
|
||||
final class CastSession {
|
||||
var onStatus: ((CastStatus) -> Void)?
|
||||
var onDisconnect: ((String?) -> Void)?
|
||||
var onError: ((String) -> Void)?
|
||||
var localHost: String? {
|
||||
guard let endpoint = connection?.currentPath?.localEndpoint,
|
||||
case .hostPort(let host, _) = endpoint else { return nil }
|
||||
return String(describing: host)
|
||||
}
|
||||
|
||||
private struct Pending {
|
||||
let namespace: String
|
||||
let destination: String
|
||||
let operation: String
|
||||
let generation: UUID?
|
||||
let completion: ([String: Any]?, CastError?) -> Void
|
||||
let deadline: Task<Void, Never>
|
||||
}
|
||||
|
||||
private let makeConnection: (CastDevice) -> NWConnection
|
||||
private let requestTimeout: Duration
|
||||
private let loadTimeout: Duration
|
||||
private let heartbeatInterval: Duration
|
||||
private let sleep: @MainActor (Duration) async throws -> Void
|
||||
private var connection: NWConnection?
|
||||
private var connectionID = UUID()
|
||||
private let source = "terebi-" + UUID().uuidString
|
||||
private var stream = CastWire.Stream()
|
||||
private var ready: CheckedContinuation<Void, any Error>?
|
||||
private var readyDeadline: Task<Void, Never>?
|
||||
private var heartbeatTask: Task<Void, Never>?
|
||||
private var awaitingPong = false
|
||||
private var nextRequest = 0
|
||||
private var pending: [Int: Pending] = [:]
|
||||
private var writes: [Data] = []
|
||||
private var writing = false
|
||||
private var appTransport: String?
|
||||
private var appSession: String?
|
||||
private var mediaSession: Int?
|
||||
private var retiredSessions: Set<Int> = []
|
||||
private var retiredLoads: [String] = []
|
||||
private var contentID: String?
|
||||
private var mediaGeneration: UUID?
|
||||
private var tracks: [CastTextTrack] = []
|
||||
private var latestStatus: CastStatus?
|
||||
private var pendingSeek: Int?
|
||||
private var pendingPlayback: Int?
|
||||
var playbackPending: Bool { pendingPlayback != nil || replayTask != nil || loading }
|
||||
private var lastStatusPoll = 0
|
||||
private var newestMediaReply = 0
|
||||
private var loading = false
|
||||
private var loadedMedia: CastMediaDescription?
|
||||
private var loadedTitle = ""
|
||||
private var replayTask: Task<Void, Never>?
|
||||
private var replayDesiredPlaying = true
|
||||
private var receiverVolume = 1.0
|
||||
|
||||
init(requestTimeout: Duration = .seconds(8), loadTimeout: Duration = .seconds(25),
|
||||
heartbeatInterval: Duration = .seconds(5),
|
||||
sleep: @escaping @MainActor (Duration) async throws -> Void = { try await Task.sleep(for: $0) }) {
|
||||
makeConnection = { device in
|
||||
let tls = NWProtocolTLS.Options()
|
||||
sec_protocol_options_set_min_tls_protocol_version(tls.securityProtocolOptions, .TLSv12)
|
||||
// Cast receivers use self-signed TLS certificates, not public Web PKI. This is
|
||||
// encrypted local-network transport to the user's selected Bonjour endpoint,
|
||||
// NOT authenticated receiver identity. mDNS and this channel are LAN-trust;
|
||||
// never claim standard CA/hostname validation or send upstream credentials.
|
||||
// Cast device-auth (Google device roots) is intentionally not impersonated.
|
||||
sec_protocol_options_set_verify_block(tls.securityProtocolOptions, { @Sendable _, _, complete in
|
||||
complete(true)
|
||||
}, DispatchQueue.global(qos: .userInitiated))
|
||||
let parameters = NWParameters(tls: tls, tcp: NWProtocolTCP.Options())
|
||||
return NWConnection(host: NWEndpoint.Host(device.host),
|
||||
port: NWEndpoint.Port(rawValue: device.port)!, using: parameters)
|
||||
}
|
||||
self.requestTimeout = requestTimeout
|
||||
self.loadTimeout = loadTimeout
|
||||
self.heartbeatInterval = heartbeatInterval
|
||||
self.sleep = sleep
|
||||
}
|
||||
|
||||
|
||||
func connect(to device: CastDevice) async throws {
|
||||
disconnect(stopReceiver: false)
|
||||
guard device.port > 0, !device.host.isEmpty else { throw CastError.disconnected }
|
||||
let id = UUID()
|
||||
connectionID = id
|
||||
let socket = makeConnection(device)
|
||||
connection = socket
|
||||
stream = CastWire.Stream()
|
||||
try await withTaskCancellationHandler {
|
||||
try Task.checkCancellation()
|
||||
try await withCheckedThrowingContinuation { continuation in
|
||||
ready = continuation
|
||||
readyDeadline = Task { [weak self] in
|
||||
guard let self else { return }
|
||||
do { try await self.sleep(.seconds(12)) } catch { return }
|
||||
guard self.connectionID == id else { return }
|
||||
self.fail(.timeout("connection"))
|
||||
}
|
||||
socket.stateUpdateHandler = { [weak self] state in
|
||||
Task { @MainActor in
|
||||
guard let self, self.connectionID == id else { return }
|
||||
switch state {
|
||||
case .ready:
|
||||
self.receive(socket, id: id)
|
||||
self.beginHandshake()
|
||||
case .failed, .cancelled: self.fail(.disconnected)
|
||||
default: break
|
||||
}
|
||||
}
|
||||
}
|
||||
socket.start(queue: DispatchQueue.global(qos: .userInitiated))
|
||||
}
|
||||
} onCancel: {
|
||||
Task { @MainActor [weak self] in
|
||||
guard let self, self.connectionID == id else { return }
|
||||
self.fail(.cancelled)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func beginHandshake() {
|
||||
do {
|
||||
try send(["type": "CONNECT", "origin": [:], "userAgent": "Terebi"],
|
||||
namespace: CastWire.connection, destination: "receiver-0")
|
||||
startHeartbeat()
|
||||
try request(["type": "LAUNCH", "appId": "CC1AD845"], namespace: CastWire.receiver,
|
||||
destination: "receiver-0", operation: "receiver launch") { [weak self] json, error in
|
||||
guard let self else { return }
|
||||
if let error { self.fail(error); return }
|
||||
guard let status = json?["status"] as? [String: Any],
|
||||
let apps = status["applications"] as? [[String: Any]],
|
||||
let app = apps.first(where: { $0["appId"] as? String == "CC1AD845" }),
|
||||
let transport = app["transportId"] as? String, !transport.isEmpty,
|
||||
let session = app["sessionId"] as? String, !session.isEmpty else {
|
||||
self.fail(.rejected("receiver launch")); return
|
||||
}
|
||||
self.appTransport = transport
|
||||
self.appSession = session
|
||||
do {
|
||||
try self.send(["type": "CONNECT", "origin": [:]],
|
||||
namespace: CastWire.connection, destination: transport)
|
||||
} catch { self.fail(.protocolError); return }
|
||||
self.readyDeadline?.cancel()
|
||||
self.readyDeadline = nil
|
||||
let ready = self.ready
|
||||
self.ready = nil
|
||||
ready?.resume()
|
||||
}
|
||||
} catch { fail(error as? CastError ?? .protocolError) }
|
||||
}
|
||||
|
||||
func load(_ media: CastMediaDescription, title: String, position: Double,
|
||||
playing: Bool, subtitle: Int?) async throws {
|
||||
guard let transport = appTransport, connection != nil else { throw CastError.disconnected }
|
||||
guard position.isFinite, position >= 0, media.duration.isFinite,
|
||||
["http", "https"].contains(media.url.scheme?.lowercased() ?? ""),
|
||||
media.url.absoluteString.utf8.count <= 1024,
|
||||
subtitle == nil || media.tracks.indices.contains(subtitle!), media.tracks.count <= 128 else {
|
||||
throw CastError.invalidMedia
|
||||
}
|
||||
cancelMediaRequests()
|
||||
if let mediaSession {
|
||||
if retiredSessions.count >= 16 { retiredSessions.removeAll() }
|
||||
retiredSessions.insert(mediaSession)
|
||||
}
|
||||
if let mediaGeneration {
|
||||
if retiredLoads.count == 64 { retiredLoads.removeFirst() }
|
||||
retiredLoads.append(mediaGeneration.uuidString)
|
||||
}
|
||||
mediaSession = nil
|
||||
latestStatus = nil
|
||||
pendingSeek = nil
|
||||
pendingPlayback = nil
|
||||
newestMediaReply = 0
|
||||
let generation = UUID()
|
||||
mediaGeneration = generation
|
||||
contentID = media.url.absoluteString
|
||||
tracks = media.tracks
|
||||
loadedMedia = media
|
||||
loadedTitle = title
|
||||
loading = true
|
||||
var info: [String: Any] = [
|
||||
"contentId": media.url.absoluteString, "contentType": media.contentType,
|
||||
"streamType": media.live ? "LIVE" : "BUFFERED",
|
||||
"metadata": ["metadataType": 0, "title": String(title.prefix(1024))],
|
||||
"customData": ["terebiSession": generation.uuidString]
|
||||
]
|
||||
if media.duration > 0, !media.live { info["duration"] = media.duration }
|
||||
if !media.tracks.isEmpty {
|
||||
info["tracks"] = media.tracks.map { track -> [String: Any] in
|
||||
["trackId": track.id, "type": "TEXT", "subtype": "SUBTITLES",
|
||||
"trackContentId": track.url.absoluteString, "trackContentType": "text/vtt",
|
||||
"name": track.label, "language": track.language]
|
||||
}
|
||||
}
|
||||
var command: [String: Any] = ["type": "LOAD", "media": info, "autoplay": playing,
|
||||
"activeTrackIds": subtitle.map { [media.tracks[$0].id] } ?? []]
|
||||
if position > 0 || !media.live { command["currentTime"] = position }
|
||||
let requestID = allocateRequest()
|
||||
try await withTaskCancellationHandler {
|
||||
try Task.checkCancellation()
|
||||
try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, any Error>) in
|
||||
do {
|
||||
try request(command, namespace: CastWire.media, destination: transport,
|
||||
operation: "media load", generation: generation, timeout: loadTimeout,
|
||||
id: requestID) { [weak self] _, error in
|
||||
if let self, self.mediaGeneration == generation { self.loading = false; self.publishSeekState() }
|
||||
if let error { continuation.resume(throwing: error) }
|
||||
else { continuation.resume() }
|
||||
}
|
||||
} catch { loading = false; continuation.resume(throwing: error) }
|
||||
}
|
||||
} onCancel: {
|
||||
Task { @MainActor [weak self] in
|
||||
guard let self, self.mediaGeneration == generation else { return }
|
||||
// A cancelled LOAD may already be on the receiver. Tear down the channel
|
||||
// rather than letting its late acknowledgement acquire ownership.
|
||||
self.fail(.cancelled)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func setPlaying(_ playing: Bool) -> Bool {
|
||||
if replayTask != nil { replayDesiredPlaying = playing; return true }
|
||||
guard playing, latestStatus?.ended == true, let loadedMedia else {
|
||||
return control(["type": playing ? "PLAY" : "PAUSE"])
|
||||
}
|
||||
let subtitle = latestStatus?.subtitle
|
||||
let title = loadedTitle
|
||||
replayDesiredPlaying = true
|
||||
let connectionID = self.connectionID
|
||||
replayTask = Task { [weak self] in
|
||||
guard let self else { return }
|
||||
do {
|
||||
try await self.load(loadedMedia, title: title, position: 0, playing: true, subtitle: subtitle)
|
||||
guard self.connectionID == connectionID else { return }
|
||||
self.replayTask = nil
|
||||
self.publishSeekState()
|
||||
if !self.replayDesiredPlaying { _ = self.setPlaying(false) }
|
||||
} catch {
|
||||
guard self.connectionID == connectionID else { return }
|
||||
self.replayTask = nil
|
||||
self.publishSeekState()
|
||||
if self.connection != nil { self.fail(error as? CastError ?? .disconnected) }
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
func seek(_ seconds: Double) -> Bool {
|
||||
guard seconds.isFinite, seconds >= 0, latestStatus?.seekable == true else { return false }
|
||||
return control(["type": "SEEK", "currentTime": seconds], seeking: true)
|
||||
}
|
||||
func setVolume(_ volume: Double) -> Bool {
|
||||
guard volume.isFinite, (0...1).contains(volume), appTransport != nil else { return false }
|
||||
return control(["type": "SET_VOLUME", "volume": ["level": volume, "muted": volume == 0]], receiver: true)
|
||||
}
|
||||
func setSubtitle(_ index: Int?) -> Bool {
|
||||
guard index == nil || tracks.indices.contains(index!) else { return false }
|
||||
return control(["type": "EDIT_TRACKS_INFO", "activeTrackIds": index.map { [tracks[$0].id] } ?? []])
|
||||
}
|
||||
|
||||
private func control(_ command: [String: Any], seeking: Bool = false, receiver: Bool = false) -> Bool {
|
||||
guard !loading, let transport = appTransport, let mediaSession else { return false }
|
||||
var command = command
|
||||
if !receiver { command["mediaSessionId"] = mediaSession }
|
||||
let id = allocateRequest()
|
||||
let type = command["type"] as? String
|
||||
let changingPlayback = type == "PLAY" || type == "PAUSE"
|
||||
do {
|
||||
try request(command, namespace: receiver ? CastWire.receiver : CastWire.media,
|
||||
destination: receiver ? "receiver-0" : transport,
|
||||
operation: (command["type"] as? String ?? "command").lowercased(),
|
||||
generation: mediaGeneration, id: id) { [weak self] _, error in
|
||||
guard let self else { return }
|
||||
if case .cancelled? = error { return }
|
||||
if self.pendingSeek == id { self.pendingSeek = nil; self.publishSeekState() }
|
||||
if self.pendingPlayback == id { self.pendingPlayback = nil; self.publishSeekState() }
|
||||
if let error { self.onError?(error.localizedDescription) }
|
||||
}
|
||||
if seeking { pendingSeek = id; publishSeekState() }
|
||||
if changingPlayback { pendingPlayback = id; publishSeekState() }
|
||||
return true
|
||||
} catch { onError?(error.localizedDescription); return false }
|
||||
}
|
||||
|
||||
private func allocateRequest() -> Int { nextRequest += 1; return nextRequest }
|
||||
|
||||
@discardableResult
|
||||
private func request(_ command: [String: Any], namespace: String, destination: String,
|
||||
operation: String, generation: UUID? = nil, timeout: Duration? = nil,
|
||||
id suppliedID: Int? = nil,
|
||||
completion: @escaping ([String: Any]?, CastError?) -> Void) throws -> Int {
|
||||
guard pending.count < 32 else { throw CastError.queueFull }
|
||||
let id = suppliedID ?? allocateRequest()
|
||||
var command = command
|
||||
command["requestId"] = id
|
||||
let sleep = self.sleep
|
||||
let duration = timeout ?? requestTimeout
|
||||
let deadline = Task { [weak self] in
|
||||
do { try await sleep(duration) } catch { return }
|
||||
self?.complete(id, json: nil, error: .timeout(operation))
|
||||
}
|
||||
pending[id] = Pending(namespace: namespace, destination: destination, operation: operation,
|
||||
generation: generation, completion: completion, deadline: deadline)
|
||||
do { try send(command, namespace: namespace, destination: destination) }
|
||||
catch { pending.removeValue(forKey: id)?.deadline.cancel(); throw error }
|
||||
return id
|
||||
}
|
||||
|
||||
private func complete(_ id: Int, json: [String: Any]?, error: CastError?) {
|
||||
guard let item = pending.removeValue(forKey: id) else { return }
|
||||
item.deadline.cancel()
|
||||
item.completion(json, error)
|
||||
}
|
||||
|
||||
private func cancelMediaRequests() {
|
||||
let ids = pending.filter { $0.value.generation != nil }.map(\.key)
|
||||
for id in ids { complete(id, json: nil, error: .cancelled) }
|
||||
}
|
||||
|
||||
private func send(_ json: [String: Any], namespace: String, destination: String) throws {
|
||||
guard connection != nil else { throw CastError.disconnected }
|
||||
guard writes.count < 32 else { throw CastError.queueFull }
|
||||
let payload = try JSONSerialization.data(withJSONObject: json)
|
||||
writes.append(try CastWire.frame(.init(source: source, destination: destination,
|
||||
namespace: namespace, payload: payload)))
|
||||
flush()
|
||||
}
|
||||
|
||||
private func flush() {
|
||||
guard !writing, let socket = connection, !writes.isEmpty else { return }
|
||||
writing = true
|
||||
let data = writes.removeFirst()
|
||||
let id = connectionID
|
||||
socket.send(content: data, completion: .contentProcessed { [weak self] error in
|
||||
Task { @MainActor in
|
||||
guard let self, self.connectionID == id else { return }
|
||||
self.writing = false
|
||||
if error != nil { self.fail(.disconnected) } else { self.flush() }
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
private func receive(_ socket: NWConnection, id: UUID) {
|
||||
socket.receive(minimumIncompleteLength: 1, maximumLength: 16 * 1024) { [weak self] data, _, ended, error in
|
||||
Task { @MainActor in
|
||||
guard let self, self.connectionID == id else { return }
|
||||
do {
|
||||
if let data {
|
||||
for message in try self.stream.append(data) {
|
||||
self.handle(message)
|
||||
guard self.connectionID == id else { return }
|
||||
}
|
||||
}
|
||||
if ended || error != nil { self.fail(.disconnected) }
|
||||
else { self.receive(socket, id: id) }
|
||||
} catch { self.fail(.protocolError) }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func handle(_ message: CastWire.Message) {
|
||||
guard message.destination == source || message.destination == "*",
|
||||
let json = try? JSONSerialization.jsonObject(with: message.payload) as? [String: Any],
|
||||
let type = json["type"] as? String else { return }
|
||||
if message.namespace == CastWire.heartbeat, message.source == "receiver-0" {
|
||||
if type == "PONG" { awaitingPong = false }
|
||||
if type == "PING" {
|
||||
do { try send(["type": "PONG"], namespace: CastWire.heartbeat, destination: "receiver-0") }
|
||||
catch { fail(.disconnected) }
|
||||
}
|
||||
return
|
||||
}
|
||||
if message.namespace == CastWire.connection, type == "CLOSE",
|
||||
message.source == appTransport || message.source == "receiver-0" {
|
||||
fail(.disconnected); return
|
||||
}
|
||||
let id = json["requestId"] as? Int
|
||||
let item = id.flatMap { pending[$0] }
|
||||
let correlated = item?.namespace == message.namespace && item?.destination == message.source &&
|
||||
(item?.generation == nil || item?.generation == mediaGeneration)
|
||||
if ["INVALID_REQUEST", "INVALID_PLAYER_STATE", "LOAD_FAILED", "LOAD_CANCELLED", "LAUNCH_ERROR"].contains(type) {
|
||||
if correlated, let id, let item { complete(id, json: nil, error: .rejected(item.operation)) }
|
||||
return
|
||||
}
|
||||
if message.namespace == CastWire.receiver, message.source == "receiver-0", type == "RECEIVER_STATUS" {
|
||||
if let status = json["status"] as? [String: Any] {
|
||||
if let volume = status["volume"] as? [String: Any],
|
||||
let level = volume["level"] as? Double, level.isFinite {
|
||||
let muted = volume["muted"] as? Bool ?? false
|
||||
receiverVolume = muted ? 0 : min(1, max(0, level))
|
||||
if let latestStatus { publish(latestStatus, volume: receiverVolume) }
|
||||
}
|
||||
if let appSession, let appTransport {
|
||||
let apps = status["applications"] as? [[String: Any]] ?? []
|
||||
guard apps.contains(where: { $0["sessionId"] as? String == appSession &&
|
||||
$0["transportId"] as? String == appTransport }) else {
|
||||
fail(.rejected("receiver ownership changed")); return
|
||||
}
|
||||
}
|
||||
}
|
||||
if correlated, let id { complete(id, json: json, error: nil) }
|
||||
return
|
||||
}
|
||||
guard message.namespace == CastWire.media, message.source == appTransport,
|
||||
type == "MEDIA_STATUS", let statuses = json["status"] as? [[String: Any]] else { return }
|
||||
let olderReply = correlated && id.map { $0 < newestMediaReply } == true
|
||||
if olderReply, id != pendingSeek, id != pendingPlayback, let id {
|
||||
complete(id, json: json, error: nil)
|
||||
return
|
||||
}
|
||||
for status in statuses {
|
||||
guard let session = status["mediaSessionId"] as? Int else { continue }
|
||||
let media = status["media"] as? [String: Any]
|
||||
let reportedContent = media?["contentId"] as? String
|
||||
let customData = media?["customData"] as? [String: Any]
|
||||
if let nonce = customData?["terebiSession"] as? String, nonce != mediaGeneration?.uuidString {
|
||||
if retiredLoads.contains(nonce) { continue }
|
||||
if !loading { fail(.rejected("media ownership changed")); return }
|
||||
continue
|
||||
}
|
||||
if mediaSession == nil {
|
||||
guard loading, correlated, item?.operation == "media load", reportedContent == contentID,
|
||||
!retiredSessions.contains(session) else { continue }
|
||||
mediaSession = session
|
||||
}
|
||||
guard session == mediaSession else {
|
||||
if !retiredSessions.contains(session), !loading {
|
||||
fail(.rejected("media ownership changed")); return
|
||||
}
|
||||
continue
|
||||
}
|
||||
if let reportedContent, reportedContent != contentID {
|
||||
fail(.rejected("media ownership changed")); return
|
||||
}
|
||||
let state = status["playerState"] as? String ?? "IDLE"
|
||||
guard ["IDLE", "PLAYING", "PAUSED", "BUFFERING"].contains(state) else {
|
||||
fail(.protocolError); return
|
||||
}
|
||||
let idle = status["idleReason"] as? String
|
||||
if state == "IDLE", idle == "ERROR" {
|
||||
if correlated, let id { complete(id, json: nil, error: .rejected("media playback")) }
|
||||
fail(.rejected("media playback")); return
|
||||
}
|
||||
if state == "IDLE", idle == "INTERRUPTED" || idle == "CANCELLED" {
|
||||
fail(.rejected("media playback stopped")); return
|
||||
}
|
||||
let receivedPosition = status["currentTime"] as? Double ?? latestStatus?.position ?? 0
|
||||
let position = olderReply && id != pendingSeek ? latestStatus?.position ?? receivedPosition : receivedPosition
|
||||
let duration = media?["duration"] as? Double ?? latestStatus?.duration ?? loadedMedia?.duration ?? 0
|
||||
guard position.isFinite, duration.isFinite else { fail(.protocolError); return }
|
||||
let supported = status["supportedMediaCommands"] as? Int
|
||||
let active = status["activeTrackIds"] as? [Int]
|
||||
let subtitle = active.map { ids in tracks.firstIndex { ids.contains($0.id) } } ?? latestStatus?.subtitle
|
||||
let live = media?["streamType"] as? String == "LIVE" || loadedMedia?.live == true
|
||||
// A late SEEK still owns its position acknowledgement, but not newer playback intent.
|
||||
let previous = olderReply && id != pendingPlayback ? latestStatus : nil
|
||||
// An old status/poll cannot clear a newer seek. Only that seek's reply does.
|
||||
if correlated, id == pendingSeek { pendingSeek = nil }
|
||||
if correlated, id == pendingPlayback { pendingPlayback = nil }
|
||||
let next = CastStatus(position: max(0, position), duration: max(0, duration),
|
||||
playing: previous?.playing ?? (state == "PLAYING" || state == "BUFFERING"),
|
||||
buffering: previous?.buffering ?? (state == "BUFFERING"), ended: previous?.ended ?? (state == "IDLE" && idle == "FINISHED"),
|
||||
volume: receiverVolume,
|
||||
seekable: !live && (supported.map { $0 & 2 != 0 } ?? latestStatus?.seekable ?? false),
|
||||
subtitle: subtitle, seekPending: pendingSeek != nil, playbackPending: playbackPending)
|
||||
latestStatus = next
|
||||
if correlated, let id { newestMediaReply = max(newestMediaReply, id) }
|
||||
onStatus?(next)
|
||||
if correlated, let id { complete(id, json: json, error: nil) }
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
private func publish(_ status: CastStatus, volume: Double? = nil) {
|
||||
let next = CastStatus(position: status.position, duration: status.duration, playing: status.playing,
|
||||
buffering: status.buffering, ended: status.ended, volume: volume ?? status.volume,
|
||||
seekable: status.seekable, subtitle: status.subtitle, seekPending: pendingSeek != nil,
|
||||
playbackPending: playbackPending)
|
||||
latestStatus = next
|
||||
onStatus?(next)
|
||||
}
|
||||
private func publishSeekState() { if let latestStatus { publish(latestStatus) } }
|
||||
|
||||
private func startHeartbeat() {
|
||||
heartbeatTask?.cancel()
|
||||
let id = connectionID
|
||||
heartbeatTask = Task { [weak self] in
|
||||
while !Task.isCancelled {
|
||||
guard let self, self.connectionID == id else { return }
|
||||
do { try await self.sleep(self.heartbeatInterval) } catch { return }
|
||||
guard self.connectionID == id else { return }
|
||||
if self.awaitingPong { self.fail(.timeout("heartbeat")); return }
|
||||
self.awaitingPong = true
|
||||
do {
|
||||
try self.send(["type": "PING"], namespace: CastWire.heartbeat, destination: "receiver-0")
|
||||
// Receivers do not reliably broadcast clock progress. Poll without
|
||||
// extrapolating receiver time, allowing only one outstanding poll.
|
||||
if !self.loading, let transport = self.appTransport, let session = self.mediaSession,
|
||||
self.pending[self.lastStatusPoll] == nil {
|
||||
self.lastStatusPoll = try self.request(["type": "GET_STATUS", "mediaSessionId": session],
|
||||
namespace: CastWire.media, destination: transport, operation: "media status",
|
||||
generation: self.mediaGeneration) { [weak self] _, error in
|
||||
if case .cancelled? = error { return }
|
||||
if let error { self?.onError?(error.localizedDescription) }
|
||||
}
|
||||
}
|
||||
} catch { self.fail(error as? CastError ?? .disconnected); return }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func disconnect(stopReceiver: Bool = true) { teardown(error: nil, stopReceiver: stopReceiver) }
|
||||
private func fail(_ error: CastError) { teardown(error: error, stopReceiver: false) }
|
||||
|
||||
private func teardown(error: CastError?, stopReceiver: Bool) {
|
||||
let socket = connection
|
||||
let connected = socket != nil
|
||||
var farewell = Data()
|
||||
if connected {
|
||||
func append(_ json: [String: Any], namespace: String, destination: String) {
|
||||
if let payload = try? JSONSerialization.data(withJSONObject: json),
|
||||
let frame = try? CastWire.frame(.init(source: source, destination: destination,
|
||||
namespace: namespace, payload: payload)) { farewell.append(frame) }
|
||||
}
|
||||
// Stop only the media session we acquired, never STOP the application: another
|
||||
// sender may be using the Default Media Receiver by the time we disconnect.
|
||||
if stopReceiver, let mediaSession, let appTransport, !loading {
|
||||
append(["type": "STOP", "mediaSessionId": mediaSession, "requestId": allocateRequest()],
|
||||
namespace: CastWire.media, destination: appTransport)
|
||||
}
|
||||
if let appTransport { append(["type": "CLOSE"], namespace: CastWire.connection, destination: appTransport) }
|
||||
append(["type": "CLOSE"], namespace: CastWire.connection, destination: "receiver-0")
|
||||
}
|
||||
connectionID = UUID()
|
||||
connection = nil
|
||||
socket?.stateUpdateHandler = nil
|
||||
readyDeadline?.cancel(); readyDeadline = nil
|
||||
heartbeatTask?.cancel(); heartbeatTask = nil
|
||||
replayTask?.cancel(); replayTask = nil
|
||||
awaitingPong = false
|
||||
let ready = self.ready
|
||||
self.ready = nil
|
||||
let requests = pending.values
|
||||
pending.removeAll()
|
||||
writes.removeAll(); writing = false
|
||||
appTransport = nil; appSession = nil; mediaSession = nil
|
||||
retiredSessions.removeAll(); contentID = nil; mediaGeneration = nil
|
||||
retiredLoads.removeAll()
|
||||
tracks.removeAll(); latestStatus = nil; pendingSeek = nil; loading = false
|
||||
pendingPlayback = nil
|
||||
loadedMedia = nil; loadedTitle = ""
|
||||
receiverVolume = 1
|
||||
ready?.resume(throwing: error ?? CastError.cancelled)
|
||||
for request in requests {
|
||||
request.deadline.cancel()
|
||||
request.completion(nil, error ?? .cancelled)
|
||||
}
|
||||
if let socket {
|
||||
if !farewell.isEmpty, error == nil {
|
||||
socket.send(content: farewell, completion: .contentProcessed { _ in socket.cancel() })
|
||||
DispatchQueue.global().asyncAfter(deadline: .now() + 0.5) { socket.cancel() }
|
||||
} else { socket.cancel() }
|
||||
}
|
||||
if connected { onDisconnect?(error?.localizedDescription) }
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user