perf: coalesce superseded queued playback seeks
This commit is contained in:
@@ -255,6 +255,11 @@ final class CastSession {
|
||||
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
|
||||
@@ -270,7 +275,7 @@ final class CastSession {
|
||||
private var awaitingPong = false
|
||||
private var nextRequest = 0
|
||||
private var pending: [Int: Pending] = [:]
|
||||
private var writes: [Data] = []
|
||||
private var writes: [Write] = []
|
||||
private var writing = false
|
||||
private var launchRequest: Int?
|
||||
private var appTransport: String?
|
||||
@@ -525,7 +530,17 @@ final class CastSession {
|
||||
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 }
|
||||
// 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
|
||||
@@ -537,8 +552,11 @@ final class CastSession {
|
||||
}
|
||||
pending[id] = Pending(namespace: namespace, destination: destination, operation: operation,
|
||||
generation: generation, completion: completion, deadline: deadline)
|
||||
do { try send(command, namespace: namespace, destination: destination) }
|
||||
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
|
||||
}
|
||||
|
||||
@@ -553,12 +571,17 @@ final class CastSession {
|
||||
for id in ids { complete(id, json: nil, error: .cancelled) }
|
||||
}
|
||||
|
||||
private func send(_ json: [String: Any], namespace: String, destination: String) throws {
|
||||
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 else { throw CastError.queueFull }
|
||||
guard writes.count < 32 || replacing != nil 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)))
|
||||
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()
|
||||
}
|
||||
|
||||
@@ -567,7 +590,7 @@ final class CastSession {
|
||||
writing = true
|
||||
let data = writes.removeFirst()
|
||||
let id = connectionID
|
||||
socket.send(content: data, completion: .contentProcessed { [weak self] error in
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user