diff --git a/cmd/mxl-player/playlist_runtime.go b/cmd/mxl-player/playlist_runtime.go new file mode 100644 index 0000000..3d88c71 --- /dev/null +++ b/cmd/mxl-player/playlist_runtime.go @@ -0,0 +1,127 @@ +package main + +import ( + "context" + "errors" + "time" + + "mxl-player/internal/playback" +) + +const playlistReadinessInterval = 10 * time.Millisecond + +var ( + ErrPlayerPlaybackRequired = errors.New("player playback is required") + ErrPlayerSessionControllerRequired = errors.New("player session controller is required") + ErrPlayerStatusStoreRequired = errors.New("player status store is required") +) + +type playerPlaylist struct { + Controller *playback.PlaylistController + Coordinator *playback.PlaylistReadinessCoordinator + Commands chan playback.PlaylistCommand + Readiness chan playback.PlaylistReadiness +} + +func newPlayerPlaylist( + playlist playback.Playlist, + retry playback.RetryPolicy, + player *playerPlayback, +) (*playerPlaylist, error) { + if player == nil { + return nil, ErrPlayerPlaybackRequired + } + if player.Controller == nil { + return nil, ErrPlayerSessionControllerRequired + } + if player.Status == nil { + return nil, ErrPlayerStatusStoreRequired + } + + commands := make(chan playback.PlaylistCommand, 32) + readiness := make(chan playback.PlaylistReadiness, 8) + controller, err := playback.NewPlaylistController( + playlist, + retry, + player.Commands, + ) + if err != nil { + return nil, err + } + coordinator, err := playback.NewPlaylistReadinessCoordinator( + controller, + player.Controller, + player.Status, + readiness, + playlistReadinessInterval, + ) + if err != nil { + return nil, err + } + + return &playerPlaylist{ + Controller: controller, + Coordinator: coordinator, + Commands: commands, + Readiness: readiness, + }, nil +} + +func (p *playerPlaylist) Run(ctx context.Context) error { + runCtx, cancel := context.WithCancel(ctx) + defer cancel() + + results := make(chan error, 2) + go func() { + results <- p.Controller.Run(runCtx, p.Commands, p.Readiness) + }() + go func() { + results <- p.Coordinator.Run(runCtx) + }() + + first := <-results + cancel() + second := <-results + + if ctx.Err() != nil { + return ctx.Err() + } + if err := playlistRuntimeError(first); err != nil { + return err + } + if err := playlistRuntimeError(second); err != nil { + return err + } + return nil +} + +func (p *playerPlaylist) Select(index int) bool { + return p.enqueue(playback.PlaylistCommand{ + Kind: playback.PlaylistSelect, + Index: index, + }) +} + +func (p *playerPlaylist) Next() bool { + return p.enqueue(playback.PlaylistCommand{Kind: playback.PlaylistNext}) +} + +func (p *playerPlaylist) Previous() bool { + return p.enqueue(playback.PlaylistCommand{Kind: playback.PlaylistPrevious}) +} + +func (p *playerPlaylist) enqueue(command playback.PlaylistCommand) bool { + select { + case p.Commands <- command: + return true + default: + return false + } +} + +func playlistRuntimeError(err error) error { + if err == nil || errors.Is(err, context.Canceled) { + return nil + } + return err +} diff --git a/cmd/mxl-player/playlist_runtime_test.go b/cmd/mxl-player/playlist_runtime_test.go new file mode 100644 index 0000000..74e013a --- /dev/null +++ b/cmd/mxl-player/playlist_runtime_test.go @@ -0,0 +1,303 @@ +package main + +import ( + "context" + "errors" + "testing" + "time" + + "mxl-player/internal/playback" +) + +type playlistTestVideoSlot struct{} + +func (playlistTestVideoSlot) Run( + ctx context.Context, + initial playback.FeedConfig, + commands <-chan playback.FeedConfig, +) error { + for { + select { + case <-ctx.Done(): + return ctx.Err() + case _, ok := <-commands: + if !ok { + return nil + } + } + } +} + +type playlistTestAudioSlot struct{} + +func (playlistTestAudioSlot) Run( + ctx context.Context, + initial playback.FeedConfig, + commands <-chan playback.FeedConfig, +) error { + for { + select { + case <-ctx.Done(): + return ctx.Err() + case _, ok := <-commands: + if !ok { + return nil + } + } + } +} + +type playlistTestSyncSlot struct{} + +func (playlistTestSyncSlot) Run( + ctx context.Context, + initial playback.SyncPairConfig, + commands <-chan playback.SyncPairConfig, +) error { + for { + select { + case <-ctx.Done(): + return ctx.Err() + case _, ok := <-commands: + if !ok { + return nil + } + } + } +} + +func TestNewPlayerPlaylistValidatesPlayer(t *testing.T) { + retry := playlistRuntimeRetry() + tests := []struct { + name string + player *playerPlayback + wantErr error + }{ + {name: "nil player", wantErr: ErrPlayerPlaybackRequired}, + {name: "nil controller", player: &playerPlayback{}, wantErr: ErrPlayerSessionControllerRequired}, + { + name: "nil status store", + player: &playerPlayback{ + Controller: newPlaylistTestSessionController(t), + }, + wantErr: ErrPlayerStatusStoreRequired, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got, err := newPlayerPlaylist(playback.Playlist{}, retry, test.player) + if !errors.Is(err, test.wantErr) { + t.Fatalf("newPlayerPlaylist() error = %v, want %v", err, test.wantErr) + } + if got != nil { + t.Fatalf("newPlayerPlaylist() = %#v, want nil", got) + } + }) + } +} + +func TestNewPlayerPlaylistWiresComponents(t *testing.T) { + player := newPlaylistTestPlayer(t) + runtime, err := newPlayerPlaylist(playback.Playlist{}, playlistRuntimeRetry(), player) + if err != nil { + t.Fatalf("newPlayerPlaylist() error = %v", err) + } + if runtime.Controller == nil || runtime.Coordinator == nil { + t.Fatalf("runtime components = %#v", runtime) + } + if runtime.Commands == nil || runtime.Readiness == nil { + t.Fatalf("runtime channels = %#v", runtime) + } +} + +func TestPlayerPlaylistNavigationHelpers(t *testing.T) { + runtime, err := newPlayerPlaylist( + playback.Playlist{}, + playlistRuntimeRetry(), + newPlaylistTestPlayer(t), + ) + if err != nil { + t.Fatalf("newPlayerPlaylist() error = %v", err) + } + tests := []struct { + send func() bool + want playback.PlaylistCommand + }{ + {send: func() bool { return runtime.Select(4) }, want: playback.PlaylistCommand{Kind: playback.PlaylistSelect, Index: 4}}, + {send: runtime.Next, want: playback.PlaylistCommand{Kind: playback.PlaylistNext}}, + {send: runtime.Previous, want: playback.PlaylistCommand{Kind: playback.PlaylistPrevious}}, + } + for _, test := range tests { + if !test.send() { + t.Fatal("navigation helper returned false") + } + if got := <-runtime.Commands; got != test.want { + t.Fatalf("navigation command = %#v, want %#v", got, test.want) + } + } +} + +func TestPlayerPlaylistNavigationQueueFull(t *testing.T) { + runtime, err := newPlayerPlaylist( + playback.Playlist{}, + playlistRuntimeRetry(), + newPlaylistTestPlayer(t), + ) + if err != nil { + t.Fatalf("newPlayerPlaylist() error = %v", err) + } + for range cap(runtime.Commands) { + if !runtime.Next() { + t.Fatal("queue filled before reaching capacity") + } + } + if runtime.Next() { + t.Fatal("Next() = true with full queue") + } +} + +func TestPlayerPlaylistRunCancellationJoinsComponents(t *testing.T) { + runtime, err := newPlayerPlaylist( + playback.Playlist{}, + playlistRuntimeRetry(), + newPlaylistTestPlayer(t), + ) + if err != nil { + t.Fatalf("newPlayerPlaylist() error = %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + go func() { result <- runtime.Run(ctx) }() + cancel() + + select { + case err := <-result: + if !errors.Is(err, context.Canceled) { + t.Fatalf("Run() error = %v, want %v", err, context.Canceled) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for playlist runtime cancellation") + } +} + +func TestPlayerPlaylistTimedEntryAdvances(t *testing.T) { + retry := playlistRuntimeRetry() + player := newPlaylistTestPlayer(t) + playlist := playback.Playlist{ + Entries: []playback.PlaylistEntry{ + { + Video: playback.PlaylistFeed{Domain: "domain", UUID: "video-1"}, + Duration: 15 * time.Millisecond, + }, + { + Video: playback.PlaylistFeed{Domain: "domain", UUID: "video-2"}, + }, + }, + } + runtime, err := newPlayerPlaylist(playlist, retry, player) + if err != nil { + t.Fatalf("newPlayerPlaylist() error = %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + sessionResult := make(chan error, 1) + go func() { + sessionResult <- player.Controller.Run( + ctx, + playback.SessionConfig{Retry: retry}, + player.Commands, + ) + }() + playlistResult := make(chan error, 1) + go func() { playlistResult <- runtime.Run(ctx) }() + + if !runtime.Next() { + t.Fatal("Next() = false") + } + first := waitForPlayerSession(t, player.Controller, func(snapshot playback.SessionSnapshot) bool { + return snapshot.Desired.Video.UUID == "video-1" + }) + player.Status.Observe(playback.Status{ + Unit: playback.UnitVideo, + State: playback.StatePlaying, + Generation: first.Generation, + Feed: first.Plan.Video, + }) + + second := waitForPlayerSession(t, player.Controller, func(snapshot playback.SessionSnapshot) bool { + return snapshot.Desired.Video.UUID == "video-2" + }) + if second.Desired.Audio.IsConfigured() { + t.Fatalf("advanced session audio = %#v, want unconfigured", second.Desired.Audio) + } + + cancel() + if err := waitForPlayerRuntimeResult(t, playlistResult); !errors.Is(err, context.Canceled) { + t.Fatalf("playlist Run() error = %v, want %v", err, context.Canceled) + } + if err := waitForPlayerRuntimeResult(t, sessionResult); !errors.Is(err, context.Canceled) { + t.Fatalf("session Run() error = %v, want %v", err, context.Canceled) + } +} + +func newPlaylistTestPlayer(t *testing.T) *playerPlayback { + t.Helper() + return &playerPlayback{ + Controller: newPlaylistTestSessionController(t), + Commands: make(chan playback.SessionCommand, 32), + Status: playback.NewStatusStore(), + } +} + +func newPlaylistTestSessionController(t *testing.T) *playback.SessionController { + t.Helper() + controller, err := playback.NewSessionController( + playlistTestVideoSlot{}, + playlistTestAudioSlot{}, + playlistTestSyncSlot{}, + func(video, audio playback.FeedConfig) bool { return video.Domain == audio.Domain }, + ) + if err != nil { + t.Fatalf("NewSessionController() error = %v", err) + } + return controller +} + +func playlistRuntimeRetry() playback.RetryPolicy { + return playback.RetryPolicy{ + MaxAttempts: 1, + InitialDelay: time.Millisecond, + MaxDelay: time.Millisecond, + } +} + +func waitForPlayerSession( + t *testing.T, + controller *playback.SessionController, + predicate func(playback.SessionSnapshot) bool, +) playback.SessionSnapshot { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if snapshot, ok := controller.Snapshot(); ok && predicate(snapshot) { + return snapshot + } + time.Sleep(time.Millisecond) + } + snapshot, _ := controller.Snapshot() + t.Fatalf("timed out waiting for session snapshot; latest = %#v", snapshot) + return playback.SessionSnapshot{} +} + +func waitForPlayerRuntimeResult(t *testing.T, result <-chan error) error { + t.Helper() + select { + case err := <-result: + return err + case <-time.After(time.Second): + t.Fatal("timed out waiting for runtime result") + return nil + } +}