diff --git a/imgui.ini b/imgui.ini index f8dd632..6a98b59 100644 --- a/imgui.ini +++ b/imgui.ini @@ -29,8 +29,8 @@ Size=550,720 Collapsed=0 [Window][Settings & Info] -Pos=1220,0 -Size=700,1080 +Pos=580,0 +Size=700,720 Collapsed=0 [Window][Seetings & Info] diff --git a/internal/playback/playlist_readiness.go b/internal/playback/playlist_readiness.go index 201860c..143a3bb 100644 --- a/internal/playback/playlist_readiness.go +++ b/internal/playback/playlist_readiness.go @@ -1,5 +1,151 @@ package playback +import ( + "context" + "errors" + "time" +) + +type PlaylistSnapshotSource interface { + Snapshot() (PlaylistSnapshot, bool) +} + +type SessionSnapshotSource interface { + Snapshot() (SessionSnapshot, bool) +} + +type PlaybackStatusSnapshotSource interface { + SnapshotAll() PlaybackStatusSnapshot +} + +type playlistReadinessTicker interface { + C() <-chan time.Time + Stop() +} + +type playlistReadinessTickerFactory func(time.Duration) playlistReadinessTicker + +type realPlaylistReadinessTicker struct { + ticker *time.Ticker +} + +func (t realPlaylistReadinessTicker) C() <-chan time.Time { return t.ticker.C } +func (t realPlaylistReadinessTicker) Stop() { t.ticker.Stop() } + +var ( + ErrPlaylistSnapshotSourceRequired = errors.New("playlist snapshot source is required") + ErrSessionSnapshotSourceRequired = errors.New("session snapshot source is required") + ErrStatusSnapshotSourceRequired = errors.New("playback status snapshot source is required") + ErrPlaylistReadinessOutputRequired = errors.New("playlist readiness output channel is required") + ErrPlaylistReadinessInterval = errors.New("playlist readiness interval must be positive") +) + +type PlaylistReadinessCoordinator struct { + playlist PlaylistSnapshotSource + session SessionSnapshotSource + statuses PlaybackStatusSnapshotSource + output chan<- PlaylistReadiness + interval time.Duration + + newTicker playlistReadinessTickerFactory +} + +func NewPlaylistReadinessCoordinator( + playlist PlaylistSnapshotSource, + session SessionSnapshotSource, + statuses PlaybackStatusSnapshotSource, + output chan<- PlaylistReadiness, + interval time.Duration, +) (*PlaylistReadinessCoordinator, error) { + if playlist == nil { + return nil, ErrPlaylistSnapshotSourceRequired + } + if session == nil { + return nil, ErrSessionSnapshotSourceRequired + } + if statuses == nil { + return nil, ErrStatusSnapshotSourceRequired + } + if output == nil { + return nil, ErrPlaylistReadinessOutputRequired + } + if interval <= 0 { + return nil, ErrPlaylistReadinessInterval + } + + return &PlaylistReadinessCoordinator{ + playlist: playlist, + session: session, + statuses: statuses, + output: output, + interval: interval, + newTicker: func(interval time.Duration) playlistReadinessTicker { + return realPlaylistReadinessTicker{ticker: time.NewTicker(interval)} + }, + }, nil +} + +func (c *PlaylistReadinessCoordinator) Run(ctx context.Context) error { + ticker := c.newTicker(c.interval) + defer ticker.Stop() + + var emittedRevision uint64 + for { + select { + case <-ctx.Done(): + return ctx.Err() + + case <-ticker.C(): + playlistSnapshot, ok := c.playlist.Snapshot() + if !ok || + !playlistSnapshot.State.HasSelection || + playlistSnapshot.Revision == 0 || + playlistSnapshot.Entry.Duration <= 0 || + playlistSnapshot.Timing.Started || + playlistSnapshot.Revision == emittedRevision { + continue + } + + sessionSnapshot, ok := c.session.Snapshot() + if !ok || !PlaylistEntryMatchesSession( + playlistSnapshot.Entry, + sessionSnapshot.Desired, + ) { + continue + } + if !IsSessionPlaying(sessionSnapshot, c.statuses.SnapshotAll()) { + continue + } + + ready := PlaylistReadiness{Revision: playlistSnapshot.Revision} + select { + case <-ctx.Done(): + return ctx.Err() + case c.output <- ready: + emittedRevision = playlistSnapshot.Revision + } + } + } +} + +func PlaylistEntryMatchesSession(entry PlaylistEntry, session SessionConfig) bool { + if err := entry.Validate(); err != nil { + return false + } + return playlistFeedMatchesSession(entry.Video, session.Video) && + playlistFeedMatchesSession(entry.Audio, session.Audio) && + entry.SyncRequested == session.SyncRequested +} + +func playlistFeedMatchesSession(playlist PlaylistFeed, session FeedConfig) bool { + if !playlist.IsConfigured() { + return !session.IsConfigured() && !session.Active + } + return session.Active && + playlist.Domain == session.Domain && + playlist.UUID == session.UUID +} + func IsSessionPlaying( session SessionSnapshot, statuses PlaybackStatusSnapshot, diff --git a/internal/playback/playlist_readiness_coordinator_test.go b/internal/playback/playlist_readiness_coordinator_test.go new file mode 100644 index 0000000..d1a65c3 --- /dev/null +++ b/internal/playback/playlist_readiness_coordinator_test.go @@ -0,0 +1,368 @@ +package playback + +import ( + "context" + "errors" + "sync" + "testing" + "time" +) + +type fakePlaylistSnapshotSource struct { + mu sync.RWMutex + snapshot PlaylistSnapshot + ok bool +} + +func (s *fakePlaylistSnapshotSource) Snapshot() (PlaylistSnapshot, bool) { + s.mu.RLock() + defer s.mu.RUnlock() + return s.snapshot, s.ok +} + +func (s *fakePlaylistSnapshotSource) set(snapshot PlaylistSnapshot, ok bool) { + s.mu.Lock() + s.snapshot = snapshot + s.ok = ok + s.mu.Unlock() +} + +type fakeSessionSnapshotSource struct { + mu sync.RWMutex + snapshot SessionSnapshot + ok bool +} + +func (s *fakeSessionSnapshotSource) Snapshot() (SessionSnapshot, bool) { + s.mu.RLock() + defer s.mu.RUnlock() + return s.snapshot, s.ok +} + +func (s *fakeSessionSnapshotSource) set(snapshot SessionSnapshot, ok bool) { + s.mu.Lock() + s.snapshot = snapshot + s.ok = ok + s.mu.Unlock() +} + +type fakePlaybackStatusSnapshotSource struct { + mu sync.RWMutex + snapshot PlaybackStatusSnapshot +} + +func (s *fakePlaybackStatusSnapshotSource) SnapshotAll() PlaybackStatusSnapshot { + s.mu.RLock() + defer s.mu.RUnlock() + return s.snapshot +} + +func (s *fakePlaybackStatusSnapshotSource) set(snapshot PlaybackStatusSnapshot) { + s.mu.Lock() + s.snapshot = snapshot + s.mu.Unlock() +} + +type fakePlaylistReadinessTicker struct { + ch chan time.Time + + mu sync.Mutex + stopped bool +} + +func newFakePlaylistReadinessTicker() *fakePlaylistReadinessTicker { + return &fakePlaylistReadinessTicker{ch: make(chan time.Time, 16)} +} + +func (t *fakePlaylistReadinessTicker) C() <-chan time.Time { return t.ch } +func (t *fakePlaylistReadinessTicker) Stop() { + t.mu.Lock() + t.stopped = true + t.mu.Unlock() +} +func (t *fakePlaylistReadinessTicker) tick() { t.ch <- time.Now() } +func (t *fakePlaylistReadinessTicker) isStopped() bool { + t.mu.Lock() + defer t.mu.Unlock() + return t.stopped +} + +func TestNewPlaylistReadinessCoordinatorValidatesDependencies(t *testing.T) { + playlist := &fakePlaylistSnapshotSource{} + session := &fakeSessionSnapshotSource{} + statuses := &fakePlaybackStatusSnapshotSource{} + output := make(chan PlaylistReadiness) + tests := []struct { + name string + playlist PlaylistSnapshotSource + session SessionSnapshotSource + statuses PlaybackStatusSnapshotSource + output chan<- PlaylistReadiness + interval time.Duration + wantErr error + }{ + {name: "playlist", session: session, statuses: statuses, output: output, interval: time.Millisecond, wantErr: ErrPlaylistSnapshotSourceRequired}, + {name: "session", playlist: playlist, statuses: statuses, output: output, interval: time.Millisecond, wantErr: ErrSessionSnapshotSourceRequired}, + {name: "statuses", playlist: playlist, session: session, output: output, interval: time.Millisecond, wantErr: ErrStatusSnapshotSourceRequired}, + {name: "output", playlist: playlist, session: session, statuses: statuses, interval: time.Millisecond, wantErr: ErrPlaylistReadinessOutputRequired}, + {name: "interval", playlist: playlist, session: session, statuses: statuses, output: output, wantErr: ErrPlaylistReadinessInterval}, + {name: "valid", playlist: playlist, session: session, statuses: statuses, output: output, interval: time.Millisecond}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + coordinator, err := NewPlaylistReadinessCoordinator( + test.playlist, + test.session, + test.statuses, + test.output, + test.interval, + ) + if !errors.Is(err, test.wantErr) { + t.Fatalf("NewPlaylistReadinessCoordinator() error = %v, want %v", err, test.wantErr) + } + if test.wantErr != nil && coordinator != nil { + t.Fatalf("coordinator = %#v, want nil", coordinator) + } + }) + } +} + +func TestPlaylistEntryMatchesSession(t *testing.T) { + entry := PlaylistEntry{ + Video: PlaylistFeed{Domain: "video-domain", UUID: "video"}, + Audio: PlaylistFeed{Domain: "audio-domain", UUID: "audio"}, + SyncRequested: true, + } + matching := entry.SessionConfig(validPlaylistRetryPolicy()) + tests := []struct { + name string + entry PlaylistEntry + session SessionConfig + want bool + }{ + {name: "matching", entry: entry, session: matching, want: true}, + {name: "retry ignored", entry: entry, session: func() SessionConfig { value := matching; value.Retry.MaxAttempts = 99; return value }(), want: true}, + {name: "wrong video UUID", entry: entry, session: func() SessionConfig { value := matching; value.Video.UUID = "other"; return value }()}, + {name: "wrong audio domain", entry: entry, session: func() SessionConfig { value := matching; value.Audio.Domain = "other"; return value }()}, + {name: "inactive video", entry: entry, session: func() SessionConfig { value := matching; value.Video.Active = false; return value }()}, + {name: "wrong sync request", entry: entry, session: func() SessionConfig { value := matching; value.SyncRequested = false; return value }()}, + { + name: "absent audio matches unconfigured inactive", + entry: PlaylistEntry{Video: entry.Video}, + session: PlaylistEntry{Video: entry.Video}.SessionConfig(validPlaylistRetryPolicy()), + want: true, + }, + { + name: "absent audio rejects configured audio", + entry: PlaylistEntry{Video: entry.Video}, + session: SessionConfig{ + Video: matching.Video, + Audio: matching.Audio, + Retry: matching.Retry, + }, + }, + {name: "invalid entry", entry: PlaylistEntry{}, session: matching}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := PlaylistEntryMatchesSession(test.entry, test.session); got != test.want { + t.Fatalf("PlaylistEntryMatchesSession() = %v, want %v", got, test.want) + } + }) + } +} + +func TestPlaylistReadinessCoordinatorEmitsOncePerRevision(t *testing.T) { + playlist, session, statuses := readyVideoSnapshots(1) + output := make(chan PlaylistReadiness, 4) + coordinator, ticker, cancel, result := startReadinessCoordinator( + t, + playlist, + session, + statuses, + output, + ) + _ = coordinator + defer cancel() + + ticker.tick() + if got := receivePlaylistReadiness(t, output); got.Revision != 1 { + t.Fatalf("readiness revision = %d, want 1", got.Revision) + } + ticker.tick() + assertNoPlaylistReadiness(t, output) + + next := playlistSnapshotForVideo(2) + playlist.set(next, true) + ticker.tick() + if got := receivePlaylistReadiness(t, output); got.Revision != 2 { + t.Fatalf("readiness revision = %d, want 2", got.Revision) + } + + cancel() + if err := waitForPlaylistResult(t, result); !errors.Is(err, context.Canceled) { + t.Fatalf("Run() error = %v, want %v", err, context.Canceled) + } + if !ticker.isStopped() { + t.Fatal("ticker was not stopped") + } +} + +func TestPlaylistReadinessCoordinatorWaitsForAllConditions(t *testing.T) { + playlist, session, statuses := readyVideoSnapshots(1) + output := make(chan PlaylistReadiness, 1) + _, ticker, cancel, result := startReadinessCoordinator(t, playlist, session, statuses, output) + defer cancel() + + tests := []struct { + name string + mutate func() + }{ + {name: "no playlist snapshot", mutate: func() { playlist.set(PlaylistSnapshot{}, false) }}, + {name: "no selection", mutate: func() { + value := playlistSnapshotForVideo(1) + value.State.HasSelection = false + playlist.set(value, true) + }}, + {name: "zero revision", mutate: func() { value := playlistSnapshotForVideo(1); value.Revision = 0; playlist.set(value, true) }}, + {name: "zero duration", mutate: func() { value := playlistSnapshotForVideo(1); value.Entry.Duration = 0; playlist.set(value, true) }}, + {name: "already started", mutate: func() { value := playlistSnapshotForVideo(1); value.Timing.Started = true; playlist.set(value, true) }}, + {name: "session mismatch", mutate: func() { + playlist.set(playlistSnapshotForVideo(1), true) + value, _ := session.Snapshot() + value.Desired.Video.UUID = "other" + session.set(value, true) + }}, + {name: "stale statuses", mutate: func() { + playlist.set(playlistSnapshotForVideo(1), true) + _, validSession, _ := readyVideoSnapshots(1) + value, _ := validSession.Snapshot() + session.set(value, true) + current := statuses.SnapshotAll() + current.Generation = 2 + current.Video.Generation = 2 + statuses.set(current) + }}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + validPlaylist, validSession, validStatuses := readyVideoSnapshots(1) + playlist.set(validPlaylist.snapshot, true) + session.set(validSession.snapshot, true) + statuses.set(validStatuses.snapshot) + test.mutate() + ticker.tick() + assertNoPlaylistReadiness(t, output) + }) + } + + cancel() + _ = waitForPlaylistResult(t, result) +} + +func TestPlaylistReadinessCoordinatorCancellationWhileBlockedSending(t *testing.T) { + playlist, session, statuses := readyVideoSnapshots(1) + output := make(chan PlaylistReadiness) + _, ticker, cancel, result := startReadinessCoordinator(t, playlist, session, statuses, output) + + ticker.tick() + time.Sleep(time.Millisecond) + cancel() + if err := waitForPlaylistResult(t, result); !errors.Is(err, context.Canceled) { + t.Fatalf("Run() error = %v, want %v", err, context.Canceled) + } +} + +func readyVideoSnapshots(revision uint64) ( + *fakePlaylistSnapshotSource, + *fakeSessionSnapshotSource, + *fakePlaybackStatusSnapshotSource, +) { + playlist := &fakePlaylistSnapshotSource{snapshot: playlistSnapshotForVideo(revision), ok: true} + entry := playlist.snapshot.Entry + desired := entry.SessionConfig(validPlaylistRetryPolicy()) + session := &fakeSessionSnapshotSource{ + snapshot: SessionSnapshot{ + Desired: desired, + Plan: SessionPlan{Topology: TopologyIndependent, Video: desired.Video}, + Generation: 5, + }, + ok: true, + } + statuses := &fakePlaybackStatusSnapshotSource{ + snapshot: PlaybackStatusSnapshot{ + Generation: 5, + Video: Status{ + Unit: UnitVideo, + State: StatePlaying, + Generation: 5, + Feed: desired.Video, + }, + HasVideo: true, + }, + } + return playlist, session, statuses +} + +func playlistSnapshotForVideo(revision uint64) PlaylistSnapshot { + entry := PlaylistEntry{ + Name: "video", + Video: PlaylistFeed{Domain: "domain", UUID: "video"}, + Duration: 10 * time.Second, + } + return PlaylistSnapshot{ + State: PlaylistState{CurrentIndex: 0, HasSelection: true}, + Entry: entry, + Revision: revision, + Timing: NewPlaylistTiming(revision, entry.Duration), + } +} + +func startReadinessCoordinator( + t *testing.T, + playlist PlaylistSnapshotSource, + session SessionSnapshotSource, + statuses PlaybackStatusSnapshotSource, + output chan<- PlaylistReadiness, +) (*PlaylistReadinessCoordinator, *fakePlaylistReadinessTicker, context.CancelFunc, <-chan error) { + t.Helper() + coordinator, err := NewPlaylistReadinessCoordinator( + playlist, + session, + statuses, + output, + time.Millisecond, + ) + if err != nil { + t.Fatalf("NewPlaylistReadinessCoordinator() error = %v", err) + } + ticker := newFakePlaylistReadinessTicker() + coordinator.newTicker = func(time.Duration) playlistReadinessTicker { return ticker } + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + go func() { result <- coordinator.Run(ctx) }() + return coordinator, ticker, cancel, result +} + +func receivePlaylistReadiness(t *testing.T, output <-chan PlaylistReadiness) PlaylistReadiness { + t.Helper() + select { + case readiness := <-output: + return readiness + case <-time.After(time.Second): + t.Fatal("timed out waiting for playlist readiness") + return PlaylistReadiness{} + } +} + +func assertNoPlaylistReadiness(t *testing.T, output <-chan PlaylistReadiness) { + t.Helper() + select { + case readiness := <-output: + t.Fatalf("unexpected playlist readiness: %#v", readiness) + case <-time.After(20 * time.Millisecond): + } +}