861 lines
42 KiB
Swift
861 lines
42 KiB
Swift
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)?
|
|
// Recoverable command errors; terminal channel failures belong to onDisconnect.
|
|
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 struct Write {
|
|
let data: Data
|
|
let seek: Int?
|
|
}
|
|
|
|
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: [Write] = []
|
|
private var writing = false
|
|
private var launchRequest: Int?
|
|
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
|
|
private var pendingVolume: (request: Int, level: Double)?
|
|
private var latestVolumeRequest = 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 {
|
|
// TLS establishment and a cold receiver application have separate deadlines.
|
|
readyDeadline?.cancel()
|
|
readyDeadline = nil
|
|
try send(["type": "CONNECT", "origin": [:], "userAgent": "Terebi"],
|
|
namespace: CastWire.connection, destination: "receiver-0")
|
|
startHeartbeat()
|
|
let id = allocateRequest()
|
|
launchRequest = id
|
|
try request(["type": "LAUNCH", "appId": "CC1AD845"], namespace: CastWire.receiver,
|
|
destination: "receiver-0", operation: "receiver launch", timeout: .seconds(30), id: id) { [weak self] _, error in
|
|
if let error { self?.fail(error) }
|
|
}
|
|
} 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["textTrackStyle"] = [
|
|
"foregroundColor": "#FFFFFFFF",
|
|
"backgroundColor": "#00000000",
|
|
"edgeType": "OUTLINE",
|
|
"edgeColor": "#000000FF",
|
|
"fontGenericFamily": "SANS_SERIF",
|
|
"fontStyle": "NORMAL",
|
|
"fontScale": 1.0, // Receiver default; tune against physical TV output.
|
|
"windowType": "NONE"
|
|
]
|
|
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, self.connection != nil else { return }
|
|
if self.pendingVolume?.request == id { self.pendingVolume = nil; self.publishSeekState() }
|
|
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() }
|
|
if type == "SET_VOLUME", let level = (command["volume"] as? [String: Any])?["level"] as? Double {
|
|
latestVolumeRequest = id
|
|
pendingVolume = (id, level)
|
|
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 {
|
|
// Only the tail is replaceable: it has not reached the socket, and every
|
|
// intervening control or protocol frame remains an ordering barrier.
|
|
let superseded: Int?
|
|
if command["type"] as? String == "SEEK", namespace == CastWire.media,
|
|
let id = writes.last?.seek, let previous = pending[id],
|
|
previous.generation == generation, previous.destination == destination {
|
|
superseded = id
|
|
} else {
|
|
superseded = nil
|
|
}
|
|
guard pending.count < 32 || superseded != nil 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, replacing: superseded)
|
|
}
|
|
catch { pending.removeValue(forKey: id)?.deadline.cancel(); throw error }
|
|
if let superseded { complete(superseded, json: nil, error: .cancelled) }
|
|
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,
|
|
replacing: Int? = nil) throws {
|
|
guard connection != nil else { throw CastError.disconnected }
|
|
guard writes.count < 32 || replacing != nil else { throw CastError.queueFull }
|
|
let payload = try JSONSerialization.data(withJSONObject: json)
|
|
let write = Write(data: try CastWire.frame(.init(source: source, destination: destination,
|
|
namespace: namespace, payload: payload)),
|
|
seek: namespace == CastWire.media && json["type"] as? String == "SEEK" ?
|
|
json["requestId"] as? Int : nil)
|
|
if replacing != nil { writes[writes.count - 1] = write }
|
|
else { writes.append(write) }
|
|
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.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] {
|
|
let olderVolumeReply = id.map { $0 > 0 && $0 < latestVolumeRequest } ?? false
|
|
let acknowledgesVolume = correlated && id == pendingVolume?.request
|
|
if !olderVolumeReply, pendingVolume == nil || acknowledgesVolume,
|
|
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 acknowledgesVolume { pendingVolume = nil }
|
|
if let latestStatus { publish(latestStatus) }
|
|
}
|
|
// Receiver readiness can be broadcast before/without the LAUNCH reply.
|
|
// Only the requested Default Media Receiver on receiver-0 can satisfy it.
|
|
if let launchRequest, 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 {
|
|
appTransport = transport
|
|
appSession = session
|
|
do {
|
|
try send(["type": "CONNECT", "origin": [:]], namespace: CastWire.connection, destination: transport)
|
|
} catch { fail(error as? CastError ?? .protocolError); return }
|
|
self.launchRequest = nil
|
|
complete(launchRequest, json: json, error: nil)
|
|
let ready = self.ready
|
|
self.ready = nil
|
|
ready?.resume()
|
|
}
|
|
if let appSession, let appTransport {
|
|
let apps = status["applications"] as? [[String: Any]] ?? []
|
|
guard apps.contains(where: { $0["appId"] as? String == "CC1AD845" &&
|
|
$0["sessionId"] as? String == appSession &&
|
|
$0["transportId"] as? String == appTransport }) else {
|
|
fail(.rejected("receiver ownership changed")); return
|
|
}
|
|
}
|
|
}
|
|
if correlated, let id, id != launchRequest, id != pendingVolume?.request {
|
|
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: pendingVolume?.level ?? 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) {
|
|
let next = CastStatus(position: status.position, duration: status.duration, playing: status.playing,
|
|
buffering: status.buffering, ended: status.ended, volume: pendingVolume?.level ?? receiverVolume,
|
|
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 self, self.connection != nil, 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; launchRequest = nil
|
|
retiredSessions.removeAll(); contentID = nil; mediaGeneration = nil
|
|
retiredLoads.removeAll()
|
|
tracks.removeAll(); latestStatus = nil; pendingSeek = nil; loading = false
|
|
pendingPlayback = nil
|
|
loadedMedia = nil; loadedTitle = ""
|
|
receiverVolume = 1
|
|
pendingVolume = nil; latestVolumeRequest = 0
|
|
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) }
|
|
}
|
|
}
|