diff --git a/Sources/LumeEngine/LumePlayer.swift b/Sources/LumeEngine/LumePlayer.swift index 4ed25d1..e4f9cb7 100644 --- a/Sources/LumeEngine/LumePlayer.swift +++ b/Sources/LumeEngine/LumePlayer.swift @@ -49,6 +49,17 @@ public final class LumePlayer { private var tickTask: Task? private let configuration: PlayerConfiguration + /// Standby session pre-opened by `prepare(next:)`, held through + /// first-frame-decoded until `load(url:)` consumes it. + private var preparedNext: (url: String, session: PlayerSession, info: MediaInfo)? + /// Supersession guards: a `load`/`prepare` that lost a race against a + /// newer call (or `stop`) discards its session instead of installing it. + private var loadGeneration: UInt64 = 0 + private var prepareGeneration: UInt64 = 0 + /// Bound on the first-frame wait during a seamless swap; past it the swap + /// proceeds anyway (losing only the no-gap guarantee). + private static let firstFrameTimeout: Double = 10 + public init(configuration: PlayerConfiguration = PlayerConfiguration()) { self.configuration = configuration } @@ -57,37 +68,209 @@ public final class LumePlayer { /// Opens `url`, replacing any previous session (each open is a fresh /// engine session — PLAN.md §3.1). + /// + /// Zero-delay switching (PLAN.md §6), governed by + /// `PlayerConfiguration.switchPolicy`: when a session matching `url` was + /// staged by `prepare(next:)`, the swap is immediate under any policy. + /// Otherwise `.overlapped` keeps the current session rendering while the + /// replacement opens through its first decoded frame (two connections + /// briefly), and `.sequential` closes the current session *first* — + /// freeing its source connection — while its last decoded frame stays + /// frozen on screen until the replacement is ready. A failed overlapped + /// open (e.g. a provider capped at one concurrent connection) falls back + /// to the sequential path before the error surfaces. public func load(url: String) async throws -> MediaInfo { + loadGeneration &+= 1 + let generation = loadGeneration + + // A prepared session for this exact URL swaps in with no open cost + // (consuming it opens nothing new), so it is honored under any policy. + if let prepared = preparedNext, prepared.url == url { + preparedNext = nil + if await prepared.session.state != .failed { + guard generation == loadGeneration else { + Task { await prepared.session.shutdown() } + throw EngineError(code: .invalidState, message: "load superseded by a newer load") + } + adopt(session: prepared.session, info: prepared.info) + return prepared.info + } + // The standby died while waiting (network drop) — fall through. + let dead = prepared.session + Task { await dead.shutdown() } + } + discardPreparedSession() + + let canSwitch = session != nil && state != .failed + if canSwitch, configuration.switchPolicy == .overlapped { + let next = PlayerSession(configuration: configuration) + do { + let info = try await next.open(url: url) + await next.waitForFirstFrame(timeout: Self.firstFrameTimeout) + guard generation == loadGeneration else { + Task { await next.shutdown() } + throw EngineError(code: .invalidState, message: "load superseded by a newer load") + } + adopt(session: next, info: info) + return info + } catch { + Task { await next.shutdown() } + guard generation == loadGeneration else { throw error } + // Fall through to the sequential retry: a provider capped at + // one concurrent connection refuses the overlapped open but + // accepts the same URL once the current stream is closed. + } + } + if canSwitch, configuration.switchPolicy != .none { + return try await sequentialLoad(url: url, generation: generation) + } + await teardownSession() + guard generation == loadGeneration else { + throw EngineError(code: .invalidState, message: "load superseded by a newer load") + } let session = PlayerSession(configuration: configuration) self.session = session state = .opening + startObservation(of: session) - eventTask = Task { [events = session.events] in - for await event in events { - self.handle(event: event) + do { + let info = try await session.open(url: url) + guard generation == loadGeneration else { + throw EngineError(code: .invalidState, message: "load superseded by a newer load") } + mediaInfo = info + duration = info.duration.map(MediaTime.seconds) + return info + } catch { + // A superseded load must not clobber the newer load's state. + guard generation == loadGeneration else { throw error } + lastError = error as? EngineError + state = .failed + throw error } - tickTask = Task { - while !Task.isCancelled { - try? await Task.sleep(for: .milliseconds(100)) - await self.tick() - } + } + + /// Single-connection switch: the current session shuts down first (its + /// display layer keeps the last decoded frame — `SystemRenderer.shutdown` + /// flushes without removing the displayed image, so the screen never + /// blanks), then the replacement opens and the layers swap at its first + /// frame. At most one source connection exists at any moment. Playback + /// pauses for the duration of the open — the price of the connection cap. + private func sequentialLoad(url: String, generation: UInt64) async throws -> MediaInfo { + eventTask?.cancel() + tickTask?.cancel() + if let old = session { + // Bounded join: the connection must actually be closed before the + // replacement opens, or a one-connection provider refuses it. + await old.shutdown() + } + state = .opening + guard generation == loadGeneration else { + throw EngineError(code: .invalidState, message: "load superseded by a newer load") } + let next = PlayerSession(configuration: configuration) do { - let info = try await session.open(url: url) - mediaInfo = info - duration = info.duration.map(MediaTime.seconds) + let info = try await next.open(url: url) + await next.waitForFirstFrame(timeout: Self.firstFrameTimeout) + guard generation == loadGeneration else { + Task { await next.shutdown() } + throw EngineError(code: .invalidState, message: "load superseded by a newer load") + } + adopt(session: next, info: info) return info } catch { + Task { await next.shutdown() } + guard generation == loadGeneration else { throw error } + // The dead session stays attached: its frozen frame keeps the + // surface alive behind whatever failure UI the app raises. lastError = error as? EngineError state = .failed throw error } } + /// Stages `url` in a standby session, opened through first-frame-decoded + /// but never played (PLAN.md §6). A following `load(url:)` for the same + /// URL swaps it in with zero delay — this powers next-episode + /// auto-advance and channel zapping. Current playback is untouched. + /// + /// Only one URL is staged at a time; a newer `prepare` replaces the + /// previous standby. The standby holds its **own source connection** and + /// read-ahead buffer until consumed or discarded — only call this when + /// the source allows a stream beside the playing one (the same budget as + /// `SwitchPolicy.overlapped`); on one-connection providers rely on the + /// sequential switch instead. + @discardableResult + public func prepare(next url: String) async throws -> MediaInfo { + discardPreparedSession() + prepareGeneration &+= 1 + let generation = prepareGeneration + + let session = PlayerSession(configuration: configuration) + do { + let info = try await session.open(url: url) + await session.waitForFirstFrame(timeout: Self.firstFrameTimeout) + guard generation == prepareGeneration else { + Task { await session.shutdown() } + throw EngineError(code: .invalidState, message: "prepare superseded by a newer prepare") + } + preparedNext = (url, session, info) + return info + } catch { + Task { await session.shutdown() } + throw error + } + } + + /// Discards the standby session staged by `prepare(next:)`, if any. + public func discardPreparedSession() { + prepareGeneration &+= 1 + guard let prepared = preparedNext else { return } + preparedNext = nil + Task { await prepared.session.shutdown() } + } + + /// Atomic swap of `prepare`d/seamlessly opened media: the new session + /// becomes the active one (its display layer replaces the old one via the + /// `displayLayer` observation) and the old session is torn down + /// asynchronously — the swap never waits for thread joins. + private func adopt(session next: PlayerSession, info: MediaInfo) { + eventTask?.cancel() + tickTask?.cancel() + if let old = session { + Task { await old.shutdown() } + } + session = next + state = .ready + mediaInfo = info + duration = info.duration.map(MediaTime.seconds) + lastError = nil + subtitleText = nil + startObservation(of: next) + if rate != 1.0 { + let rate = rate + Task { await next.setRate(rate) } + } + } + + /// Event pump + 10 Hz position tick for the active session. + private func startObservation(of session: PlayerSession) { + eventTask = Task { [events = session.events] in + for await event in events { + self.handle(event: event) + } + } + tickTask = Task { + while !Task.isCancelled { + try? await Task.sleep(for: .milliseconds(100)) + await self.tick() + } + } + } + public func play() { let session = session Task { await session?.play() } @@ -127,6 +310,8 @@ public final class LumePlayer { } public func stop() async { + loadGeneration &+= 1 + discardPreparedSession() await teardownSession() state = .idle position = 0 diff --git a/Sources/LumeEngineCore/Session/PlayerSession.swift b/Sources/LumeEngineCore/Session/PlayerSession.swift index e10f287..45abe14 100644 --- a/Sources/LumeEngineCore/Session/PlayerSession.swift +++ b/Sources/LumeEngineCore/Session/PlayerSession.swift @@ -46,6 +46,28 @@ public struct PlayerConfiguration: Sendable { /// Seconds without clock progress (while expected) before `.stalled` fires. public var stallThreshold: Double = 8 + /// How `LumePlayer.load(url:)` replaces an already-open source + /// (zero-delay switching, PLAN.md §6). + public enum SwitchPolicy: Sendable, Equatable { + /// Zero-delay: the replacement opens behind the still-playing session + /// and swaps in once it holds its first decoded frame. Briefly holds + /// **two** source connections — use only when the source allows a + /// second concurrent stream (a refused overlapped open falls back to + /// `.sequential` automatically). + case overlapped + /// Single-connection zero-black: the current session closes *first* + /// (releasing its connection), its last decoded frame stays frozen on + /// screen while the replacement opens, and the layers swap at first + /// frame. Playback pauses during the switch but the screen never + /// blanks. The safe choice for providers capped at one connection. + case sequential + /// Teardown-first with a blank surface (no switching aid). + case none + } + + /// Switch behavior for `LumePlayer.load(url:)`; see `SwitchPolicy`. + public var switchPolicy: SwitchPolicy = .overlapped + public init() {} } @@ -78,7 +100,7 @@ public actor PlayerSession { /// `activeCues(at: renderer.currentTime)` from any thread. public nonisolated let subtitles = SubtitleStore() - private(set) var state: State = .idle { + public private(set) var state: State = .idle { didSet { if state != oldValue { eventSink.yield(.stateChanged(state)) } } @@ -372,6 +394,35 @@ public actor PlayerSession { ) } + // MARK: Zero-delay switching support (PLAN.md §6) + + /// Suspends until the renderer holds the first decoded frame of the + /// presentation lane — video when a video track is active, else audio. + /// The renderers accept media while the clock is paused, so a freshly + /// opened session reaches this point without ever playing: this is the + /// "prepared through first-frame-decoded" gate for zero-delay switching + /// (a standby session whose display layer already shows a picture can be + /// swapped in with no black gap). + /// + /// Returns `false` if the session fails, is torn down, or the timeout + /// elapses first — callers may still swap then; they just lose the + /// no-gap guarantee. + @discardableResult + public func waitForFirstFrame(timeout: Double = 10) async -> Bool { + let deadline = Date(timeIntervalSinceNow: timeout) + while Date() < deadline { + if state == .failed || state == .idle && info != nil { return false } + let lowWater = renderer.enqueuedLowWaterMark + if videoDecoder != nil { + if MediaTime.isValid(lowWater.video) { return true } + } else if audioDecoder != nil { + if MediaTime.isValid(lowWater.audio) { return true } + } + try? await Task.sleep(for: .milliseconds(50)) + } + return false + } + // MARK: Seek /// Seeks to `position` seconds (media-relative). Flush order matters: diff --git a/Tests/LumeEngineCoreTests/SeamlessSwitchingTests.swift b/Tests/LumeEngineCoreTests/SeamlessSwitchingTests.swift new file mode 100644 index 0000000..93c9f7d --- /dev/null +++ b/Tests/LumeEngineCoreTests/SeamlessSwitchingTests.swift @@ -0,0 +1,195 @@ +import Foundation +import Testing +@testable import LumeEngine +@testable import LumeEngineCore + +/// Zero-delay stream switching (PLAN.md §6, issue #16): `prepare(next:)` +/// stages a standby session through first-frame-decoded and `load(url:)` +/// swaps it in atomically; without a prepared session, a seamless load keeps +/// the old session alive until the replacement holds its first frame. +@Suite("Zero-delay switching", .serialized) +struct SeamlessSwitchingTests { + private func makeConfiguration() -> PlayerConfiguration { + var configuration = PlayerConfiguration() + configuration.muted = true + configuration.bufferTarget = 0.3 + return configuration + } + + @Test("waitForFirstFrame resolves on an opened, never-played session", .timeLimit(.minutes(1))) + func firstFrameGate() async throws { + let session = PlayerSession(configuration: makeConfiguration()) + _ = try await session.open(url: try Fixtures.path("basic.mp4")) + + // Paused throughout: the renderers must accept the first frame with + // the clock stopped (that's what makes a standby swap gapless). + let ready = await session.waitForFirstFrame(timeout: 10) + #expect(ready, "first frame should decode without playback starting") + + let lowWater = session.renderer.enqueuedLowWaterMark + #expect(MediaTime.isValid(lowWater.video), "video renderer should hold the first frame") + #expect(session.renderer.rate == 0, "the standby clock must never run") + + await session.shutdown() + } + + @Test("waitForFirstFrame reports failure instead of hanging", .timeLimit(.minutes(1))) + func firstFrameGateFailure() async { + let session = PlayerSession(configuration: makeConfiguration()) + _ = try? await session.open(url: "/nonexistent/definitely-missing.mp4") + let ready = await session.waitForFirstFrame(timeout: 5) + #expect(!ready) + await session.shutdown() + } + + @Test("prepare(next:) then load swaps sessions atomically", .timeLimit(.minutes(1))) + @MainActor + func prepareThenLoad() async throws { + let player = LumePlayer(configuration: makeConfiguration()) + + _ = try await player.load(url: try Fixtures.path("basic.mp4")) + let firstLayer = player.displayLayer + player.play() + try await eventually(player, reaches: .playing) + + let nextURL = try Fixtures.path("multitrack.mkv") + let preparedInfo = try await player.prepare(next: nextURL) + #expect(preparedInfo.audioTracks.count == 2) + // Preparation must not disturb current playback. + #expect(player.state == .playing) + #expect(player.displayLayer === firstLayer) + + // The staged session is consumed: the swap needs no second open. + let start = Date() + let info = try await player.load(url: nextURL) + #expect(info.audioTracks.count == 2) + #expect(Date().timeIntervalSince(start) < 2, "a prepared load must not re-open the source") + #expect(player.displayLayer !== firstLayer, "the swap must install the standby session's layer") + + player.play() + try await eventually(player, reaches: .playing) + #expect(player.position < 9, "position must belong to the new session") + + await player.stop() + } + + @Test("seamless load keeps the old session until the new one has a frame", .timeLimit(.minutes(1))) + @MainActor + func seamlessLoadWithoutPrepare() async throws { + let player = LumePlayer(configuration: makeConfiguration()) + + _ = try await player.load(url: try Fixtures.path("basic.mp4")) + let firstLayer = player.displayLayer + player.play() + try await eventually(player, reaches: .playing) + + let info = try await player.load(url: try Fixtures.path("multitrack.mkv")) + #expect(info.audioTracks.count == 2) + #expect(player.displayLayer !== firstLayer) + + player.play() + try await eventually(player, reaches: .playing) + await player.stop() + } + + @Test("seamless load of a dead source falls back cold and fails loudly", .timeLimit(.minutes(1))) + @MainActor + func seamlessLoadFailure() async throws { + let player = LumePlayer(configuration: makeConfiguration()) + + _ = try await player.load(url: try Fixtures.path("basic.mp4")) + player.play() + try await eventually(player, reaches: .playing) + + await #expect(throws: EngineError.self) { + _ = try await player.load(url: "/nonexistent/definitely-missing.mp4") + } + // The failed load replaced the old session (load's contract), so the + // player ends in a reportable failed state — never a silent limbo. + // (The event pump delivers the session's state trail asynchronously, + // so allow it to settle.) + try await eventually(player, reaches: .failed) + + await player.stop() + } + + @Test("sequential policy switches on one connection at a time", .timeLimit(.minutes(1))) + @MainActor + func sequentialSwitch() async throws { + var configuration = makeConfiguration() + configuration.switchPolicy = .sequential + let player = LumePlayer(configuration: configuration) + + _ = try await player.load(url: try Fixtures.path("basic.mp4")) + let firstLayer = player.displayLayer + player.play() + try await eventually(player, reaches: .playing) + + // The old session closes before the new one opens (one connection); + // the surface swaps only once the replacement holds its first frame. + let info = try await player.load(url: try Fixtures.path("multitrack.mkv")) + #expect(info.audioTracks.count == 2) + #expect(player.displayLayer !== firstLayer) + + player.play() + try await eventually(player, reaches: .playing) + #expect(player.position < 9, "position must belong to the new session") + + await player.stop() + } + + @Test("sequential load of a dead source keeps the frozen surface and fails", .timeLimit(.minutes(1))) + @MainActor + func sequentialLoadFailure() async throws { + var configuration = makeConfiguration() + configuration.switchPolicy = .sequential + let player = LumePlayer(configuration: configuration) + + _ = try await player.load(url: try Fixtures.path("basic.mp4")) + let firstLayer = player.displayLayer + player.play() + try await eventually(player, reaches: .playing) + + await #expect(throws: EngineError.self) { + _ = try await player.load(url: "/nonexistent/definitely-missing.mp4") + } + try await eventually(player, reaches: .failed) + // The dead session stays attached so its last frame keeps the surface + // alive behind the app's failure UI. + #expect(player.displayLayer === firstLayer) + + await player.stop() + } + + @Test("switchPolicy .none restores the cold teardown-first path", .timeLimit(.minutes(1))) + @MainActor + func coldSwitchWhenDisabled() async throws { + var configuration = makeConfiguration() + configuration.switchPolicy = .none + let player = LumePlayer(configuration: configuration) + + _ = try await player.load(url: try Fixtures.path("basic.mp4")) + player.play() + try await eventually(player, reaches: .playing) + + let info = try await player.load(url: try Fixtures.path("multitrack.mkv")) + #expect(info.audioTracks.count == 2) + player.play() + try await eventually(player, reaches: .playing) + + await player.stop() + } + + @MainActor + private func eventually( + _ player: LumePlayer, + reaches state: LumePlayer.State, + within seconds: Double = 10 + ) async throws { + let deadline = Date(timeIntervalSinceNow: seconds) + while player.state != state && Date() < deadline { + try await Task.sleep(for: .milliseconds(100)) + } + #expect(player.state == state) + } +}