Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
18 kB · 584 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585// Package livehls is the in-memory "muxl HLS for livestreams" window: it turns// the muxl event stream (init + canonical segments) into a sliding window of// recent segments held in memory, plus the HLS playlists that address them.//// Nothing is written to the filesystem. The canonical .m4s segments are the// durable unit (archived elsewhere, by ValidateMP4); this window only keeps// the most recent few for live playback and serves them straight from RAM://// - per-track init segment (ftyp+moov) — the EXT-X-MAP target// - a windowed ring of recent canonical segments, addressed by a monotonic// media-sequence number//// The segment bytes are appended verbatim, so any C2PA/S2PA signature over a// segment is preserved and a player fetches exactly what was signed. Playback// is plain fMP4 HLS — EXT-X-MAP for the init plus one URI per segment — with// an advancing EXT-X-MEDIA-SEQUENCE and no EXT-X-ENDLIST until// [Writer.Finalize] (which turns the live playlist into a VOD playlist).//// Typical use, driven off the muxl event stream a stream's segments arrive on// (locally signed or replicated from another node)://// w := livehls.NewWriter(livehls.WithWindow(6))// w.Observe(initEvent)// for ev := range segmentEvents {// w.Observe(ev)// }// // serve w.MasterPlaylist(...), w.MediaPlaylist(...), w.InitSegment(tid),// // w.SegmentData(tid, seq) over HTTP.package livehls
import ( "fmt" "math" "sort" "strconv" "strings" "sync" "time"
"stream.place/streamplace/pkg/muxl")
// Segment is one canonical segment held in the live window: a track's// verbatim [c2pa?][muxl][moof][mdat] bytes plus the timing the playlist needs.type Segment struct { // Seq is the monotonic media-sequence number, assigned in arrival order // and never reused — the value a player addresses the segment by and the // basis for EXT-X-MEDIA-SEQUENCE. Seq uint64 DurationTicks uint64 SampleCount uint32 data []byte // addedAt is the wall-clock time the segment was observed; segments older // than the Writer's retention are evicted so a stalled/ended stream's // window empties instead of a player replaying it forever. addedAt time.Time // open is a fragment still being assembled from signed segments (see // WithMinFragment); it is not advertised or served until it closes. open bool // pieces is how many signed segments the fragment is made of. pieces int}
// Size is the segment's byte length.func (s Segment) Size() int { return len(s.data) }
// Track is the per-track config plus its (windowed) in-memory segments.type Track struct { Type string Codec string Timescale uint32 Width uint32 Height uint32 Channels uint32 SampleRate uint32
Segments []Segment
// init is the per-track ftyp+moov (EXT-X-MAP target); not part of the // window. nextSeq is the next media-sequence number to hand out. init []byte nextSeq uint64}
// Writer assembles an in-memory live HLS window from the muxl event stream.// Safe for concurrent Observe and read/serve.type Writer struct { mu sync.Mutex tracks map[string]*Track order []string // track ids, sorted window int // segments advertised per track; 0 = all retained minDur time.Duration // the advertised window never spans less than this; 0 = count only minFrag time.Duration // signed segments shorter than this are joined into one fragment; 0 = none retention time.Duration // max segment age before eviction; 0 = keep 2×window instead finished bool now func() time.Time // clock, overridable in tests}
// Option configures a Writer.type Option func(*Writer)
// WithWindow advertises the newest n segments per track in the media// playlist (a sliding window), advancing EXT-X-MEDIA-SEQUENCE as older ones// leave it. Segments that have left the playlist stay fetchable until they// age out (WithRetention; without one, until 2n segments are held): RFC// 8216 §6.2.2 wants a segment served for its own duration plus the// playlist's after its URI is removed, and a player that is a little behind// otherwise finds the oldest listed segment gone the moment a new one// arrives. n <= 0 advertises everything retained (an "event"/VOD-style// window — unbounded memory, avoid for 24/7 streams).func WithWindow(n int) Option { return func(w *Writer) { w.window = n }}
// WithMinFragment joins signed segments into playlist fragments of at least// d: a segment shorter than d opens a fragment that the following segments// are appended to until it spans d, and only then is it advertised. Canonical// segments concatenate blindly (no container header) and every one starts// at a keyframe, so the join is a plain byte append and the fragment stays// independent. The point is a passage the encoder keyframes every few// frames: served one segment per keyframe it costs a player a round trip// per 40ms of video, which no buffer survives. Latency cost: a fragment is// held back until it spans d, so at most d during such a passage and nothing// otherwise. d <= 0 advertises every segment as it comes.func WithMinFragment(d time.Duration) Option { return func(w *Writer) { w.minFrag = d }}
// WithMinDuration keeps the count window from shrinking below d of media:// a segment is only evicted by count while the rest still spans at least// d. Segments are cut at keyframes, and an encoder that puts a keyframe on// every scene cut can hand over a run of one- and two-frame segments; with// a count-only window that run evicts whole seconds of the stream in a// moment, and a player a few seconds behind live finds its next segment// gone (404) — the stall at "the same part of the video" every time. Time// eviction (WithRetention) is unaffected. Needs the track's timescale (from// the catalog); until it's known the count alone applies.func WithMinDuration(d time.Duration) Option { return func(w *Writer) { w.minDur = d }}
// WithRetention evicts segments older than d (by wall-clock arrival time),// independent of the count window. Unlike the count window — which only evicts// when new segments arrive — this is also applied on reads, so a stalled or// ended stream's segments age out and its window empties rather than a player// replaying the last few forever. d <= 0 disables time eviction.func WithRetention(d time.Duration) Option { return func(w *Writer) { w.retention = d }}
// NewWriter returns an empty in-memory live HLS window.func NewWriter(opts ...Option) *Writer { w := &Writer{tracks: map[string]*Track{}, now: time.Now} for _, opt := range opts { opt(w) } return w}
// Observe folds one muxl event into the window. An "init" event supplies the// per-track init segments and catalog; "segment"/"signed-segment" events// append each track's canonical-segment bytes (copied — the caller may reuse// its buffer) and evict beyond the window.func (w *Writer) Observe(ev *muxl.MuxlEvent) error { w.mu.Lock() defer w.mu.Unlock()
switch ev.Type { case "init": for tid, initBytes := range ev.TrackInits { w.track(tid).init = append([]byte(nil), initBytes...) } if ev.Catalog != nil { w.applyCatalog(ev.Catalog) }
case "segment", "signed-segment": now := w.now() for _, tid := range sortedKeys(ev.Tracks) { t := w.track(tid) if n := len(t.Segments); n > 0 && t.Segments[n-1].open { // Grow the open fragment; it closes once it spans minFrag. f := &t.Segments[n-1] f.DurationTicks += ev.Durations[tid] f.SampleCount += ev.SampleCounts[tid] f.data = append(f.data, ev.Tracks[tid]...) f.pieces++ f.open = w.minFrag > 0 && f.seconds(t.Timescale) < w.minFrag.Seconds() continue } seg := Segment{ Seq: t.nextSeq, DurationTicks: ev.Durations[tid], SampleCount: ev.SampleCounts[tid], data: append([]byte(nil), ev.Tracks[tid]...), addedAt: now, pieces: 1, } // A timescale is needed to measure it; until the catalog has // arrived every segment stands alone. seg.open = w.minFrag > 0 && t.Timescale > 0 && seg.seconds(t.Timescale) < w.minFrag.Seconds() t.Segments = append(t.Segments, seg) t.nextSeq++ if w.retention <= 0 && w.window > 0 { // No time-based retention: hold a window's worth of grace // behind the advertised playlist, then let go. if start, _ := w.advertisedRange(t); start > w.window { drop := start - w.window t.Segments = append(t.Segments[:0:0], t.Segments[drop:]...) } } } w.evictExpired(now)
default: return fmt.Errorf("livehls: unexpected event type %q", ev.Type) } return nil}
// evictExpired drops segments older than the retention window (leading entries,// since they're in arrival order). Caller holds w.mu. No-op when retention is// unset. Called both on Observe and on reads so a stalled stream — which sends// no more events to trigger Observe — still ages out and empties.func (w *Writer) evictExpired(now time.Time) { if w.retention <= 0 { return } cutoff := now.Add(-w.retention) for _, t := range w.tracks { drop := 0 for drop < len(t.Segments) && t.Segments[drop].addedAt.Before(cutoff) { drop++ } if drop > 0 { t.Segments = append(t.Segments[:0:0], t.Segments[drop:]...) } }}
// Empty reports whether the window holds no segments on any track (after aging// out expired ones). The serving layer uses this to drop an ended stream's// window — freeing it and returning "no live stream" so players stop replaying// the stale tail.func (w *Writer) Empty() bool { w.mu.Lock() defer w.mu.Unlock() w.evictExpired(w.now()) for _, t := range w.tracks { if len(t.Segments) > 0 { return false } } return true}
// advertised is the slice of a track's retained segments the media playlist// lists: closed fragments only, the newest `window` of them, extended back// as far as needed to span minDur. Caller holds w.mu.func (w *Writer) advertised(t *Track) []Segment { start, end := w.advertisedRange(t) return t.Segments[start:end]}
// advertisedRange is advertised as indexes into t.Segments. Caller holds w.mu.func (w *Writer) advertisedRange(t *Track) (start, end int) { end = len(t.Segments) for end > 0 && t.Segments[end-1].open { end-- } if w.window <= 0 || end <= w.window { return 0, end } start = end - w.window if w.minDur > 0 && t.Timescale > 0 { span := 0.0 for i := end - 1; i >= 0; i-- { span += t.Segments[i].seconds(t.Timescale) if span >= w.minDur.Seconds() { if i < start { start = i } break } if i == 0 { start = 0 } } } return start, end}
// Finalize marks the stream complete; subsequent media playlists carry// EXT-X-ENDLIST and the VOD playlist type. A fragment still open is closed// as it is — the tail of the stream, however short, is worth serving.func (w *Writer) Finalize() { w.mu.Lock() defer w.mu.Unlock() w.finished = true for _, t := range w.tracks { if n := len(t.Segments); n > 0 { t.Segments[n-1].open = false } }}
// TrackIDs returns the known track ids in sorted order.func (w *Writer) TrackIDs() []string { w.mu.Lock() defer w.mu.Unlock() return append([]string(nil), w.order...)}
// VideoTrack describes one video track of the window: a variant of the// master playlist.type VideoTrack struct { ID string Codec string Width uint32 Height uint32}
// VideoTracks lists the window's video tracks in master-playlist order.func (w *Writer) VideoTracks() []VideoTrack { w.mu.Lock() defer w.mu.Unlock() var out []VideoTrack for _, tid := range w.order { t := w.tracks[tid] if t == nil || t.Type != "video" { continue } out = append(out, VideoTrack{ID: tid, Codec: t.Codec, Width: t.Width, Height: t.Height}) } return out}
// Track returns a snapshot of the track's current windowed segments (segment// byte payloads are shared, not copied), or nil if unknown.func (w *Writer) Track(trackID string) *Track { w.mu.Lock() defer w.mu.Unlock() t := w.tracks[trackID] if t == nil { return nil } cp := *t cp.Segments = append([]Segment(nil), w.advertised(t)...) cp.init = nil return &cp}
// InitSegment returns the per-track init (ftyp+moov) bytes — the EXT-X-MAP// target — or nil if unknown.func (w *Writer) InitSegment(trackID string) []byte { w.mu.Lock() defer w.mu.Unlock() if t := w.tracks[trackID]; t != nil { return t.init } return nil}
// SegmentData returns the bytes of the segment with media-sequence seq for// trackID, or nil if it's unknown, still being assembled, or has aged out.// Segments that have left the playlist but not aged out are still served// (see WithWindow).func (w *Writer) SegmentData(trackID string, seq uint64) []byte { w.mu.Lock() defer w.mu.Unlock() w.evictExpired(w.now()) t := w.tracks[trackID] if t == nil { return nil } for _, s := range t.Segments { if s.Seq == seq && !s.open { return s.data } } return nil}
// PrimaryAudioTrackID returns the track ID of the primary audio track,// preferring AAC (broadest HLS support) over other codecs. Returns ""// if no audio track exists. The selection mirrors MasterPlaylist's logic.func (w *Writer) PrimaryAudioTrackID() string { w.mu.Lock() defer w.mu.Unlock() primaryAudio := "" for _, tid := range w.order { if t := w.tracks[tid]; t != nil && t.Type == "audio" { if primaryAudio == "" { primaryAudio = tid } if strings.HasPrefix(t.Codec, "mp4a") { return tid } } } return primaryAudio}
// MediaPlaylist renders the live HLS media playlist for trackID. initURL is// the EXT-X-MAP target (the per-track init); segURI maps a segment's// media-sequence number to its URI. Returns "" for an unknown track.func (w *Writer) MediaPlaylist(trackID, initURL string, segURI func(seq uint64) string) string { w.mu.Lock() defer w.mu.Unlock() w.evictExpired(w.now()) t := w.tracks[trackID] if t == nil { return "" }
segs := w.advertised(t) maxDur := 0.0 for _, s := range segs { if d := s.seconds(t.Timescale); d > maxDur { maxDur = d } } target := int(math.Ceil(maxDur)) if target < 1 { target = 1 } mediaSeq := uint64(0) if len(segs) > 0 { mediaSeq = segs[0].Seq }
var b strings.Builder b.WriteString("#EXTM3U\n#EXT-X-VERSION:7\n") fmt.Fprintf(&b, "#EXT-X-TARGETDURATION:%d\n", target) fmt.Fprintf(&b, "#EXT-X-MEDIA-SEQUENCE:%d\n", mediaSeq) b.WriteString("#EXT-X-INDEPENDENT-SEGMENTS\n") if w.finished { b.WriteString("#EXT-X-PLAYLIST-TYPE:VOD\n") } fmt.Fprintf(&b, "#EXT-X-MAP:URI=%q\n", initURL) for _, s := range segs { fmt.Fprintf(&b, "#EXTINF:%.6f,\n%s\n", s.seconds(t.Timescale), segURI(s.Seq)) } if w.finished { b.WriteString("#EXT-X-ENDLIST\n") } return b.String()}
// MasterPlaylist renders the HLS master playlist. trackURL maps a track id to// its media-playlist URL. Audio tracks become EXT-X-MEDIA renditions in the// "audio" group; video tracks become EXT-X-STREAM-INF variants.func (w *Writer) MasterPlaylist(trackURL func(trackID string) string) string { w.mu.Lock() defer w.mu.Unlock() w.evictExpired(w.now())
var b strings.Builder b.WriteString("#EXTM3U\n#EXT-X-VERSION:7\n#EXT-X-INDEPENDENT-SEGMENTS\n")
// Pick the primary audio rendition: prefer AAC (broadest HLS support — // Safari has no Opus) so it's the default; any other codec (Opus) stays a // selectable alternate. A segment may carry both after codec completion. primaryAudio := "" for _, tid := range w.order { if t := w.tracks[tid]; t.Type == "audio" { if primaryAudio == "" { primaryAudio = tid } if strings.HasPrefix(t.Codec, "mp4a") { primaryAudio = tid break } } } var audioCodec string haveAudio := primaryAudio != "" for _, tid := range w.order { t := w.tracks[tid] if t.Type != "audio" { continue } def := "NO" if tid == primaryAudio { def = "YES" audioCodec = t.Codec } fmt.Fprintf(&b, "#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID=%q,NAME=%q,DEFAULT=%s,AUTOSELECT=YES,CHANNELS=%q,URI=%q\n", "audio", t.Codec, def, strconv.Itoa(int(maxU32(t.Channels, 2))), trackURL(tid)) } for _, tid := range w.order { t := w.tracks[tid] if t.Type != "video" { continue } codecs := t.Codec if audioCodec != "" { codecs = t.Codec + "," + audioCodec } bw := t.peakBitrate() if haveAudio { fmt.Fprintf(&b, "#EXT-X-STREAM-INF:AUDIO=%q,BANDWIDTH=%d,CODECS=%q,RESOLUTION=%dx%d\n", "audio", bw, codecs, t.Width, t.Height) } else { fmt.Fprintf(&b, "#EXT-X-STREAM-INF:BANDWIDTH=%d,CODECS=%q,RESOLUTION=%dx%d\n", bw, codecs, t.Width, t.Height) } b.WriteString(trackURL(tid) + "\n") } return b.String()}
// --- internals ---
func (w *Writer) track(tid string) *Track { t := w.tracks[tid] if t == nil { t = &Track{Type: "unknown"} w.tracks[tid] = t w.order = append(w.order, tid) sort.Strings(w.order) } return t}
// applyCatalog fills per-track config from the muxl catalog.func (w *Writer) applyCatalog(cat *muxl.MuxlCatalog) { if cat.Video != nil { for _, c := range cat.Video.Renditions { t := w.track(strconv.FormatUint(uint64(c.TrackID()), 10)) t.Type = "video" t.Codec = c.Codec t.Timescale = c.Timescale() t.Width = c.CodedWidth t.Height = c.CodedHeight } } if cat.Audio != nil { for _, c := range cat.Audio.Renditions { t := w.track(strconv.FormatUint(uint64(c.TrackID()), 10)) t.Type = "audio" t.Codec = c.Codec t.Timescale = c.Timescale() t.Channels = c.NumberOfChannels t.SampleRate = c.SampleRate } }}
func (s Segment) seconds(timescale uint32) float64 { if timescale == 0 { return 0 } return float64(s.DurationTicks) / float64(timescale)}
// peakBitrate estimates BANDWIDTH (bits/sec) from the windowed segments.func (t *Track) peakBitrate() int { var bytes int64 var ticks uint64 for _, s := range t.Segments { bytes += int64(s.Size()) ticks += s.DurationTicks } if ticks == 0 || t.Timescale == 0 { return 0 } secs := float64(ticks) / float64(t.Timescale) return int(float64(bytes*8) / secs)}
func sortedKeys(m map[string][]byte) []string { keys := make([]string, 0, len(m)) for k := range m { keys = append(keys, k) } sort.Strings(keys) return keys}
func maxU32(a, b uint32) uint32 { if a > b { return a } return b}