package playback import ( "context" "errors" "sync" "time" ) type PlaylistReadiness struct { Revision uint64 } type playlistTimer interface { C() <-chan time.Time Stop() bool } type playlistTimerFactory func(time.Duration) playlistTimer type realPlaylistTimer struct { timer *time.Timer } func (t realPlaylistTimer) C() <-chan time.Time { return t.timer.C } func (t realPlaylistTimer) Stop() bool { return t.timer.Stop() } type PlaylistController struct { playlist Playlist retry RetryPolicy sessions chan<- SessionCommand now func() time.Time newTimer playlistTimerFactory mu sync.RWMutex snapshot PlaylistSnapshot hasSnapshot bool } type PlaylistSnapshot struct { State PlaylistState Entry PlaylistEntry Revision uint64 Timing PlaylistTimingState } var ( ErrNilSessionCommandChannel = errors.New("session-command channel is nil") ) func NewPlaylistController( playlist Playlist, retry RetryPolicy, sessions chan<- SessionCommand, ) (*PlaylistController, error) { if err := playlist.Validate(); err != nil { return nil, err } if err := retry.Validate(); err != nil { return nil, err } if sessions == nil { return nil, ErrNilSessionCommandChannel } return &PlaylistController{ playlist: playlist, retry: retry, sessions: sessions, now: time.Now, newTimer: func(duration time.Duration) playlistTimer { return realPlaylistTimer{timer: time.NewTimer(duration)} }, }, nil } func (c *PlaylistController) Run( ctx context.Context, commands <-chan PlaylistCommand, readiness <-chan PlaylistReadiness, ) error { state := PlaylistState{} revision := uint64(0) timing := PlaylistTimingState{} var timer playlistTimer var timerC <-chan time.Time var timerRevision uint64 c.publish(state, revision, timing) stopTimer := func() { stopPlaylistTimer(timer) timer = nil timerC = nil } defer stopTimer() for { select { case <-ctx.Done(): return ctx.Err() case command, ok := <-commands: if !ok { return nil } next, sessionCommand, apply, err := ApplyPlaylistSelection( c.playlist, state, command, c.retry, ) if err != nil { continue } if apply { stopTimer() select { case <-ctx.Done(): return ctx.Err() case c.sessions <- sessionCommand: } revision++ entry, _ := next.Entry(c.playlist) timing = NewPlaylistTiming(revision, entry.Duration) } state = next c.publish(state, revision, timing) case ready, ok := <-readiness: if !ok { readiness = nil continue } nextTiming, started := StartPlaylistTiming( timing, ready.Revision, c.now(), ) if !started { continue } timing = nextTiming timerRevision = timing.Revision timer = c.newTimer(timing.Duration) timerC = timer.C() c.publish(state, revision, timing) case firedAt := <-timerC: firedRevision := timerRevision timer = nil timerC = nil nextTiming, expired := ExpirePlaylistTiming( timing, firedRevision, firedAt, ) if !expired { continue } timing = nextTiming next, sessionCommand, apply, err := ApplyPlaylistSelection( c.playlist, state, PlaylistCommand{Kind: PlaylistNext}, c.retry, ) if err != nil { c.publish(state, revision, timing) continue } if apply { select { case <-ctx.Done(): return ctx.Err() case c.sessions <- sessionCommand: } revision++ entry, _ := next.Entry(c.playlist) timing = NewPlaylistTiming(revision, entry.Duration) } state = next c.publish(state, revision, timing) } } } func (c *PlaylistController) Snapshot() (PlaylistSnapshot, bool) { c.mu.RLock() defer c.mu.RUnlock() return c.snapshot, c.hasSnapshot } func (c *PlaylistController) publish( state PlaylistState, revision uint64, timing PlaylistTimingState, ) { entry, _ := state.Entry(c.playlist) c.mu.Lock() c.snapshot = PlaylistSnapshot{ State: state, Entry: entry, Revision: revision, Timing: timing, } c.hasSnapshot = true c.mu.Unlock() } func stopPlaylistTimer(timer playlistTimer) { if timer == nil || timer.Stop() { return } select { case <-timer.C(): default: } }