hl-valdemar/terebi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Releases Wiki Activity

fix: separate Cast launch readiness, preserve volume intent and reuse seek inputs

Browse Source
This commit is contained in:
Valdemar Høgsvig Lorenzen
2026-10-13 21:40:19 +02:00
parent 697158aa02
commit b9e63cd760
8 changed files with 365 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Sources/App/CastSession.swift
+49 -29
View File
@@ -271,6 +271,7 @@ final class CastSession {
private var pending: [Int: Pending] = [:]
private var writes: [Data] = []
private var writing = false
private var launchRequest: Int?
private var appTransport: String?
private var appSession: String?
private var mediaSession: Int?
@@ -291,6 +292,8 @@ final class CastSession {
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),
@@ -359,31 +362,17 @@ final class CastSession {
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") { [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()
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) }
}
@@ -511,6 +500,7 @@ final class CastSession {
operation: (command["type"] as? String ?? "command").lowercased(),
generation: mediaGeneration, id: id) { [weak self] _, error in
guard let self 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() }
@@ -518,6 +508,11 @@ final class CastSession {
}
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 }
}
@@ -624,21 +619,45 @@ final class CastSession {
}
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 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 let latestStatus { publish(latestStatus, volume: receiverVolume) }
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["sessionId"] as? String == appSession &&
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 { complete(id, json: json, error: nil) }
if correlated, let id, id != launchRequest, id != pendingVolume?.request {
complete(id, json: json, error: nil)
}
return
}
guard message.namespace == CastWire.media, message.source == appTransport,
@@ -700,7 +719,7 @@ final class CastSession {
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,
volume: pendingVolume?.level ?? receiverVolume,
seekable: !live && (supported.map { $0 & 2 != 0 } ?? latestStatus?.seekable ?? false),
subtitle: subtitle, seekPending: pendingSeek != nil, playbackPending: playbackPending)
latestStatus = next
@@ -711,9 +730,9 @@ final class CastSession {
}
}
private func publish(_ status: CastStatus, volume: Double? = nil) {
private func publish(_ status: CastStatus) {
let next = CastStatus(position: status.position, duration: status.duration, playing: status.playing,
buffering: status.buffering, ended: status.ended, volume: volume ?? status.volume,
buffering: status.buffering, ended: status.ended, volume: pendingVolume?.level ?? receiverVolume,
seekable: status.seekable, subtitle: status.subtitle, seekPending: pendingSeek != nil,
playbackPending: playbackPending)
latestStatus = next
@@ -783,13 +802,14 @@ final class CastSession {
let requests = pending.values
pending.removeAll()
writes.removeAll(); writing = false
appTransport = nil; appSession = nil; mediaSession = nil
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()
Sources/App/TerebiApp.swift
+21 -1
View File
@@ -1469,7 +1469,7 @@ private struct PlaybackSlider: NSViewRepresentable {
context.coordinator.parent = self
slider.minValue = range.lowerBound
slider.maxValue = range.upperBound
slider.doubleValue = value
slider.setModelValue(value)
slider.isEnabled = enabled
slider.editingChanged = onEditingChanged
if let cell = slider.cell as? Cell, cell.bufferedRange != bufferedRange {
@@ -1493,6 +1493,21 @@ private struct PlaybackSlider: NSViewRepresentable {
final class Slider: NSSlider {
var editingChanged: (Bool) -> Void = { _ in }
private(set) var tracking = false
func setModelValue(_ value: Double) { if !tracking { doubleValue = value } }
#if PLAYER_CHECK
static func feedbackCheck() -> Bool {
let slider = Slider()
slider.minValue = 0
slider.maxValue = 1
slider.doubleValue = 0.8
slider.tracking = true
slider.setModelValue(0.79)
let heldUserValue = slider.doubleValue == 0.8
slider.tracking = false
slider.setModelValue(0.79)
return heldUserValue && slider.doubleValue == 0.79
}
#endif
override func mouseDown(with event: NSEvent) {
guard isEnabled else { return }
tracking = true
@@ -1538,6 +1553,11 @@ private struct PlaybackSlider: NSViewRepresentable {
}
}
#if PLAYER_CHECK
@MainActor
func castVolumeSliderFeedbackCheck() -> Bool { PlaybackSlider.Slider.feedbackCheck() }
#endif
@MainActor
private struct VideoSurface: NSViewRepresentable {
let view: NSView
Sources/Core/cast_media.cpp
+55 -11
View File
@@ -146,14 +146,14 @@ struct Input {
}
};
struct Track {
std::unique_ptr<Input> input;
Input *input;
AVStream *stream = nullptr;
AVCodecContext *decoder = nullptr;
Packet pending;
Frame decoded;
bool eof = false, draining = false, remux = false;
double nextTime = 0;
Track(std::unique_ptr<Input> in, AVMediaType type, double start, bool live) : input(std::move(in)) {
Track(Input &in, AVMediaType type, double start, bool live) : input(&in) {
int index = av_find_best_stream(input->format, type, -1, -1, nullptr, 0);
if (index < 0) return;
stream = input->format->streams[index];
@@ -212,7 +212,7 @@ struct Track {
auto p = packet();
input->arm();
for (;;) {
require(!Input::interrupt(input.get()));
require(!Input::interrupt(input));
int result = av_read_frame(input->format, p.get());
if (result == AVERROR_EOF) { eof = true; return; }
check(result);
@@ -229,7 +229,7 @@ struct Track {
void nextFrame() {
decoded.reset(); auto f = frame();
for (;;) {
require(!Input::interrupt(input.get()));
require(!Input::interrupt(input));
int result = avcodec_receive_frame(decoder, f.get());
if (result == 0) {
decoded = std::move(f);
@@ -440,6 +440,7 @@ struct OutputTrack {
}
};
struct Pipeline {
std::unique_ptr<Input> videoInput, audioInput;
std::unique_ptr<Track> video, audio;
std::unique_ptr<OutputTrack> videoOutput, audioOutput;
bool live;
@@ -450,17 +451,32 @@ struct Pipeline {
duration = std::max(source.duration, primary->duration());
bool hasVideo = av_find_best_stream(primary->format, AVMEDIA_TYPE_VIDEO, -1, -1, nullptr, 0) >= 0;
bool hasAudio = av_find_best_stream(primary->format, AVMEDIA_TYPE_AUDIO, -1, -1, nullptr, 0) >= 0;
if (hasVideo) video = std::make_unique<Track>(std::move(primary), AVMEDIA_TYPE_VIDEO, start, source.live);
if (hasVideo) videoInput = std::move(primary);
if (!source.audio.empty()) {
audio = std::make_unique<Track>(std::make_unique<Input>(stop, source.audio, source.audioHeaders, false, source.restricted, source.live), AVMEDIA_TYPE_AUDIO, start, source.live);
require(audio->stream != nullptr);
audioInput = std::make_unique<Input>(stop, source.audio, source.audioHeaders, false, source.restricted, source.live);
} else if (hasAudio) {
if (!primary) primary = std::make_unique<Input>(stop, source.video, source.videoHeaders, source.local, source.restricted, source.live);
audio = std::make_unique<Track>(std::move(primary), AVMEDIA_TYPE_AUDIO, start, source.live);
audioInput = std::move(primary);
}
reset(start);
if (!source.audio.empty()) require(audio && audio->stream);
require(video || audio);
if (!source.live) require(std::isfinite(duration) && duration > 0 && duration <= 86400 * 14);
}
void reset(double start) {
videoOutput.reset(); audioOutput.reset();
video.reset(); audio.reset();
// Fresh codecs preserve independent segment/AAC priming; demuxers retain
// their already-probed metadata and seek indexes across random requests.
if (videoInput) {
videoInput->operationDeadline = std::numeric_limits<int64_t>::max();
video = std::make_unique<Track>(*videoInput, AVMEDIA_TYPE_VIDEO, start, live);
}
if (audioInput) {
audioInput->operationDeadline = std::numeric_limits<int64_t>::max();
audio = std::make_unique<Track>(*audioInput, AVMEDIA_TYPE_AUDIO, start, live);
}
}
Segment segment(int64_t id, double start, double end) {
Mux mux;
if (video) {
@@ -507,6 +523,8 @@ struct CastMedia {
Source source;
std::atomic<bool> cancelled{false};
std::mutex lock;
std::mutex vodLock;
std::unique_ptr<Pipeline> vodPipeline;
std::condition_variable wake;
int listener = -1;
std::vector<int> clients;
@@ -607,6 +625,32 @@ struct CastMedia {
if (!source.live) text += "#EXT-X-ENDLIST\n";
return {text.begin(), text.end()};
}
std::shared_ptr<const Segment> vodSegment(int64_t id) {
auto cached = [&]() -> std::shared_ptr<const Segment> {
std::lock_guard guard(lock);
auto found = std::find_if(segments.begin(), segments.end(), [&](const auto &segment) { return segment->id == id; });
return found == segments.end() ? nullptr : *found;
};
if (auto segment = cached()) return segment;
// ponytail: uncached VOD generation is serialized; use a bounded input
// pool only if parallel uncached requests justify duplicate probe state.
std::lock_guard generation(vodLock);
require(!cancelled);
if (auto segment = cached()) return segment;
try {
if (vodPipeline) vodPipeline->reset(id * segmentSeconds);
else vodPipeline = std::make_unique<Pipeline>(source, cancelled, id * segmentSeconds);
auto segment = std::make_shared<const Segment>(vodPipeline->segment(
id, id * segmentSeconds, std::min(duration, (id + 1) * segmentSeconds)));
{ std::lock_guard guard(lock); segments.push_back(segment); while (segments.size() > windowSegments) segments.pop_front(); }
return segment;
} catch (...) {
// A failed demux/decode is never cached or reused. A later receiver
// request can prepare fresh inputs; this request still receives 503.
vodPipeline.reset();
throw;
}
}
void sendAll(int socket, const uint8_t *bytes, size_t size) {
const auto deadline = Clock::now() + std::chrono::seconds(15);
while (size) {
@@ -682,8 +726,8 @@ struct CastMedia {
if (found == segments.end()) status = 404; else retained = std::shared_ptr<const std::vector<uint8_t>>(*found, &(*found)->bytes);
} else if (double(id) * segmentSeconds >= duration) status = 404;
else {
Pipeline pipeline(source, cancelled, id * segmentSeconds);
data = pipeline.segment(id, id * segmentSeconds, std::min(duration, (id + 1) * segmentSeconds)).bytes;
auto segment = vodSegment(id);
retained = std::shared_ptr<const std::vector<uint8_t>>(segment, &segment->bytes);
}
} else status = 404;
response(socket, status, type, retained ? *retained : data, head, range);
@@ -728,7 +772,7 @@ extern "C" bool cast_media_start(CastMedia *media, const char *host, char url[51
std::unique_lock guard(media->lock);
require(media->wake.wait_for(guard, std::chrono::seconds(50), [&] { return media->segments.size() >= 2 || media->liveFailed || media->cancelled; }));
require(media->segments.size() >= 2 && !media->liveFailed && !media->cancelled);
}
} else media->vodPipeline = std::move(pipeline);
for (auto &worker : media->servers) worker = std::thread([media] { media->server(); });
require(media->base.size() + 11 < 512); std::snprintf(url, 512, "%sindex.m3u8", media->base.c_str()); *duration = media->source.live ? 0 : media->duration;
return true;