open-standards/wai/wai-sink-swift/Sources/WaiSink/PlayerSession.swift

// The native sink integrated with the platform's player (native-sink §4.4, §5, §6): the player
// is given the relay's URL for the presentation, so every playlist and segment it requests passes
// through the sink; the session observes it every 200 ms, logs its first frame, its stalls and
// its end, ends a session nobody can see, and posts the bundle to the owner's collector.
import AVFoundation
import Foundation

@MainActor
public final class WaiPlayerSession {
    public let sink: WaiSink
    public let relay: WaiRelay
    public let player: AVPlayer
    public let upstream: URL
    private let attention: WaiAttention
    private let collector: WaiCollector?
    private var timer: Timer?
    private var observers: [NSObjectProtocol] = []
    private var statusObservation: NSKeyValueObservation?
    private var presented = false
    private var stalled = false
    private var seekedSinceLast = false
    private var cmcdEvery: TimeInterval
    private var lastCMCD = Date()
    private let startUs: UInt64
    private(set) public var bundle: Data?
    private(set) public var ended: WaiSink.Ended?
    /// Called once, when the session ends, with its bundle and how it ended.
    public var onEnd: ((Data, WaiSink.Ended) -> Void)?
    /// Called with the player's item before the player is given it: where a host attaches its
    /// outputs (a host with no layer on screen attaches a video output, so the player decodes).
    public var prepareItem: ((AVPlayerItem) -> Void)?

    /// A session for the presentation at `url`, under `config`; `collector` the owner's, or nil to
    /// keep the bundle for the host.
    public init(url: URL, config: WaiSink.Config, attention: WaiAttention, collector: WaiCollector? = nil,
                player: AVPlayer = AVPlayer(), seed: Data? = nil, cmcdEvery: TimeInterval = 10) throws {
        startUs = WaiSink.nowUs
        sink = try WaiSink(config: config, seed: seed, startUs: startUs)
        relay = WaiRelay(sink: sink)
        try relay.start()
        upstream = url
        self.attention = attention
        self.collector = collector
        self.player = player
        self.cmcdEvery = cmcdEvery
    }

    /// Give the player the presentation through the relay and start playing.
    public func start() {
        let item = AVPlayerItem(url: relay.relayURL(for: upstream))
        let nc = NotificationCenter.default
        observers.append(nc.addObserver(forName: AVPlayerItem.didPlayToEndTimeNotification, object: item, queue: .main) { [weak self] _ in
            MainActor.assumeIsolated { self?.finish(.ended) }
        })
        observers.append(nc.addObserver(forName: AVPlayerItem.failedToPlayToEndTimeNotification, object: item, queue: .main) { [weak self] _ in
            MainActor.assumeIsolated { self?.finish(.error) }
        })
        prepareItem?(item)
        statusObservation = item.observe(\.status, options: [.new]) { [weak self] it, _ in
            if it.status == .failed {
                DispatchQueue.main.async { MainActor.assumeIsolated { self?.finish(.error) } }
            }
        }
        player.replaceCurrentItem(with: item)
        let t = Timer(timeInterval: 0.2, repeats: true) { [weak self] _ in MainActor.assumeIsolated { self?.tick() } }
        RunLoop.main.add(t, forMode: .common)
        timer = t
        player.play()
    }

    /// Seek, noting it for the next observation (§5.1).
    public func seek(to time: CMTime) {
        seekedSinceLast = true
        player.seek(to: time)
    }

    /// The viewer stopped it.
    public func stop() {
        finish(.user)
    }

    private func tick() {
        guard ended == nil else { return }
        let status = player.timeControlStatus
        let playing = status == .playing && player.rate == 1
        if playing && !presented {
            presented = true
            try? sink.event("present", atUs: WaiSink.nowUs)
        } else if presented && status == .waitingToPlayAtSpecifiedRate && !stalled {
            stalled = true
            try? sink.event("stall-start", atUs: WaiSink.nowUs)
        } else if stalled && status == .playing {
            stalled = false
            try? sink.event("stall-end", atUs: WaiSink.nowUs)
        }
        var obs: [String: Any] = [
            "t_us": WaiSink.nowUs,
            "playing": playing,
            "visible_bp": attention.visibleBasisPoints(),
            "audible": !player.isMuted && player.volume > 0,
            "seeked": seekedSinceLast,
            "interaction": attention.takeInteraction(),
        ]
        seekedSinceLast = false
        if let d = player.currentItem?.currentDate() {
            obs["program_ms"] = Int64((d.timeIntervalSince1970 * 1000).rounded(.down))
        }
        if let end = try? sink.observe(obs) {
            // §6.1: nobody can see it, or nobody attends to it: stop the player's requests.
            finish(end)
            return
        }
        if let c = collector, Date().timeIntervalSince(lastCMCD) >= cmcdEvery, let batch = try? sink.takeCMCD() {
            lastCMCD = Date()
            c.postCMCD(batch)
        }
    }

    private func finish(_ how: WaiSink.Ended) {
        guard ended == nil else { return }
        ended = how
        timer?.invalidate()
        timer = nil
        for o in observers { NotificationCenter.default.removeObserver(o) }
        observers = []
        statusObservation = nil
        player.pause()
        player.replaceCurrentItem(with: nil)
        let now = max(WaiSink.nowUs, startUs + 1)
        try? sink.event(how == .ended ? "end" : "abandon", atUs: now)
        let batch = try? sink.takeCMCD()
        guard let b = try? sink.end(how, atUs: now) else { return }
        bundle = b
        relay.stop()
        if let c = collector {
            if let batch { c.postCMCD(batch) }
            c.postSession(b)
        }
        onEnd?(b, how)
    }
}