From b160e3aba2cfff3eaef4636ddc4c914963b7f610 Mon Sep 17 00:00:00 2001 From: Dmitry Sergeev Date: Tue, 1 Sep 2026 20:02:42 +0300 Subject: [PATCH] PlaylistController --- internal/playback/playlist_controller.go | 102 ++++++ internal/playback/playlist_controller_test.go | 316 ++++++++++++++++++ 2 files changed, 418 insertions(+) create mode 100644 internal/playback/playlist_controller.go create mode 100644 internal/playback/playlist_controller_test.go diff --git a/internal/playback/playlist_controller.go b/internal/playback/playlist_controller.go new file mode 100644 index 0000000..5552555 --- /dev/null +++ b/internal/playback/playlist_controller.go @@ -0,0 +1,102 @@ +package playback + +import ( + "context" + "errors" + "sync" +) + +type PlaylistController struct { + playlist Playlist + retry RetryPolicy + sessions chan<- SessionCommand + + mu sync.RWMutex + snapshot PlaylistSnapshot + hasSnapshot bool +} + +type PlaylistSnapshot struct { + State PlaylistState + Entry PlaylistEntry +} + +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, + }, nil +} + +func (c *PlaylistController) Run( + ctx context.Context, + commands <-chan PlaylistCommand, +) error { + state := PlaylistState{} + c.publish(state) + + 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 { + select { + case <-ctx.Done(): + return ctx.Err() + case c.sessions <- sessionCommand: + } + } + + state = next + c.publish(state) + } + } +} + +func (c *PlaylistController) Snapshot() (PlaylistSnapshot, bool) { + c.mu.RLock() + defer c.mu.RUnlock() + return c.snapshot, c.hasSnapshot +} + +func (c *PlaylistController) publish(state PlaylistState) { + entry, _ := state.Entry(c.playlist) + + c.mu.Lock() + c.snapshot = PlaylistSnapshot{State: state, Entry: entry} + c.hasSnapshot = true + c.mu.Unlock() +} diff --git a/internal/playback/playlist_controller_test.go b/internal/playback/playlist_controller_test.go new file mode 100644 index 0000000..d2beba1 --- /dev/null +++ b/internal/playback/playlist_controller_test.go @@ -0,0 +1,316 @@ +package playback + +import ( + "context" + "errors" + "sync" + "testing" + "time" +) + +func TestNewPlaylistControllerValidatesConfiguration(t *testing.T) { + validPlaylist := navigationPlaylist(false) + validRetry := validPlaylistRetryPolicy() + validSessions := make(chan SessionCommand) + tests := []struct { + name string + playlist Playlist + retry RetryPolicy + sessions chan<- SessionCommand + wantErr error + }{ + { + name: "invalid playlist", + playlist: Playlist{Entries: []PlaylistEntry{ + {}, + }}, + retry: validRetry, + sessions: validSessions, + wantErr: ErrPlaylistEntryEmpty, + }, + { + name: "invalid retry", + playlist: validPlaylist, + retry: RetryPolicy{}, + sessions: validSessions, + wantErr: ErrInvalidRetryDelay, + }, + { + name: "nil session commands", + playlist: validPlaylist, + retry: validRetry, + wantErr: ErrNilSessionCommandChannel, + }, + { + name: "valid", + playlist: validPlaylist, + retry: validRetry, + sessions: validSessions, + }, + { + name: "empty playlist is valid", + playlist: Playlist{}, + retry: validRetry, + sessions: validSessions, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + controller, err := NewPlaylistController(test.playlist, test.retry, test.sessions) + if !errors.Is(err, test.wantErr) { + t.Fatalf("NewPlaylistController() error = %v, want %v", err, test.wantErr) + } + if test.wantErr != nil && controller != nil { + t.Fatalf("NewPlaylistController() controller = %#v, want nil", controller) + } + if test.wantErr == nil && controller == nil { + t.Fatal("NewPlaylistController() controller is nil") + } + }) + } +} + +func TestPlaylistControllerPublishesInitialSnapshot(t *testing.T) { + controller, commands, _, cancel, result := startPlaylistController(t, navigationPlaylist(false), 0) + defer cancel() + + snapshot := waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool { + return !snapshot.State.HasSelection + }) + if snapshot.Entry != (PlaylistEntry{}) { + t.Fatalf("initial entry = %#v, want zero value", snapshot.Entry) + } + + close(commands) + if err := waitForPlaylistResult(t, result); err != nil { + t.Fatalf("Run() error = %v", err) + } +} + +func TestPlaylistControllerForwardsManualCommands(t *testing.T) { + controller, commands, sessions, cancel, result := startPlaylistController(t, navigationPlaylist(true), 4) + defer cancel() + + tests := []struct { + command PlaylistCommand + wantIndex int + wantUUID string + }{ + {command: PlaylistCommand{Kind: PlaylistSelect, Index: 1}, wantIndex: 1, wantUUID: "audio-2"}, + {command: PlaylistCommand{Kind: PlaylistNext}, wantIndex: 2, wantUUID: "video-3"}, + {command: PlaylistCommand{Kind: PlaylistPrevious}, wantIndex: 1, wantUUID: "audio-2"}, + {command: PlaylistCommand{Kind: PlaylistSelect, Index: 1}, wantIndex: 1, wantUUID: "audio-2"}, + } + + for _, test := range tests { + commands <- test.command + session := receivePlaylistSession(t, sessions) + if session.Kind != CommandSetSession { + t.Fatalf("session kind = %v, want %v", session.Kind, CommandSetSession) + } + gotUUID := session.Session.Video.UUID + if gotUUID == "" { + gotUUID = session.Session.Audio.UUID + } + if gotUUID != test.wantUUID { + t.Fatalf("session UUID = %q, want %q", gotUUID, test.wantUUID) + } + waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool { + return snapshot.State.HasSelection && snapshot.State.CurrentIndex == test.wantIndex + }) + } + + close(commands) + if err := waitForPlaylistResult(t, result); err != nil { + t.Fatalf("Run() error = %v", err) + } +} + +func TestPlaylistControllerNonLoopingBoundarySendsNothing(t *testing.T) { + _, commands, sessions, cancel, result := startPlaylistController(t, navigationPlaylist(false), 2) + defer cancel() + + commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 2} + _ = receivePlaylistSession(t, sessions) + commands <- PlaylistCommand{Kind: PlaylistNext} + + select { + case command := <-sessions: + t.Fatalf("unexpected session command at boundary: %#v", command) + case <-time.After(20 * time.Millisecond): + } + + close(commands) + if err := waitForPlaylistResult(t, result); err != nil { + t.Fatalf("Run() error = %v", err) + } +} + +func TestPlaylistControllerIgnoresInvalidCommand(t *testing.T) { + _, commands, sessions, cancel, result := startPlaylistController(t, navigationPlaylist(false), 2) + defer cancel() + + commands <- PlaylistCommand{} + commands <- PlaylistCommand{Kind: PlaylistNext} + if session := receivePlaylistSession(t, sessions); session.Session.Video.UUID != "video-1" { + t.Fatalf("session after invalid command = %#v", session) + } + + close(commands) + if err := waitForPlaylistResult(t, result); err != nil { + t.Fatalf("Run() error = %v", err) + } +} + +func TestPlaylistControllerCommitsStateAfterSessionDelivery(t *testing.T) { + sessions := make(chan SessionCommand) + controller, err := NewPlaylistController(navigationPlaylist(false), validPlaylistRetryPolicy(), sessions) + if err != nil { + t.Fatalf("NewPlaylistController() error = %v", err) + } + commands := make(chan PlaylistCommand, 1) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + result := make(chan error, 1) + go func() { result <- controller.Run(ctx, commands) }() + + waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool { + return !snapshot.State.HasSelection + }) + commands <- PlaylistCommand{Kind: PlaylistNext} + time.Sleep(time.Millisecond) + snapshot, ok := controller.Snapshot() + if !ok || snapshot.State.HasSelection { + t.Fatalf("snapshot before delivery = %#v, %v; want no selection", snapshot, ok) + } + + _ = receivePlaylistSession(t, sessions) + waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool { + return snapshot.State.HasSelection && snapshot.State.CurrentIndex == 0 + }) + + close(commands) + if err := waitForPlaylistResult(t, result); err != nil { + t.Fatalf("Run() error = %v", err) + } +} + +func TestPlaylistControllerClosedCommandsReturnsNil(t *testing.T) { + _, commands, _, cancel, result := startPlaylistController(t, navigationPlaylist(false), 0) + defer cancel() + close(commands) + + if err := waitForPlaylistResult(t, result); err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } +} + +func TestPlaylistControllerCancellationWhileReceiving(t *testing.T) { + _, _, _, cancel, result := startPlaylistController(t, navigationPlaylist(false), 0) + cancel() + + if err := waitForPlaylistResult(t, result); !errors.Is(err, context.Canceled) { + t.Fatalf("Run() error = %v, want %v", err, context.Canceled) + } +} + +func TestPlaylistControllerCancellationWhileSending(t *testing.T) { + sessions := make(chan SessionCommand) + controller, err := NewPlaylistController(navigationPlaylist(false), validPlaylistRetryPolicy(), sessions) + if err != nil { + t.Fatalf("NewPlaylistController() error = %v", err) + } + commands := make(chan PlaylistCommand, 1) + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + go func() { result <- controller.Run(ctx, commands) }() + + commands <- PlaylistCommand{Kind: PlaylistNext} + cancel() + if err := waitForPlaylistResult(t, result); !errors.Is(err, context.Canceled) { + t.Fatalf("Run() error = %v, want %v", err, context.Canceled) + } +} + +func TestPlaylistControllerSnapshotConcurrentReads(t *testing.T) { + controller, commands, _, cancel, result := startPlaylistController(t, navigationPlaylist(true), 64) + defer cancel() + + var readers sync.WaitGroup + for range 8 { + readers.Add(1) + go func() { + defer readers.Done() + for range 100 { + _, _ = controller.Snapshot() + } + }() + } + for range 32 { + commands <- PlaylistCommand{Kind: PlaylistNext} + } + readers.Wait() + close(commands) + if err := waitForPlaylistResult(t, result); err != nil { + t.Fatalf("Run() error = %v", err) + } +} + +func startPlaylistController( + t *testing.T, + playlist Playlist, + sessionBuffer int, +) (*PlaylistController, chan PlaylistCommand, chan SessionCommand, context.CancelFunc, <-chan error) { + t.Helper() + sessions := make(chan SessionCommand, sessionBuffer) + controller, err := NewPlaylistController(playlist, validPlaylistRetryPolicy(), sessions) + if err != nil { + t.Fatalf("NewPlaylistController() error = %v", err) + } + commands := make(chan PlaylistCommand, 64) + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + go func() { result <- controller.Run(ctx, commands) }() + return controller, commands, sessions, cancel, result +} + +func waitForPlaylistSnapshot( + t *testing.T, + controller *PlaylistController, + predicate func(PlaylistSnapshot) bool, +) PlaylistSnapshot { + t.Helper() + deadline := time.Now().Add(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 playlist snapshot; latest = %#v", snapshot) + return PlaylistSnapshot{} +} + +func receivePlaylistSession(t *testing.T, sessions <-chan SessionCommand) SessionCommand { + t.Helper() + select { + case command := <-sessions: + return command + case <-time.After(time.Second): + t.Fatal("timed out waiting for session command") + return SessionCommand{} + } +} + +func waitForPlaylistResult(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 playlist controller") + return nil + } +}