Cleanout #4

Merged
itten merged 7 commits from cleanout into main 2026-09-02 00:46:14 +03:00
29 changed files with 970 additions and 153 deletions
+24 -10
View File
@@ -11,8 +11,20 @@ at any time. Synchronization is a runtime relationship between the slots, not a
startup mode.
The design must leave a clean extension point for an `mxlfabrics` reader after
the local MXL player is stable. It must also support a later playlist layer
without moving playlist timing or selection into media readers.
the local MXL player is stable. Playlist timing and selection must remain above
media readers and playback workers.
## Current status
- Stages 0 through 11 are complete.
- The reader-factory extension point from Stage 12 is complete; the
`mxlfabrics` implementation remains future work.
- Stage 13 is complete, including manual/timed navigation, looping,
pause/resume, failure policies, playlist-level retry configuration, and GUI
diagnostics.
- Playlist lifecycle messages are explicit `PlaylistEvent` values produced by
`PlaylistEventCoordinator`; runtime cancellation owns shutdown, so shared
event channels are not closed by either endpoint.
## Required behaviour
@@ -61,7 +73,7 @@ without moving playlist timing or selection into media readers.
- Stopping both members of a synchronized group closes the group atomically.
- Stop commands cancel active reads and retry backoff promptly.
### Future playlist behaviour
### Playlist behaviour
A playlist is an ordered list of playback entries. Each entry describes a
complete desired session state and may contain:
@@ -244,22 +256,24 @@ transition.
4. Start independent workers for both configured slots.
5. Let either worker begin playing without waiting for the other.
## Future playlist model
## Playlist model
The exact public types can be chosen later, but the intended model is:
The implemented model is conceptually:
```go
type PlaylistEntry struct {
Name string
VideoUUID string
AudioUUID string
Video PlaylistFeed // domain + UUID, optional
Audio PlaylistFeed // domain + UUID, optional
SyncRequested bool
Duration time.Duration
}
type Playlist struct {
Entries []PlaylistEntry
Loop bool
Entries []PlaylistEntry
Loop bool
OnFailure PlaylistFailurePolicy // wait or next
Retry *RetryPolicy // optional playlist-wide override
}
```
@@ -505,7 +519,7 @@ Acceptance criteria:
- Fake readers can drive all controller and supervisor tests.
- Local MXL remains the reference implementation.
### Stage 13 — Simple playlist
### Stage 13 — Simple playlist (complete)
Work:
+20
View File
@@ -2,6 +2,7 @@ package main
import (
"mxl-player/internal/playback"
"strings"
"time"
)
@@ -17,6 +18,25 @@ func resolveDomain(shared, override string) string {
return shared
}
func normalizeFeedInput(domain, uuid string) (string, string) {
return strings.TrimSpace(domain), strings.TrimSpace(uuid)
}
func resolveRetryPolicy(
cli playback.RetryPolicy,
cliMaxAttemptsSet bool,
playlist playback.Playlist,
) playback.RetryPolicy {
if playlist.Retry == nil {
return cli
}
resolved := *playlist.Retry
if cliMaxAttemptsSet {
resolved.MaxAttempts = cli.MaxAttempts
}
return resolved
}
func (a appArgs) playbackConfig() playback.SessionConfig {
return playback.SessionConfig{
Video: playback.FeedConfig{
+58 -1
View File
@@ -1,6 +1,11 @@
package main
import "testing"
import (
"testing"
"time"
"mxl-player/internal/playback"
)
func TestAppArgsPlaybackConfig(t *testing.T) {
tests := []struct {
@@ -139,3 +144,55 @@ func TestAppArgsPlaybackConfig(t *testing.T) {
})
}
}
func TestResolveRetryPolicy(t *testing.T) {
cli := playback.RetryPolicy{
MaxAttempts: 0, InitialDelay: time.Second, MaxDelay: 5 * time.Second,
}
fileRetry := playback.RetryPolicy{
MaxAttempts: 4, InitialDelay: 250 * time.Millisecond, MaxDelay: 2 * time.Second,
}
tests := []struct {
name string
playlist playback.Playlist
cliMaxAttemptsSet bool
want playback.RetryPolicy
}{
{name: "no playlist retry uses CLI", want: cli},
{
name: "playlist retry is used by default",
playlist: playback.Playlist{Retry: &fileRetry},
want: fileRetry,
},
{
name: "explicit CLI infinite overrides playlist attempts",
playlist: playback.Playlist{Retry: &fileRetry},
cliMaxAttemptsSet: true,
want: playback.RetryPolicy{
MaxAttempts: 0, InitialDelay: fileRetry.InitialDelay, MaxDelay: fileRetry.MaxDelay,
},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
if got := resolveRetryPolicy(cli, test.cliMaxAttemptsSet, test.playlist); got != test.want {
t.Fatalf("resolveRetryPolicy() = %#v, want %#v", got, test.want)
}
})
}
}
func TestNormalizeFeedInput(t *testing.T) {
domain, uuid := normalizeFeedInput(
" \t/dev/shm/mxl\n",
"\r 5fbec3b1-1b0f-417d-9059-8b94a47197ef \t",
)
if domain != "/dev/shm/mxl" {
t.Fatalf("domain = %q", domain)
}
if uuid != "5fbec3b1-1b0f-417d-9059-8b94a47197ef" {
t.Fatalf("UUID = %q", uuid)
}
}
+24
View File
@@ -0,0 +1,24 @@
package main
import "time"
const guiFrameInterval = time.Second / 60
// videoPollInterval bounds GUI latency when no video producer is waiting.
// It is not a video-rate cap: Next returns immediately whenever a frame
// arrives, including for sources faster than this interval.
const videoPollInterval = 8 * time.Millisecond
func remainingFrameTime(start, now time.Time, interval time.Duration) time.Duration {
remaining := interval - now.Sub(start)
if remaining < 0 {
return 0
}
return remaining
}
func paceFrame(start time.Time) {
if remaining := remainingFrameTime(start, time.Now(), guiFrameInterval); remaining > 0 {
time.Sleep(remaining)
}
}
+29
View File
@@ -0,0 +1,29 @@
package main
import (
"testing"
"time"
)
func TestRemainingFrameTime(t *testing.T) {
start := time.Date(2026, time.September, 2, 0, 0, 0, 0, time.UTC)
interval := 16 * time.Millisecond
tests := []struct {
name string
elapsed time.Duration
want time.Duration
}{
{name: "no work", want: interval},
{name: "partial budget", elapsed: 5 * time.Millisecond, want: 11 * time.Millisecond},
{name: "exact budget", elapsed: interval},
{name: "over budget", elapsed: 20 * time.Millisecond},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
got := remainingFrameTime(start, start.Add(test.elapsed), interval)
if got != test.want {
t.Fatalf("remainingFrameTime() = %v, want %v", got, test.want)
}
})
}
}
+81 -5
View File
@@ -22,7 +22,7 @@ import (
const (
APP_NAME = "MXL Player"
APP_VER = "0.1.0"
APP_VER = "1.0.0"
WIN_WIDTH int32 = 1280
WIN_HEIGHT int32 = 720
)
@@ -176,6 +176,11 @@ func main() {
os.Exit(2)
}
configuredPlaylist = playlist
retryPolicy = resolveRetryPolicy(
retryPolicy,
flagSet.Changed("max-attempts"),
configuredPlaylist,
)
}
if args.VideoDomain == "" {
args.VideoDomain = args.Domain
@@ -372,6 +377,8 @@ func main() {
}
}
doReconnect := func() {
videoDomainStr, videoStr = normalizeFeedInput(videoDomainStr, videoStr)
audioDomainStr, audioStr = normalizeFeedInput(audioDomainStr, audioStr)
videoActive = videoStr != ""
audioActive = audioStr != ""
@@ -465,6 +472,8 @@ func main() {
displayedVideoWidth uint32 = placeholderWidth
displayedVideoHeight uint32 = placeholderHeight
displayedVideoStride uint32 = placeholderStride
displayedVideoSource playback.FeedConfig
hasDisplayedVideo bool
fps float64
dropTracker videoDropTracker
@@ -480,6 +489,8 @@ func main() {
// ImGui
var (
statsWindowWidth float32 = 460
statsWindowHeight float32 = 510
settingWindowWidth float32 = 700
settingsWindowState bool = true
)
@@ -523,6 +534,7 @@ func main() {
if err := r.RecreateSwapchain(); err != nil {
if errors.Is(err, renderer.ErrMinimized) {
resized = true
paceFrame(frameStart)
continue
}
panic(err)
@@ -534,7 +546,7 @@ func main() {
var shownSource playback.FeedConfig
hasFrame := false
frameCtx, frameCancel := context.WithTimeout(ctx, 100*time.Millisecond)
frameCtx, frameCancel := context.WithTimeout(ctx, videoPollInterval)
pendingFrame, frameErr := videoBridge.Next(frameCtx)
frameCancel()
@@ -570,6 +582,16 @@ func main() {
}
snapshot, hasSnapshot := player.Controller.Snapshot()
desiredVideo := desiredVideoFeed(snapshot, hasSnapshot)
if !desiredVideo.Active ||
desiredVideo.Domain != displayedVideoSource.Domain ||
desiredVideo.UUID != displayedVideoSource.UUID {
hasDisplayedVideo = false
}
if hasFrame {
displayedVideoSource = shownSource
hasDisplayedVideo = shouldShowVideo(desiredVideo, shownSource)
}
// stats
if hasFrame {
@@ -592,8 +614,8 @@ func main() {
if r != nil {
gui.BeginFrame(time.Since(lastFrame), int32(r.Extent().Width), int32(r.Extent().Height))
if showStats {
cimgui.SetNextWindowPos(cimgui.Vec2{X: 0, Y: 0})
cimgui.SetNextWindowSize(cimgui.Vec2{X: 460, Y: 510})
cimgui.SetNextWindowPos(cimgui.Vec2{X: float32(r.Extent().Width) - statsWindowWidth, Y: 0})
cimgui.SetNextWindowSize(cimgui.Vec2{X: statsWindowWidth, Y: statsWindowHeight})
cimgui.BeginV("Stats", &showStats, cimgui.WindowFlagsNoResize|cimgui.WindowFlagsNoScrollbar)
mediaStats := player.MediaStats.Snapshot()
cimgui.SeparatorText("Video")
@@ -717,7 +739,13 @@ func main() {
drawFeedsSections := func() {
cimgui.SeparatorText("Video")
cimgui.InputTextWithHint("Video domain", "/dev/shm/mxl", &videoDomainStr, 0, nil)
if cimgui.IsItemDeactivatedAfterEdit() {
videoDomainStr, _ = normalizeFeedInput(videoDomainStr, "")
}
cimgui.InputTextWithHint("Video UUID", "", &videoStr, 0, nil)
if cimgui.IsItemDeactivatedAfterEdit() {
_, videoStr = normalizeFeedInput("", videoStr)
}
if videoActive {
cimgui.SameLine()
if cimgui.Button("Stop##video") {
@@ -742,7 +770,13 @@ func main() {
}
cimgui.SeparatorText("Audio")
cimgui.InputTextWithHint("Audio domain", "/dev/shm/mxl", &audioDomainStr, 0, nil)
if cimgui.IsItemDeactivatedAfterEdit() {
audioDomainStr, _ = normalizeFeedInput(audioDomainStr, "")
}
cimgui.InputTextWithHint("Audio UUID", "", &audioStr, 0, nil)
if cimgui.IsItemDeactivatedAfterEdit() {
_, audioStr = normalizeFeedInput("", audioStr)
}
if audioActive {
cimgui.SameLine()
if cimgui.Button("Stop##audio") {
@@ -811,6 +845,46 @@ func main() {
} else {
cimgui.Text("End behavior: stop")
}
cimgui.Text(fmt.Sprintf(
"Failure behavior: %s",
configuredPlaylist.OnFailure,
))
if retryPolicy.MaxAttempts == 0 {
cimgui.Text("Retries: infinite")
} else {
cimgui.Text(fmt.Sprintf(
"Attempts per entry: %d",
retryPolicy.MaxAttempts,
))
}
cimgui.Text(fmt.Sprintf(
"Retry delay: %s to %s",
retryPolicy.InitialDelay,
retryPolicy.MaxDelay,
))
if hasPlaylistSnapshot && playlistSnapshot.HasFailure {
failure := playlistSnapshot.Failure
name := failure.EntryName
if name == "" {
name = fmt.Sprintf("Entry %d", failure.EntryIndex+1)
}
cimgui.SeparatorText("Last failure")
cimgui.TextWrapped(fmt.Sprintf(
"%s: %s failed",
name,
failure.Status.Unit,
))
cimgui.Text(fmt.Sprintf(
"Attempts: %d | failed attempts: %d",
failure.Status.Attempt,
failure.Status.FailedAttempts,
))
cimgui.Text(fmt.Sprintf("Policy: %s", failure.Policy))
if failure.Status.Err != nil {
cimgui.TextWrapped(failure.Status.Err.Error())
}
}
preview := "No entry selected"
if hasPlaylistSnapshot && playlistSnapshot.State.HasSelection {
@@ -1003,7 +1077,7 @@ func main() {
}
if settingsWindowState {
cimgui.SetNextWindowPos(cimgui.Vec2{X: float32(r.Extent().Width) - settingWindowWidth, Y: 0})
cimgui.SetNextWindowPos(cimgui.Vec2{X: 0, Y: 0})
cimgui.SetNextWindowSize(cimgui.Vec2{X: settingWindowWidth, Y: float32(r.Extent().Height)})
if cimgui.BeginV("Settings & Info", &settingsWindowState, cimgui.WindowFlagsNone) {
drawSettingsContents()
@@ -1017,11 +1091,13 @@ func main() {
displayedVideoWidth,
displayedVideoHeight,
displayedVideoStride,
hasDisplayedVideo,
)
if errors.Is(err, renderer.ErrOutOfDate) {
if rerr := r.RecreateSwapchain(); rerr != nil {
if errors.Is(rerr, renderer.ErrMinimized) {
resized = true
paceFrame(frameStart)
continue
}
panic(rerr)
+45
View File
@@ -14,6 +14,13 @@ type playlistFile struct {
Entries []playlistFileEntry `json:"entries"`
Loop bool `json:"loop"`
OnFailure string `json:"on_failure"`
Retry *playlistFileRetry `json:"retry"`
}
type playlistFileRetry struct {
MaxAttempts *int `json:"max_attempts"`
InitialDelay string `json:"initial_delay"`
MaxDelay string `json:"max_delay"`
}
type playlistFileEntry struct {
@@ -73,6 +80,13 @@ func decodePlaylistFile(reader io.Reader) (playback.Playlist, error) {
"on_failure %q: %w", file.OnFailure, playback.ErrPlaylistFailurePolicy,
)
}
if file.Retry != nil {
retry, err := decodePlaylistRetry(*file.Retry)
if err != nil {
return playback.Playlist{}, err
}
playlist.Retry = &retry
}
for index, entry := range file.Entries {
duration := time.Duration(0)
if entry.Duration != "" {
@@ -103,6 +117,37 @@ func decodePlaylistFile(reader io.Reader) (playback.Playlist, error) {
return playlist, nil
}
func decodePlaylistRetry(file playlistFileRetry) (playback.RetryPolicy, error) {
retry := playback.RetryPolicy{
InitialDelay: initialRetryDelay,
MaxDelay: maxRetryDelay,
}
if file.MaxAttempts != nil {
retry.MaxAttempts = *file.MaxAttempts
}
var err error
if file.InitialDelay != "" {
retry.InitialDelay, err = time.ParseDuration(file.InitialDelay)
if err != nil {
return playback.RetryPolicy{}, fmt.Errorf(
"retry initial_delay %q: %w", file.InitialDelay, err,
)
}
}
if file.MaxDelay != "" {
retry.MaxDelay, err = time.ParseDuration(file.MaxDelay)
if err != nil {
return playback.RetryPolicy{}, fmt.Errorf(
"retry max_delay %q: %w", file.MaxDelay, err,
)
}
}
if err := retry.Validate(); err != nil {
return playback.RetryPolicy{}, fmt.Errorf("validate playlist retry: %w", err)
}
return retry, nil
}
func playlistFileFeedToPlayback(feed *playlistFileFeed) playback.PlaylistFeed {
if feed == nil {
return playback.PlaylistFeed{}
+39
View File
@@ -91,6 +91,41 @@ func TestDecodePlaylistFileAllowsEmptyPlaylist(t *testing.T) {
}
}
func TestDecodePlaylistFileRetry(t *testing.T) {
input := `{
"retry": {
"max_attempts": 3,
"initial_delay": "250ms",
"max_delay": "2s"
},
"entries": []
}`
got, err := decodePlaylistFile(strings.NewReader(input))
if err != nil {
t.Fatalf("decodePlaylistFile() error = %v", err)
}
if got.Retry == nil {
t.Fatal("decodePlaylistFile() retry = nil")
}
want := playback.RetryPolicy{
MaxAttempts: 3, InitialDelay: 250 * time.Millisecond, MaxDelay: 2 * time.Second,
}
if *got.Retry != want {
t.Fatalf("retry = %#v, want %#v", *got.Retry, want)
}
}
func TestDecodePlaylistFileRetryDefaults(t *testing.T) {
got, err := decodePlaylistFile(strings.NewReader(`{"retry":{},"entries":[]}`))
if err != nil {
t.Fatalf("decodePlaylistFile() error = %v", err)
}
if got.Retry == nil || got.Retry.MaxAttempts != 0 ||
got.Retry.InitialDelay != initialRetryDelay || got.Retry.MaxDelay != maxRetryDelay {
t.Fatalf("retry = %#v", got.Retry)
}
}
func TestDecodePlaylistFileRejectsInvalidInput(t *testing.T) {
tests := []struct {
name string
@@ -103,6 +138,10 @@ func TestDecodePlaylistFileRejectsInvalidInput(t *testing.T) {
{name: "unknown field", input: `{"unknown": true}`, wantText: "unknown field"},
{name: "multiple roots", input: `{"entries": []} {"entries": []}`, wantText: "multiple root values"},
{name: "invalid failure policy", input: `{"on_failure":"skip","entries":[]}`, wantErr: playback.ErrPlaylistFailurePolicy},
{name: "negative retry attempts", input: `{"retry":{"max_attempts":-1},"entries":[]}`, wantErr: playback.ErrInvalidMaxAttempts},
{name: "invalid initial delay", input: `{"retry":{"initial_delay":"soon"},"entries":[]}`, wantText: "retry initial_delay"},
{name: "invalid maximum delay", input: `{"retry":{"max_delay":"later"},"entries":[]}`, wantText: "retry max_delay"},
{name: "invalid retry range", input: `{"retry":{"initial_delay":"2s","max_delay":"1s"},"entries":[]}`, wantErr: playback.ErrInvalidRetryRange},
{
name: "invalid duration",
input: `{"entries":[{"video":{"domain":"/video","uuid":"video"},"duration":"later"}]}`,
+19 -15
View File
@@ -9,7 +9,7 @@ import (
"mxl-player/internal/playback"
)
const playlistReadinessInterval = 10 * time.Millisecond
const playlistEventInterval = 10 * time.Millisecond
var (
ErrPlayerPlaybackRequired = errors.New("player playback is required")
@@ -18,10 +18,14 @@ var (
)
type playerPlaylist struct {
Controller *playback.PlaylistController
Coordinator *playback.PlaylistReadinessCoordinator
Commands chan playback.PlaylistCommand
Readiness chan playback.PlaylistReadiness
Controller *playback.PlaylistController
// commands is written by GUI-facing methods and consumed by Controller.
// events is written by coordinator and consumed by Controller. Run owns
// both goroutine lifecycles; cancellation replaces channel closing.
coordinator *playback.PlaylistEventCoordinator
commands chan playback.PlaylistCommand
events chan playback.PlaylistEvent
}
func newPlayerPlaylist(
@@ -40,7 +44,7 @@ func newPlayerPlaylist(
}
commands := make(chan playback.PlaylistCommand, 32)
readiness := make(chan playback.PlaylistReadiness, 8)
events := make(chan playback.PlaylistEvent, 8)
controller, err := playback.NewPlaylistController(
playlist,
retry,
@@ -49,12 +53,12 @@ func newPlayerPlaylist(
if err != nil {
return nil, err
}
coordinator, err := playback.NewPlaylistReadinessCoordinator(
coordinator, err := playback.NewPlaylistEventCoordinator(
controller,
player.Controller,
player.Status,
readiness,
playlistReadinessInterval,
events,
playlistEventInterval,
)
if err != nil {
return nil, err
@@ -62,9 +66,9 @@ func newPlayerPlaylist(
return &playerPlaylist{
Controller: controller,
Coordinator: coordinator,
Commands: commands,
Readiness: readiness,
coordinator: coordinator,
commands: commands,
events: events,
}, nil
}
@@ -74,10 +78,10 @@ func (p *playerPlaylist) Run(ctx context.Context) error {
results := make(chan error, 2)
go func() {
results <- p.Controller.Run(runCtx, p.Commands, p.Readiness)
results <- p.Controller.Run(runCtx, p.commands, p.events)
}()
go func() {
results <- p.Coordinator.Run(runCtx)
results <- p.coordinator.Run(runCtx)
}()
first := <-results
@@ -121,7 +125,7 @@ func (p *playerPlaylist) Resume() bool {
func (p *playerPlaylist) enqueue(command playback.PlaylistCommand) bool {
select {
case p.Commands <- command:
case p.commands <- command:
return true
default:
return false
+4 -4
View File
@@ -104,10 +104,10 @@ func TestNewPlayerPlaylistWiresComponents(t *testing.T) {
if err != nil {
t.Fatalf("newPlayerPlaylist() error = %v", err)
}
if runtime.Controller == nil || runtime.Coordinator == nil {
if runtime.Controller == nil || runtime.coordinator == nil {
t.Fatalf("runtime components = %#v", runtime)
}
if runtime.Commands == nil || runtime.Readiness == nil {
if runtime.commands == nil || runtime.events == nil {
t.Fatalf("runtime channels = %#v", runtime)
}
}
@@ -135,7 +135,7 @@ func TestPlayerPlaylistNavigationHelpers(t *testing.T) {
if !test.send() {
t.Fatal("navigation helper returned false")
}
if got := <-runtime.Commands; got != test.want {
if got := <-runtime.commands; got != test.want {
t.Fatalf("navigation command = %#v, want %#v", got, test.want)
}
}
@@ -150,7 +150,7 @@ func TestPlayerPlaylistNavigationQueueFull(t *testing.T) {
if err != nil {
t.Fatalf("newPlayerPlaylist() error = %v", err)
}
for range cap(runtime.Commands) {
for range cap(runtime.commands) {
if !runtime.Next() {
t.Fatal("queue filled before reaching capacity")
}
+30
View File
@@ -0,0 +1,30 @@
package main
import "mxl-player/internal/playback"
func desiredVideoFeed(
snapshot playback.SessionSnapshot,
available bool,
) playback.FeedConfig {
if !available {
return playback.FeedConfig{}
}
switch snapshot.Plan.Topology {
case playback.TopologyIndependent:
if snapshot.Plan.Video.Active {
return snapshot.Plan.Video
}
case playback.TopologySynchronized:
if snapshot.Plan.Sync.Active() {
return snapshot.Plan.Sync.Video
}
}
return playback.FeedConfig{}
}
func shouldShowVideo(
desired playback.FeedConfig,
delivered playback.FeedConfig,
) bool {
return desired.Active && sameVideoSource(desired, delivered)
}
+46
View File
@@ -0,0 +1,46 @@
package main
import (
"testing"
"mxl-player/internal/playback"
)
func TestDesiredVideoFeed(t *testing.T) {
video := playback.FeedConfig{Domain: "/video", UUID: "video", Active: true}
tests := []struct {
name string
available bool
plan playback.SessionPlan
want playback.FeedConfig
}{
{name: "snapshot unavailable"},
{name: "idle", available: true, plan: playback.SessionPlan{Topology: playback.TopologyIdle}},
{name: "audio only", available: true, plan: playback.SessionPlan{Topology: playback.TopologyIndependent}},
{name: "independent video", available: true, plan: playback.SessionPlan{Topology: playback.TopologyIndependent, Video: video}, want: video},
{name: "synchronized video", available: true, plan: playback.SessionPlan{Topology: playback.TopologySynchronized, Sync: playback.SyncPairConfig{Video: video, Audio: playback.FeedConfig{Active: true}}}, want: video},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
got := desiredVideoFeed(playback.SessionSnapshot{Plan: test.plan}, test.available)
if got != test.want {
t.Fatalf("desiredVideoFeed() = %#v, want %#v", got, test.want)
}
})
}
}
func TestShouldShowVideoRequiresDesiredSource(t *testing.T) {
desired := playback.FeedConfig{Domain: "/video", UUID: "video", Active: true}
if !shouldShowVideo(desired, desired) {
t.Fatal("matching active video was hidden")
}
if shouldShowVideo(playback.FeedConfig{}, desired) {
t.Fatal("video was shown without an active desired feed")
}
other := desired
other.UUID = "other"
if shouldShowVideo(desired, other) {
t.Fatal("frame from old source was shown")
}
}
+8
View File
@@ -0,0 +1,8 @@
5fbec3b1-1b0f-417d-9059-8b94a47197ed
5fbec3b1-1b0f-417d-9059-8b94a47197ec
5fbec3b1-1b0f-417d-9059-8b94a47197ef
5fbec3b1-1b0f-417d-9059-8b94a47197eb
2618979d-76a5-45e0-83cb-0f192978d1cd
9d2a041b-01cf-4ee4-bffa-188fe093c99b
+3 -2
View File
@@ -4,11 +4,12 @@ Size=400,400
Collapsed=0
[Window][Settings & Info]
Pos=580,0
Size=700,720
Pos=0,0
Size=700,1080
Collapsed=0
[Window][Stats]
Pos=1460,0
Size=460,510
Collapsed=0
+13
View File
@@ -3,6 +3,8 @@ package imgui
import (
"time"
"mxl-player/internal/sdl"
cimgui "github.com/AllenDang/cimgui-go/imgui"
)
@@ -17,9 +19,20 @@ func New() *Context {
ctx := cimgui.CreateContext()
cimgui.SetCurrentContext(ctx)
io := cimgui.CurrentIO()
cimgui.CurrentPlatformIO().SetClipboardHandler(sdlClipboardHandler{})
return &Context{ctx: ctx, io: io}
}
type sdlClipboardHandler struct{}
func (sdlClipboardHandler) GetClipboard() string {
return sdl.GetClipboardText()
}
func (sdlClipboardHandler) SetClipboard(text string) {
sdl.SetClipboardText(text)
}
func (c *Context) Destroy() {
cimgui.DestroyContext()
}
+20
View File
@@ -7,6 +7,13 @@ import (
cimgui "github.com/AllenDang/cimgui-go/imgui"
)
const (
sdlKModShift uint16 = 0x0001 | 0x0002
sdlKModCtrl uint16 = 0x0040 | 0x0080
sdlKModAlt uint16 = 0x0100 | 0x0200
sdlKModGUI uint16 = 0x0400 | 0x0800
)
// SDL3 event (128 byte raw buffer) -> imgui
func (c *Context) ProcessEvent(event *[128]byte) {
eventType := *(*uint32)(unsafe.Pointer(&event[0]))
@@ -15,7 +22,9 @@ func (c *Context) ProcessEvent(event *[128]byte) {
switch eventType {
case sdl.EventKeyDown, sdl.EventKeyUp:
scancode := *(*uint32)(unsafe.Pointer(&event[24]))
modifiers := *(*uint16)(unsafe.Pointer(&event[32]))
down := eventType == sdl.EventKeyDown
c.addKeyModifiers(modifiers)
key := sdlScancodeToImGuiKey(scancode)
if key >= 0 {
c.io.AddKeyEvent(key, down)
@@ -47,6 +56,13 @@ func (c *Context) ProcessEvent(event *[128]byte) {
}
}
func (c *Context) addKeyModifiers(modifiers uint16) {
c.io.AddKeyEvent(cimgui.ModCtrl, modifiers&sdlKModCtrl != 0)
c.io.AddKeyEvent(cimgui.ModShift, modifiers&sdlKModShift != 0)
c.io.AddKeyEvent(cimgui.ModAlt, modifiers&sdlKModAlt != 0)
c.io.AddKeyEvent(cimgui.ModSuper, modifiers&sdlKModGUI != 0)
}
func sdlScancodeToImGuiKey(scancode uint32) cimgui.Key {
switch scancode {
case 40: // SDL_SCANCODE_RETURN
@@ -79,6 +95,10 @@ func sdlScancodeToImGuiKey(scancode uint32) cimgui.Key {
return cimgui.KeyLeftAlt
case 230: // SDL_SCANCODE_RALT
return cimgui.KeyRightAlt
case 227: // SDL_SCANCODE_LGUI
return cimgui.KeyLeftSuper
case 231: // SDL_SCANCODE_RGUI
return cimgui.KeyRightSuper
default:
// Letters A-Z: scancodes 4-29 map to ImGui KeyA-KeyZ
if scancode >= 4 && scancode <= 29 {
+6
View File
@@ -56,6 +56,7 @@ type Playlist struct {
Entries []PlaylistEntry
Loop bool
OnFailure PlaylistFailurePolicy
Retry *RetryPolicy
}
func (f PlaylistFeed) IsConfigured() bool {
@@ -95,6 +96,11 @@ func (p Playlist) Validate() error {
if err := p.OnFailure.Validate(); err != nil {
return err
}
if p.Retry != nil {
if err := p.Retry.Validate(); err != nil {
return fmt.Errorf("playlist retry: %w", err)
}
}
for index, entry := range p.Entries {
if err := entry.Validate(); err != nil {
return fmt.Errorf("playlist entry %d: %w", index, err)
+52 -19
View File
@@ -17,12 +17,9 @@ const (
type PlaylistEvent struct {
Revision uint64
Kind PlaylistEventKind
Failure Status
}
// PlaylistReadiness is retained as an alias for callers that only publish
// ready events. Its zero Kind is PlaylistEventReady.
type PlaylistReadiness = PlaylistEvent
type playlistTimer interface {
C() <-chan time.Time
Stop() bool
@@ -47,13 +44,24 @@ type PlaylistController struct {
mu sync.RWMutex
snapshot PlaylistSnapshot
hasSnapshot bool
failure PlaylistFailure
}
type PlaylistFailure struct {
EntryIndex int
EntryName string
Revision uint64
Policy PlaylistFailurePolicy
Status Status
}
type PlaylistSnapshot struct {
State PlaylistState
Entry PlaylistEntry
Revision uint64
Timing PlaylistTimingState
State PlaylistState
Entry PlaylistEntry
Revision uint64
Timing PlaylistTimingState
Failure PlaylistFailure
HasFailure bool
}
var (
@@ -88,7 +96,7 @@ func NewPlaylistController(
func (c *PlaylistController) Run(
ctx context.Context,
commands <-chan PlaylistCommand,
readiness <-chan PlaylistReadiness,
events <-chan PlaylistEvent,
) error {
state := PlaylistState{}
revision := uint64(0)
@@ -155,6 +163,7 @@ func (c *PlaylistController) Run(
continue
}
if apply {
c.clearFailure()
stopTimer()
select {
case <-ctx.Done():
@@ -169,16 +178,23 @@ func (c *PlaylistController) Run(
state = next
c.publish(state, revision, timing)
case ready, ok := <-readiness:
case event, ok := <-events:
if !ok {
readiness = nil
events = nil
continue
}
if ready.Kind == PlaylistEventFailed {
if ready.Revision != revision {
if event.Kind == PlaylistEventFailed {
if event.Revision != revision {
continue
}
stopTimer()
c.setFailure(PlaylistFailure{
EntryIndex: state.CurrentIndex,
EntryName: c.playlist.Entries[state.CurrentIndex].Name,
Revision: revision,
Policy: c.playlist.OnFailure,
Status: event.Failure,
})
// A failed entry must not retain a live or apparently active
// duration clock, even when the policy is to wait.
timing = NewPlaylistTiming(revision, timing.Duration)
@@ -208,8 +224,11 @@ func (c *PlaylistController) Run(
c.publish(state, revision, timing)
continue
}
if event.Kind != PlaylistEventReady {
continue
}
if timing.Paused &&
ready.Revision == timing.Revision &&
event.Revision == timing.Revision &&
timing.Duration > 0 &&
!timing.Expired {
timing.Ready = true
@@ -218,7 +237,7 @@ func (c *PlaylistController) Run(
}
nextTiming, started := StartPlaylistTiming(
timing,
ready.Revision,
event.Revision,
c.now(),
)
if !started {
@@ -285,15 +304,29 @@ func (c *PlaylistController) publish(
c.mu.Lock()
c.snapshot = PlaylistSnapshot{
State: state,
Entry: entry,
Revision: revision,
Timing: timing,
State: state,
Entry: entry,
Revision: revision,
Timing: timing,
Failure: c.failure,
HasFailure: c.failure.Revision != 0,
}
c.hasSnapshot = true
c.mu.Unlock()
}
func (c *PlaylistController) setFailure(failure PlaylistFailure) {
c.mu.Lock()
c.failure = failure
c.mu.Unlock()
}
func (c *PlaylistController) clearFailure() {
c.mu.Lock()
c.failure = PlaylistFailure{}
c.mu.Unlock()
}
func stopPlaylistTimer(timer playlistTimer) {
if timer == nil || timer.Stop() {
return
@@ -68,9 +68,11 @@ func TestPlaylistControllerStartsOneTimerForMatchingReadiness(t *testing.T) {
return snapshot.Revision == 1
})
readiness <- PlaylistReadiness{Revision: 0}
readiness <- PlaylistEvent{Revision: 0}
assertNoPlaylistTimer(t, timers)
readiness <- PlaylistReadiness{Revision: 1}
readiness <- PlaylistEvent{Revision: 1, Kind: PlaylistEventKind(99)}
assertNoPlaylistTimer(t, timers)
readiness <- PlaylistEvent{Revision: 1}
timer := receiveFakePlaylistTimer(t, timers)
snapshot := waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool {
@@ -79,7 +81,7 @@ func TestPlaylistControllerStartsOneTimerForMatchingReadiness(t *testing.T) {
if snapshot.Timing.Deadline != now.Add(10*time.Second) {
t.Fatalf("deadline = %v, want %v", snapshot.Timing.Deadline, now.Add(10*time.Second))
}
readiness <- PlaylistReadiness{Revision: 1}
readiness <- PlaylistEvent{Revision: 1}
assertNoPlaylistTimer(t, timers)
if timer.isStopped() {
t.Fatal("timer stopped after duplicate readiness")
@@ -101,7 +103,7 @@ func TestPlaylistControllerTimerExpiryAdvances(t *testing.T) {
commands <- PlaylistCommand{Kind: PlaylistNext}
_ = receivePlaylistSession(t, sessions)
readiness <- PlaylistReadiness{Revision: 1}
readiness <- PlaylistEvent{Revision: 1}
timer := receiveFakePlaylistTimer(t, timers)
timer.fire(now.Add(10 * time.Second))
@@ -132,7 +134,7 @@ func TestPlaylistControllerLoopingTimerExpiryWraps(t *testing.T) {
commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 1}
_ = receivePlaylistSession(t, sessions)
readiness <- PlaylistReadiness{Revision: 1}
readiness <- PlaylistEvent{Revision: 1}
timer := receiveFakePlaylistTimer(t, timers)
waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool {
return snapshot.Timing.Started
@@ -160,7 +162,7 @@ func TestPlaylistControllerFinalExpiryStopsWithoutCommand(t *testing.T) {
commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 1}
_ = receivePlaylistSession(t, sessions)
readiness <- PlaylistReadiness{Revision: 1}
readiness <- PlaylistEvent{Revision: 1}
timer := receiveFakePlaylistTimer(t, timers)
waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool {
return snapshot.Timing.Started
@@ -192,7 +194,7 @@ func TestPlaylistControllerManualSelectionStopsOldTimer(t *testing.T) {
commands <- PlaylistCommand{Kind: PlaylistNext}
_ = receivePlaylistSession(t, sessions)
readiness <- PlaylistReadiness{Revision: 1}
readiness <- PlaylistEvent{Revision: 1}
oldTimer := receiveFakePlaylistTimer(t, timers)
commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 1}
@@ -229,7 +231,7 @@ func TestPlaylistControllerZeroDurationDoesNotCreateTimer(t *testing.T) {
commands <- PlaylistCommand{Kind: PlaylistNext}
_ = receivePlaylistSession(t, sessions)
readiness <- PlaylistReadiness{Revision: 1}
readiness <- PlaylistEvent{Revision: 1}
assertNoPlaylistTimer(t, timers)
close(commands)
@@ -244,7 +246,7 @@ func TestPlaylistControllerCancellationStopsTimer(t *testing.T) {
commands <- PlaylistCommand{Kind: PlaylistNext}
_ = receivePlaylistSession(t, sessions)
readiness <- PlaylistReadiness{Revision: 1}
readiness <- PlaylistEvent{Revision: 1}
timer := receiveFakePlaylistTimer(t, timers)
cancel()
if err := waitForPlaylistResult(t, result); !errors.Is(err, context.Canceled) {
@@ -262,7 +264,7 @@ func TestPlaylistControllerPauseAndResumeTimer(t *testing.T) {
commands <- PlaylistCommand{Kind: PlaylistNext}
_ = receivePlaylistSession(t, sessions)
readiness <- PlaylistReadiness{Revision: 1}
readiness <- PlaylistEvent{Revision: 1}
oldTimer := receiveFakePlaylistTimer(t, timers)
waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool {
return snapshot.Timing.Started
@@ -344,7 +346,7 @@ func TestPlaylistControllerRecordsQueuedReadinessWhilePaused(t *testing.T) {
return snapshot.Timing.Paused
})
readiness <- PlaylistReadiness{Revision: 1}
readiness <- PlaylistEvent{Revision: 1}
waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool {
return snapshot.Timing.Paused && snapshot.Timing.Ready
})
@@ -368,7 +370,7 @@ func startTimedPlaylistController(
) (
*PlaylistController,
chan PlaylistCommand,
chan PlaylistReadiness,
chan PlaylistEvent,
chan SessionCommand,
chan *fakePlaylistTimer,
time.Time,
@@ -390,7 +392,7 @@ func startTimedPlaylistController(
return timer
}
commands := make(chan PlaylistCommand, 16)
readiness := make(chan PlaylistReadiness, 16)
readiness := make(chan PlaylistEvent, 16)
ctx, cancel := context.WithCancel(context.Background())
result := make(chan error, 1)
go func() { result <- controller.Run(ctx, commands, readiness) }()
@@ -18,45 +18,45 @@ type PlaybackStatusSnapshotSource interface {
SnapshotAll() PlaybackStatusSnapshot
}
type playlistReadinessTicker interface {
type playlistEventTicker interface {
C() <-chan time.Time
Stop()
}
type playlistReadinessTickerFactory func(time.Duration) playlistReadinessTicker
type playlistEventTickerFactory func(time.Duration) playlistEventTicker
type realPlaylistReadinessTicker struct {
type realPlaylistEventTicker struct {
ticker *time.Ticker
}
func (t realPlaylistReadinessTicker) C() <-chan time.Time { return t.ticker.C }
func (t realPlaylistReadinessTicker) Stop() { t.ticker.Stop() }
func (t realPlaylistEventTicker) C() <-chan time.Time { return t.ticker.C }
func (t realPlaylistEventTicker) 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")
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")
ErrPlaylistEventOutputRequired = errors.New("playlist event output channel is required")
ErrPlaylistEventInterval = errors.New("playlist event interval must be positive")
)
type PlaylistReadinessCoordinator struct {
type PlaylistEventCoordinator struct {
playlist PlaylistSnapshotSource
session SessionSnapshotSource
statuses PlaybackStatusSnapshotSource
output chan<- PlaylistReadiness
output chan<- PlaylistEvent
interval time.Duration
newTicker playlistReadinessTickerFactory
newTicker playlistEventTickerFactory
}
func NewPlaylistReadinessCoordinator(
func NewPlaylistEventCoordinator(
playlist PlaylistSnapshotSource,
session SessionSnapshotSource,
statuses PlaybackStatusSnapshotSource,
output chan<- PlaylistReadiness,
output chan<- PlaylistEvent,
interval time.Duration,
) (*PlaylistReadinessCoordinator, error) {
) (*PlaylistEventCoordinator, error) {
if playlist == nil {
return nil, ErrPlaylistSnapshotSourceRequired
}
@@ -67,25 +67,25 @@ func NewPlaylistReadinessCoordinator(
return nil, ErrStatusSnapshotSourceRequired
}
if output == nil {
return nil, ErrPlaylistReadinessOutputRequired
return nil, ErrPlaylistEventOutputRequired
}
if interval <= 0 {
return nil, ErrPlaylistReadinessInterval
return nil, ErrPlaylistEventInterval
}
return &PlaylistReadinessCoordinator{
return &PlaylistEventCoordinator{
playlist: playlist,
session: session,
statuses: statuses,
output: output,
interval: interval,
newTicker: func(interval time.Duration) playlistReadinessTicker {
return realPlaylistReadinessTicker{ticker: time.NewTicker(interval)}
newTicker: func(interval time.Duration) playlistEventTicker {
return realPlaylistEventTicker{ticker: time.NewTicker(interval)}
},
}, nil
}
func (c *PlaylistReadinessCoordinator) Run(ctx context.Context) error {
func (c *PlaylistEventCoordinator) Run(ctx context.Context) error {
ticker := c.newTicker(c.interval)
defer ticker.Stop()
@@ -112,13 +112,14 @@ func (c *PlaylistReadinessCoordinator) Run(ctx context.Context) error {
continue
}
statuses := c.statuses.SnapshotAll()
if IsSessionFailed(sessionSnapshot, statuses) {
if failure, failed := SessionFailureStatus(sessionSnapshot, statuses); failed {
if playlistSnapshot.Revision == emittedFailedRevision {
continue
}
failed := PlaylistEvent{
Revision: playlistSnapshot.Revision,
Kind: PlaylistEventFailed,
Failure: failure,
}
select {
case <-ctx.Done():
@@ -138,7 +139,7 @@ func (c *PlaylistReadinessCoordinator) Run(ctx context.Context) error {
continue
}
ready := PlaylistReadiness{Revision: playlistSnapshot.Revision}
ready := PlaylistEvent{Revision: playlistSnapshot.Revision}
select {
case <-ctx.Done():
return ctx.Err()
@@ -217,24 +218,39 @@ func IsSessionFailed(
session SessionSnapshot,
statuses PlaybackStatusSnapshot,
) bool {
_, failed := SessionFailureStatus(session, statuses)
return failed
}
func SessionFailureStatus(
session SessionSnapshot,
statuses PlaybackStatusSnapshot,
) (Status, bool) {
if statuses.Generation != session.Generation {
return false
return Status{}, false
}
switch session.Plan.Topology {
case TopologyIndependent:
return (session.Plan.Video.Active && statusIsFailed(
if session.Plan.Video.Active && statusIsFailed(
statuses.Video, statuses.HasVideo, session.Generation, session.Plan.Video,
)) || (session.Plan.Audio.Active && statusIsFailed(
) {
return statuses.Video, true
}
if session.Plan.Audio.Active && statusIsFailed(
statuses.Audio, statuses.HasAudio, session.Generation, session.Plan.Audio,
))
) {
return statuses.Audio, true
}
return Status{}, false
case TopologySynchronized:
return statuses.HasSync &&
failed := statuses.HasSync &&
statuses.Sync.Generation == session.Generation &&
statuses.Sync.State == StateFailed &&
sameSyncSource(statuses.Sync.Pair, session.Plan.Sync)
return statuses.Sync, failed
default:
return false
return Status{}, false
}
}
@@ -63,55 +63,55 @@ func (s *fakePlaybackStatusSnapshotSource) set(snapshot PlaybackStatusSnapshot)
s.mu.Unlock()
}
type fakePlaylistReadinessTicker struct {
type fakePlaylistEventTicker struct {
ch chan time.Time
mu sync.Mutex
stopped bool
}
func newFakePlaylistReadinessTicker() *fakePlaylistReadinessTicker {
return &fakePlaylistReadinessTicker{ch: make(chan time.Time, 16)}
func newFakePlaylistEventTicker() *fakePlaylistEventTicker {
return &fakePlaylistEventTicker{ch: make(chan time.Time, 16)}
}
func (t *fakePlaylistReadinessTicker) C() <-chan time.Time { return t.ch }
func (t *fakePlaylistReadinessTicker) Stop() {
func (t *fakePlaylistEventTicker) C() <-chan time.Time { return t.ch }
func (t *fakePlaylistEventTicker) Stop() {
t.mu.Lock()
t.stopped = true
t.mu.Unlock()
}
func (t *fakePlaylistReadinessTicker) tick() { t.ch <- time.Now() }
func (t *fakePlaylistReadinessTicker) isStopped() bool {
func (t *fakePlaylistEventTicker) tick() { t.ch <- time.Now() }
func (t *fakePlaylistEventTicker) isStopped() bool {
t.mu.Lock()
defer t.mu.Unlock()
return t.stopped
}
func TestNewPlaylistReadinessCoordinatorValidatesDependencies(t *testing.T) {
func TestNewPlaylistEventCoordinatorValidatesDependencies(t *testing.T) {
playlist := &fakePlaylistSnapshotSource{}
session := &fakeSessionSnapshotSource{}
statuses := &fakePlaybackStatusSnapshotSource{}
output := make(chan PlaylistReadiness)
output := make(chan PlaylistEvent)
tests := []struct {
name string
playlist PlaylistSnapshotSource
session SessionSnapshotSource
statuses PlaybackStatusSnapshotSource
output chan<- PlaylistReadiness
output chan<- PlaylistEvent
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: "output", playlist: playlist, session: session, statuses: statuses, interval: time.Millisecond, wantErr: ErrPlaylistEventOutputRequired},
{name: "interval", playlist: playlist, session: session, statuses: statuses, output: output, wantErr: ErrPlaylistEventInterval},
{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(
coordinator, err := NewPlaylistEventCoordinator(
test.playlist,
test.session,
test.statuses,
@@ -119,7 +119,7 @@ func TestNewPlaylistReadinessCoordinatorValidatesDependencies(t *testing.T) {
test.interval,
)
if !errors.Is(err, test.wantErr) {
t.Fatalf("NewPlaylistReadinessCoordinator() error = %v, want %v", err, test.wantErr)
t.Fatalf("NewPlaylistEventCoordinator() error = %v, want %v", err, test.wantErr)
}
if test.wantErr != nil && coordinator != nil {
t.Fatalf("coordinator = %#v, want nil", coordinator)
@@ -174,10 +174,10 @@ func TestPlaylistEntryMatchesSession(t *testing.T) {
}
}
func TestPlaylistReadinessCoordinatorEmitsOncePerRevision(t *testing.T) {
func TestPlaylistEventCoordinatorEmitsOncePerRevision(t *testing.T) {
playlist, session, statuses := readyVideoSnapshots(1)
output := make(chan PlaylistReadiness, 4)
coordinator, ticker, cancel, result := startReadinessCoordinator(
output := make(chan PlaylistEvent, 4)
coordinator, ticker, cancel, result := startEventCoordinator(
t,
playlist,
session,
@@ -188,16 +188,16 @@ func TestPlaylistReadinessCoordinatorEmitsOncePerRevision(t *testing.T) {
defer cancel()
ticker.tick()
if got := receivePlaylistReadiness(t, output); got.Revision != 1 {
if got := receivePlaylistEvent(t, output); got.Revision != 1 {
t.Fatalf("readiness revision = %d, want 1", got.Revision)
}
ticker.tick()
assertNoPlaylistReadiness(t, output)
assertNoPlaylistEvent(t, output)
next := playlistSnapshotForVideo(2)
playlist.set(next, true)
ticker.tick()
if got := receivePlaylistReadiness(t, output); got.Revision != 2 {
if got := receivePlaylistEvent(t, output); got.Revision != 2 {
t.Fatalf("readiness revision = %d, want 2", got.Revision)
}
@@ -210,10 +210,10 @@ func TestPlaylistReadinessCoordinatorEmitsOncePerRevision(t *testing.T) {
}
}
func TestPlaylistReadinessCoordinatorWaitsForAllConditions(t *testing.T) {
func TestPlaylistEventCoordinatorWaitsForAllConditions(t *testing.T) {
playlist, session, statuses := readyVideoSnapshots(1)
output := make(chan PlaylistReadiness, 1)
_, ticker, cancel, result := startReadinessCoordinator(t, playlist, session, statuses, output)
output := make(chan PlaylistEvent, 1)
_, ticker, cancel, result := startEventCoordinator(t, playlist, session, statuses, output)
defer cancel()
tests := []struct {
@@ -256,7 +256,7 @@ func TestPlaylistReadinessCoordinatorWaitsForAllConditions(t *testing.T) {
statuses.set(validStatuses.snapshot)
test.mutate()
ticker.tick()
assertNoPlaylistReadiness(t, output)
assertNoPlaylistEvent(t, output)
})
}
@@ -264,10 +264,10 @@ func TestPlaylistReadinessCoordinatorWaitsForAllConditions(t *testing.T) {
_ = waitForPlaylistResult(t, result)
}
func TestPlaylistReadinessCoordinatorCancellationWhileBlockedSending(t *testing.T) {
func TestPlaylistEventCoordinatorCancellationWhileBlockedSending(t *testing.T) {
playlist, session, statuses := readyVideoSnapshots(1)
output := make(chan PlaylistReadiness)
_, ticker, cancel, result := startReadinessCoordinator(t, playlist, session, statuses, output)
output := make(chan PlaylistEvent)
_, ticker, cancel, result := startEventCoordinator(t, playlist, session, statuses, output)
ticker.tick()
time.Sleep(time.Millisecond)
@@ -322,15 +322,15 @@ func playlistSnapshotForVideo(revision uint64) PlaylistSnapshot {
}
}
func startReadinessCoordinator(
func startEventCoordinator(
t *testing.T,
playlist PlaylistSnapshotSource,
session SessionSnapshotSource,
statuses PlaybackStatusSnapshotSource,
output chan<- PlaylistReadiness,
) (*PlaylistReadinessCoordinator, *fakePlaylistReadinessTicker, context.CancelFunc, <-chan error) {
output chan<- PlaylistEvent,
) (*PlaylistEventCoordinator, *fakePlaylistEventTicker, context.CancelFunc, <-chan error) {
t.Helper()
coordinator, err := NewPlaylistReadinessCoordinator(
coordinator, err := NewPlaylistEventCoordinator(
playlist,
session,
statuses,
@@ -338,28 +338,28 @@ func startReadinessCoordinator(
time.Millisecond,
)
if err != nil {
t.Fatalf("NewPlaylistReadinessCoordinator() error = %v", err)
t.Fatalf("NewPlaylistEventCoordinator() error = %v", err)
}
ticker := newFakePlaylistReadinessTicker()
coordinator.newTicker = func(time.Duration) playlistReadinessTicker { return ticker }
ticker := newFakePlaylistEventTicker()
coordinator.newTicker = func(time.Duration) playlistEventTicker { 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 {
func receivePlaylistEvent(t *testing.T, output <-chan PlaylistEvent) PlaylistEvent {
t.Helper()
select {
case readiness := <-output:
return readiness
case <-time.After(time.Second):
t.Fatal("timed out waiting for playlist readiness")
return PlaylistReadiness{}
return PlaylistEvent{}
}
}
func assertNoPlaylistReadiness(t *testing.T, output <-chan PlaylistReadiness) {
func assertNoPlaylistEvent(t *testing.T, output <-chan PlaylistEvent) {
t.Helper()
select {
case readiness := <-output:
+32 -3
View File
@@ -2,6 +2,7 @@ package playback
import (
"context"
"errors"
"testing"
"time"
)
@@ -145,14 +146,22 @@ func TestPlaylistControllerFailurePolicy(t *testing.T) {
t.Fatalf("NewPlaylistController() error = %v", err)
}
commands := make(chan PlaylistCommand, 2)
events := make(chan PlaylistReadiness, 2)
events := make(chan PlaylistEvent, 2)
ctx, cancel := context.WithCancel(context.Background())
result := make(chan error, 1)
go func() { result <- controller.Run(ctx, commands, events) }()
commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 0}
<-sessions
events <- PlaylistEvent{Revision: 1, Kind: PlaylistEventFailed}
failureErr := errors.New("flow unavailable")
events <- PlaylistEvent{
Revision: 1,
Kind: PlaylistEventFailed,
Failure: Status{
Unit: UnitVideo, State: StateFailed,
Attempt: 2, FailedAttempts: 2, Err: failureErr,
},
}
if test.wantAdvance {
select {
@@ -160,16 +169,23 @@ func TestPlaylistControllerFailurePolicy(t *testing.T) {
case <-time.After(time.Second):
t.Fatal("failure did not advance playlist")
}
snapshot, _ := controller.Snapshot()
snapshot := waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool {
return snapshot.State.CurrentIndex == 1 && snapshot.Revision == 2
})
if snapshot.State.CurrentIndex != 1 || snapshot.Revision != 2 {
t.Fatalf("snapshot = %#v, want index 1 revision 2", snapshot)
}
assertPlaylistFailure(t, snapshot, failureErr)
} else {
select {
case command := <-sessions:
t.Fatalf("unexpected session command: %#v", command)
case <-time.After(20 * time.Millisecond):
}
snapshot := waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool {
return snapshot.HasFailure
})
assertPlaylistFailure(t, snapshot, failureErr)
}
cancel()
@@ -179,3 +195,16 @@ func TestPlaylistControllerFailurePolicy(t *testing.T) {
})
}
}
func assertPlaylistFailure(t *testing.T, snapshot PlaylistSnapshot, wantErr error) {
t.Helper()
if !snapshot.HasFailure {
t.Fatal("snapshot has no playlist failure")
}
failure := snapshot.Failure
if failure.EntryIndex != 0 || failure.EntryName != "first" ||
failure.Status.Unit != UnitVideo || failure.Status.Attempt != 2 ||
failure.Status.FailedAttempts != 2 || !errors.Is(failure.Status.Err, wantErr) {
t.Fatalf("failure = %#v", failure)
}
}
+175
View File
@@ -0,0 +1,175 @@
package playback
import (
"context"
"errors"
"testing"
"time"
)
func TestPlaylistFailurePipeline(t *testing.T) {
for _, test := range []struct {
name string
policy PlaylistFailurePolicy
wantAdvance bool
}{
{name: "wait retains failed entry", policy: PlaylistFailureWait},
{name: "next advances failed entry", policy: PlaylistFailureNext, wantAdvance: true},
} {
t.Run(test.name, func(t *testing.T) {
playlist := Playlist{
OnFailure: test.policy,
Entries: []PlaylistEntry{
{
Name: "independent",
Video: PlaylistFeed{Domain: "/video", UUID: "video"},
Audio: PlaylistFeed{Domain: "/audio", UUID: "audio"},
},
{Name: "next", Video: PlaylistFeed{Domain: "/video", UUID: "next"}},
},
}
h := startPlaylistPipeline(t, playlist)
defer h.stop(t)
h.commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 0}
selected := receivePlaylistSession(t, h.sessions).Session
waitForPlaylistSnapshot(t, h.controller, func(snapshot PlaylistSnapshot) bool {
return snapshot.Revision == 1
})
h.setSession(selected, 4)
failureErr := errors.New("audio retries exhausted")
h.statuses.set(PlaybackStatusSnapshot{
Generation: 4,
Audio: Status{
Unit: UnitAudio, State: StateFailed, Generation: 4,
Feed: selected.Audio, Attempt: 3, FailedAttempts: 3, Err: failureErr,
},
HasAudio: true,
})
h.ticker.tick()
if test.wantAdvance {
next := receivePlaylistSession(t, h.sessions)
if next.Session.Video.UUID != "next" || next.Session.Audio.IsConfigured() {
t.Fatalf("advanced session = %#v", next.Session)
}
} else {
select {
case command := <-h.sessions:
t.Fatalf("wait policy advanced with %#v", command)
case <-time.After(20 * time.Millisecond):
}
}
snapshot := waitForPlaylistSnapshot(t, h.controller, func(snapshot PlaylistSnapshot) bool {
return snapshot.HasFailure
})
if snapshot.Failure.Status.Unit != UnitAudio ||
!errors.Is(snapshot.Failure.Status.Err, failureErr) {
t.Fatalf("failure snapshot = %#v", snapshot.Failure)
}
wantIndex := 0
if test.wantAdvance {
wantIndex = 1
}
if snapshot.State.CurrentIndex != wantIndex {
t.Fatalf("current index = %d, want %d", snapshot.State.CurrentIndex, wantIndex)
}
})
}
}
func TestPlaylistManualSelectionWhileFeedIsReconnecting(t *testing.T) {
playlist := navigationPlaylist(false)
h := startPlaylistPipeline(t, playlist)
defer h.stop(t)
h.commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 0}
selected := receivePlaylistSession(t, h.sessions).Session
waitForPlaylistSnapshot(t, h.controller, func(snapshot PlaylistSnapshot) bool {
return snapshot.Revision == 1
})
h.setSession(selected, 2)
h.statuses.set(PlaybackStatusSnapshot{
Generation: 2,
Video: Status{
Unit: UnitVideo, State: StateReconnecting, Generation: 2,
Feed: selected.Video, Attempt: 2, FailedAttempts: 1,
},
HasVideo: true,
})
h.ticker.tick()
h.commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 1}
next := receivePlaylistSession(t, h.sessions)
if next.Session.Audio.UUID != "audio-2" || next.Session.Video.IsConfigured() {
t.Fatalf("manual selection session = %#v", next.Session)
}
waitForPlaylistSnapshot(t, h.controller, func(snapshot PlaylistSnapshot) bool {
return snapshot.Revision == 2 && snapshot.State.CurrentIndex == 1
})
}
type playlistPipelineHarness struct {
controller *PlaylistController
commands chan PlaylistCommand
sessions chan SessionCommand
session *fakeSessionSnapshotSource
statuses *fakePlaybackStatusSnapshotSource
ticker *fakePlaylistEventTicker
cancel context.CancelFunc
results chan error
}
func startPlaylistPipeline(t *testing.T, playlist Playlist) *playlistPipelineHarness {
t.Helper()
sessions := make(chan SessionCommand, 8)
controller, err := NewPlaylistController(playlist, validPlaylistRetryPolicy(), sessions)
if err != nil {
t.Fatal(err)
}
events := make(chan PlaylistEvent, 8)
session := &fakeSessionSnapshotSource{}
statuses := &fakePlaybackStatusSnapshotSource{}
coordinator, err := NewPlaylistEventCoordinator(
controller, session, statuses, events, time.Millisecond,
)
if err != nil {
t.Fatal(err)
}
ticker := newFakePlaylistEventTicker()
coordinator.newTicker = func(time.Duration) playlistEventTicker { return ticker }
commands := make(chan PlaylistCommand, 8)
ctx, cancel := context.WithCancel(context.Background())
results := make(chan error, 2)
go func() { results <- controller.Run(ctx, commands, events) }()
go func() { results <- coordinator.Run(ctx) }()
return &playlistPipelineHarness{
controller: controller, commands: commands, sessions: sessions,
session: session, statuses: statuses, ticker: ticker,
cancel: cancel, results: results,
}
}
func (h *playlistPipelineHarness) setSession(desired SessionConfig, generation uint64) {
plan, err := BuildSessionPlan(desired, func(FeedConfig, FeedConfig) bool { return false })
if err != nil {
panic(err)
}
h.session.set(SessionSnapshot{Desired: desired, Plan: plan, Generation: generation}, true)
}
func (h *playlistPipelineHarness) stop(t *testing.T) {
t.Helper()
h.cancel()
for range 2 {
select {
case err := <-h.results:
if !errors.Is(err, context.Canceled) {
t.Fatalf("pipeline Run() error = %v", err)
}
case <-time.After(time.Second):
t.Fatal("playlist pipeline did not stop")
}
}
}
@@ -344,6 +344,38 @@ func TestSessionControllerCancellationStopsAndJoinsRuntime(t *testing.T) {
}
}
func TestSessionControllerStopAllWhileSessionIsActive(t *testing.T) {
events := make(chan controllerEvent, 32)
controller := newRecordingController(t, events)
initial := validCommandSession()
initial.SyncRequested = false
commands := make(chan SessionCommand)
done := make(chan error, 1)
go func() { done <- controller.Run(context.Background(), initial, commands) }()
receiveIndependentStarts(t, events)
commands <- SessionCommand{Kind: CommandStopAll}
stopped := map[Unit]bool{}
for len(stopped) < 2 {
event := receiveControllerEvent(t, events)
if event.action != "stop" || (event.unit != UnitVideo && event.unit != UnitAudio) {
t.Fatalf("unexpected stop event: %+v", event)
}
stopped[event.unit] = true
}
snapshot := waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
return snapshot.Plan.Topology == TopologyIdle
})
if snapshot.Desired.Video.Active || snapshot.Desired.Audio.Active {
t.Fatalf("stopped desired session = %#v", snapshot.Desired)
}
close(commands)
if err := <-done; err != nil {
t.Fatalf("Run() error = %v", err)
}
}
func TestSessionControllerSnapshotTracksDesiredPlanAndGeneration(t *testing.T) {
events := make(chan controllerEvent, 64)
controller := newRecordingController(t, events)
+13
View File
@@ -57,6 +57,19 @@ func (b *VideoBridge) Next(
}
}
// TryNext returns a frame only when a producer is already waiting. It never
// blocks the caller, allowing UI/render loops to run independently of video
// frame cadence. A returned frame has the same completion requirements as
// one returned by Next.
func (b *VideoBridge) TryNext() (*PendingVideoFrame, bool) {
select {
case pending := <-b.requests:
return pending, true
default:
return nil, false
}
}
func (f *PendingVideoFrame) Complete(err error) {
f.completeOnce.Do(func() {
f.result <- err
+49
View File
@@ -144,3 +144,52 @@ func TestVideoBridgeNextHonorsCancellation(t *testing.T) {
t.Fatalf("Next() error = %v, want %v", err, context.Canceled)
}
}
func TestVideoBridgeTryNextReturnsImmediatelyWhenEmpty(t *testing.T) {
bridge := NewVideoBridge()
if pending, ok := bridge.TryNext(); ok || pending != nil {
t.Fatalf("TryNext() = %#v, %t; want nil, false", pending, ok)
}
}
func TestVideoBridgeTryNextDeliversWithoutCopyAndRequiresCompletion(t *testing.T) {
bridge := NewVideoBridge()
frame := VideoFrame{Index: 9, Payload: []byte{1, 2, 3}}
consumeResult := make(chan error, 1)
started := make(chan struct{})
go func() {
close(started)
consumeResult <- bridge.ConsumeVideo(context.Background(), frame)
}()
<-started
deadline := time.Now().Add(videoBridgeTestTimeout)
var pending *PendingVideoFrame
for pending == nil && time.Now().Before(deadline) {
pending, _ = bridge.TryNext()
if pending == nil {
time.Sleep(time.Millisecond)
}
}
if pending == nil {
t.Fatal("TryNext() did not receive waiting producer")
}
if &pending.Frame.Payload[0] != &frame.Payload[0] {
t.Fatal("TryNext() copied borrowed payload")
}
select {
case err := <-consumeResult:
t.Fatalf("ConsumeVideo() returned before completion: %v", err)
default:
}
pending.Complete(nil)
select {
case err := <-consumeResult:
if err != nil {
t.Fatalf("ConsumeVideo() error = %v", err)
}
case <-time.After(videoBridgeTestTimeout):
t.Fatal("ConsumeVideo() did not return after completion")
}
}
+19 -12
View File
@@ -455,7 +455,12 @@ func validateFramePayload(
// DrawFrame acquires an image, records commands, submits, and presents.
// Returns ErrOutOfDate if the swapchain needs recreation
func (r *Renderer) DrawFrame(videoW, videoH, stride uint32) error {
func (r *Renderer) DrawFrame(
videoW uint32,
videoH uint32,
stride uint32,
showVideo bool,
) error {
imageIndex, res := r.dev.AcquireNextImage(r.swapchain, r.imageAvailable, ^uint64(0))
if res == vk.ErrorOutOfDateKHR || res == vk.SuboptimalKHR {
return ErrOutOfDate
@@ -483,7 +488,7 @@ func (r *Renderer) DrawFrame(videoW, videoH, stride uint32) error {
r.fbs[imageIndex],
vk.Rect2D{Offset: vk.Offset2D{X: 0, Y: 0}, Extent: r.extent},
[]vk.ClearValue{
vk.ClearColor(0.0, 0.0, 0.0, 1.0),
vk.ClearColor(0.025, 0.03, 0.04, 1.0),
vk.ClearDepthStencil(1.0, 0),
},
)
@@ -494,17 +499,19 @@ func (r *Renderer) DrawFrame(videoW, videoH, stride uint32) error {
MinDepth: 0, MaxDepth: 1,
})
cmd.SetScissor(vk.Rect2D{Offset: vk.Offset2D{X: 0, Y: 0}, Extent: r.extent})
cmd.BindPipeline(r.decodePipeline)
cmd.BindDescriptorSet(r.decodeLayout, 0, r.decodeSet)
pc := PushConstants{
Width: videoW,
Height: videoH,
StrideBytes: stride,
WinW: r.extent.Width,
WinH: r.extent.Height,
if showVideo {
cmd.BindPipeline(r.decodePipeline)
cmd.BindDescriptorSet(r.decodeLayout, 0, r.decodeSet)
pc := PushConstants{
Width: videoW,
Height: videoH,
StrideBytes: stride,
WinW: r.extent.Width,
WinH: r.extent.Height,
}
cmd.PushConstants(r.decodeLayout, vk.ShaderStageFragment, 0, unsafe.Pointer(&pc), 20)
cmd.Draw(3, 1, 0, 0)
}
cmd.PushConstants(r.decodeLayout, vk.ShaderStageFragment, 0, unsafe.Pointer(&pc), 20)
cmd.Draw(3, 1, 0, 0)
if r.ImGuiDraw != nil {
r.ImGuiDraw(cmd)
}
+26 -2
View File
@@ -63,8 +63,11 @@ var (
sdlGetAudioPlaybackDevices func(count *int32) uintptr
sdlGetAudioDeviceName func(devid uint32) uintptr
sdlStartTextInput func(window uintptr)
sdlStopTextInput func(window uintptr)
sdlStartTextInput func(window uintptr)
sdlStopTextInput func(window uintptr)
sdlGetClipboardText func() uintptr
sdlSetClipboardText func(text *byte) bool
sdlFree func(memory uintptr)
)
var loaded = false
@@ -98,6 +101,9 @@ func Load() error {
// input
purego.RegisterLibFunc(&sdlStartTextInput, h, "SDL_StartTextInput")
purego.RegisterLibFunc(&sdlStopTextInput, h, "SDL_StopTextInput")
purego.RegisterLibFunc(&sdlGetClipboardText, h, "SDL_GetClipboardText")
purego.RegisterLibFunc(&sdlSetClipboardText, h, "SDL_SetClipboardText")
purego.RegisterLibFunc(&sdlFree, h, "SDL_free")
loaded = true
return nil
}
@@ -197,3 +203,21 @@ func GetAudioPlaybackDevices() []AudioDevice {
// Input wrappers
func StartTextInput(window uintptr) { sdlStartTextInput(window) }
func StopTextInput(window uintptr) { sdlStopTextInput(window) }
func GetClipboardText() string {
text := sdlGetClipboardText()
if text == 0 {
return ""
}
result := cstr(text)
sdlFree(text)
return result
}
func SetClipboardText(text string) bool {
bytes := make([]byte, len(text)+1)
copy(bytes, text)
result := sdlSetClipboardText(&bytes[0])
runtime.KeepAlive(bytes)
return result
}
+5
View File
@@ -1,6 +1,11 @@
{
"loop": true,
"on_failure": "next",
"retry": {
"max_attempts": 3,
"initial_delay": "500ms",
"max_delay": "5s"
},
"entries": [
{
"name": "timelapse",