source-aware playback statuses.

cmd/mxl-player/main.go

@@ -730,7 +730,7 @@ func main() {
 					}
 
 				}
-				if cimgui.CollapsingHeaderTreeNodeFlagsV("Debug Info", collapsingHeaderFlags) {
+				if cimgui.CollapsingHeaderTreeNodeFlagsV("Debug Info", cimgui.TreeNodeFlagsNone) {
 					drawDebugSection()
 				}
 			}

internal/playback/audio_worker.go

@@ -72,8 +72,9 @@ func (s *stabilityAudioSink) ConsumeAudio(
 	return err
 }
 
-func (w *AudioWorker) emit(ctx context.Context, status Status) {
+func (w *AudioWorker) emit(ctx context.Context, config FeedConfig, status Status) {
 	status.Generation = generationFromContext(ctx)
+	status.Feed = config
 	if w.observer != nil {
 		w.observer(status)
 	}
@@ -100,7 +101,7 @@ func (w *AudioWorker) Run(
 		if attemptNumber > 1 {
 			state = StateReconnecting
 		}
-		w.emit(ctx, Status{
+		w.emit(ctx, config, Status{
 			Unit:    UnitAudio,
 			State:   state,
 			Attempt: attemptNumber,
@@ -109,7 +110,7 @@ func (w *AudioWorker) Run(
 		attemptSink := &stabilityAudioSink{
 			sink: w.sink,
 			onStable: func() {
-				w.emit(ctx, Status{
+				w.emit(ctx, config, Status{
 					Unit:    UnitAudio,
 					State:   StatePlaying,
 					Attempt: attemptNumber,
@@ -135,7 +136,7 @@ func (w *AudioWorker) Run(
 			return
 		}
 
-		w.emit(ctx, Status{
+		w.emit(ctx, config, Status{
 			Unit:           UnitAudio,
 			State:          StateReconnecting,
 			Attempt:        attemptNumber + 1,
@@ -155,11 +156,11 @@ func (w *AudioWorker) Run(
 	)
 
 	if ctx.Err() != nil {
-		w.emit(ctx, Status{
+		w.emit(ctx, config, Status{
 			Unit:  UnitAudio,
 			State: StateStopping,
 		})
-		w.emit(ctx, Status{
+		w.emit(ctx, config, Status{
 			Unit:  UnitAudio,
 			State: StateIdle,
 		})
@@ -167,7 +168,7 @@ func (w *AudioWorker) Run(
 	}
 
 	if err != nil {
-		w.emit(ctx, Status{
+		w.emit(ctx, config, Status{
 			Unit:           UnitAudio,
 			State:          StateFailed,
 			Attempt:        attemptNumber,
@@ -177,7 +178,7 @@ func (w *AudioWorker) Run(
 		return err
 	}
 
-	w.emit(ctx, Status{
+	w.emit(ctx, config, Status{
 		Unit:  UnitAudio,
 		State: StateIdle,
 	})

internal/playback/audio_worker_test.go

@@ -127,10 +127,8 @@ func TestAudioWorkerStatusesInheritGeneration(t *testing.T) {
 		func(error) bool { return true },
 		func(status Status) { statuses = append(statuses, status) },
 	)
-	_ = worker.Run(
+	config := FeedConfig{Domain: "/audio", UUID: "audio", Active: true}
-		withGeneration(context.Background(), 8),
+	_ = worker.Run(withGeneration(context.Background(), 8), config)
-		FeedConfig{Domain: "/audio", UUID: "audio", Active: true},
-	)
 	if len(statuses) == 0 {
 		t.Fatal("no statuses emitted")
 	}
@@ -138,6 +136,9 @@ func TestAudioWorkerStatusesInheritGeneration(t *testing.T) {
 		if status.Generation != 8 {
 			t.Fatalf("status generation = %d, want 8: %+v", status.Generation, status)
 		}
+		if status.Feed != config {
+			t.Fatalf("status feed = %#v, want %#v", status.Feed, config)
+		}
 	}
 }
 

internal/playback/playlist_readiness.go

@@ -18,6 +18,7 @@ func IsSessionPlaying(
 			statuses.Video,
 			statuses.HasVideo,
 			session.Generation,
+			session.Plan.Video,
 		) {
 			return false
 		}
@@ -25,17 +26,17 @@ func IsSessionPlaying(
 			statuses.Audio,
 			statuses.HasAudio,
 			session.Generation,
+			session.Plan.Audio,
 		) {
 			return false
 		}
 		return true
 
 	case TopologySynchronized:
-		return statusIsPlaying(
+		return statuses.HasSync &&
-			statuses.Sync,
+			statuses.Sync.Generation == session.Generation &&
-			statuses.HasSync,
+			statuses.Sync.State == StatePlaying &&
-			session.Generation,
+			sameSyncSource(statuses.Sync.Pair, session.Plan.Sync)
-		)
 
 	case TopologyIdle:
 		return false
@@ -45,8 +46,23 @@ func IsSessionPlaying(
 	}
 }
 
-func statusIsPlaying(status Status, present bool, generation uint64) bool {
+func statusIsPlaying(
+	status Status,
+	present bool,
+	generation uint64,
+	feed FeedConfig,
+) bool {
 	return present &&
 		status.Generation == generation &&
-		status.State == StatePlaying
+		status.State == StatePlaying &&
+		sameFeedSource(status.Feed, feed)
+}
+
+func sameFeedSource(a, b FeedConfig) bool {
+	return a.Domain == b.Domain && a.UUID == b.UUID
+}
+
+func sameSyncSource(a, b SyncPairConfig) bool {
+	return sameFeedSource(a.Video, b.Video) &&
+		sameFeedSource(a.Audio, b.Audio)
 }

internal/playback/playlist_readiness_test.go

@@ -5,7 +5,14 @@ import "testing"
 func TestIsSessionPlaying(t *testing.T) {
 	const generation = 4
 	playing := func(unit Unit) Status {
-		return Status{Unit: unit, State: StatePlaying, Generation: generation}
+		status := Status{Unit: unit, State: StatePlaying, Generation: generation}
+		switch unit {
+		case UnitVideo:
+			status.Feed = FeedConfig{UUID: "video"}
+		case UnitAudio:
+			status.Feed = FeedConfig{UUID: "audio"}
+		}
+		return status
 	}
 	tests := []struct {
 		name     string
@@ -171,6 +178,72 @@ func TestIsSessionPlaying(t *testing.T) {
 				HasSync:    true,
 			},
 		},
+		{
+			name: "video status has wrong UUID",
+			session: SessionSnapshot{
+				Generation: generation,
+				Plan: SessionPlan{
+					Topology: TopologyIndependent,
+					Video:    FeedConfig{Domain: "domain", UUID: "video", Active: true},
+				},
+			},
+			statuses: PlaybackStatusSnapshot{
+				Generation: generation,
+				Video: Status{
+					Unit:       UnitVideo,
+					State:      StatePlaying,
+					Generation: generation,
+					Feed:       FeedConfig{Domain: "domain", UUID: "other"},
+				},
+				HasVideo: true,
+			},
+		},
+		{
+			name: "audio status has wrong domain",
+			session: SessionSnapshot{
+				Generation: generation,
+				Plan: SessionPlan{
+					Topology: TopologyIndependent,
+					Audio:    FeedConfig{Domain: "audio-domain", UUID: "audio", Active: true},
+				},
+			},
+			statuses: PlaybackStatusSnapshot{
+				Generation: generation,
+				Audio: Status{
+					Unit:       UnitAudio,
+					State:      StatePlaying,
+					Generation: generation,
+					Feed:       FeedConfig{Domain: "other-domain", UUID: "audio"},
+				},
+				HasAudio: true,
+			},
+		},
+		{
+			name: "sync status has wrong audio source",
+			session: SessionSnapshot{
+				Generation: generation,
+				Plan: SessionPlan{
+					Topology: TopologySynchronized,
+					Sync: SyncPairConfig{
+						Video: FeedConfig{Domain: "domain", UUID: "video"},
+						Audio: FeedConfig{Domain: "domain", UUID: "audio"},
+					},
+				},
+			},
+			statuses: PlaybackStatusSnapshot{
+				Generation: generation,
+				Sync: Status{
+					Unit:       UnitSync,
+					State:      StatePlaying,
+					Generation: generation,
+					Pair: SyncPairConfig{
+						Video: FeedConfig{Domain: "domain", UUID: "video"},
+						Audio: FeedConfig{Domain: "domain", UUID: "other-audio"},
+					},
+				},
+				HasSync: true,
+			},
+		},
 		{
 			name: "idle",
 			session: SessionSnapshot{

internal/playback/state.go

@@ -29,6 +29,10 @@ type Status struct {
 	Unit       Unit
 	State      State
 	Generation uint64
+
+	Feed FeedConfig     // Video/Audio worker
+	Pair SyncPairConfig // Sync worker
+
 	Attempt        int
 	FailedAttempts int
 	RetryIn        time.Duration

internal/playback/sync_worker.go

@@ -59,8 +59,13 @@ func NewSyncWorker(
 	}, nil
 }
 
-func (w *SyncWorker) emit(ctx context.Context, status Status) {
+func (w *SyncWorker) emit(
+	ctx context.Context,
+	pair SyncPairConfig,
+	status Status,
+) {
 	status.Generation = generationFromContext(ctx)
+	status.Pair = pair
 	if w.observer != nil {
 		w.observer(status)
 	}
@@ -80,6 +85,7 @@ func (w *SyncWorker) Run(
 	if !videoConfig.Active || !audioConfig.Active {
 		return ErrSyncFeedsInactive
 	}
+	pair := SyncPairConfig{Video: videoConfig, Audio: audioConfig}
 
 	attemptNumber := 0
 	var latestRetry retryEvent
@@ -91,7 +97,7 @@ func (w *SyncWorker) Run(
 		if attemptNumber > 1 {
 			state = StateReconnecting
 		}
-		w.emit(ctx, Status{
+		w.emit(ctx, pair, Status{
 			Unit:    UnitSync,
 			State:   state,
 			Attempt: attemptNumber,
@@ -100,7 +106,7 @@ func (w *SyncWorker) Run(
 		attemptAudioSink := &stabilityAudioSink{
 			sink: w.audioSink,
 			onStable: func() {
-				w.emit(ctx, Status{
+				w.emit(ctx, pair, Status{
 					Unit:    UnitSync,
 					State:   StatePlaying,
 					Attempt: attemptNumber,
@@ -134,7 +140,7 @@ func (w *SyncWorker) Run(
 		if !event.WillRetry {
 			return
 		}
-		w.emit(ctx, Status{
+		w.emit(ctx, pair, Status{
 			Unit:           UnitSync,
 			State:          StateReconnecting,
 			Attempt:        attemptNumber + 1,
@@ -152,18 +158,18 @@ func (w *SyncWorker) Run(
 		observeRetry,
 	)
 	if ctx.Err() != nil {
-		w.emit(ctx, Status{
+		w.emit(ctx, pair, Status{
 			Unit:  UnitSync,
 			State: StateStopping,
 		})
-		w.emit(ctx, Status{
+		w.emit(ctx, pair, Status{
 			Unit:  UnitSync,
 			State: StateIdle,
 		})
 		return ctx.Err()
 	}
 	if err != nil {
-		w.emit(ctx, Status{
+		w.emit(ctx, pair, Status{
 			Unit:           UnitSync,
 			State:          StateFailed,
 			Attempt:        attemptNumber,
@@ -172,7 +178,7 @@ func (w *SyncWorker) Run(
 		})
 		return err
 	}
-	w.emit(ctx, Status{
+	w.emit(ctx, pair, Status{
 		Unit:  UnitSync,
 		State: StateIdle,
 	})

internal/playback/sync_worker_test.go

@@ -146,6 +146,10 @@ func TestSyncWorkerStatusesInheritGeneration(t *testing.T) {
 		if status.Generation != 9 {
 			t.Fatalf("status generation = %d, want 9: %+v", status.Generation, status)
 		}
+		wantPair := SyncPairConfig{Video: video, Audio: audio}
+		if status.Pair != wantPair {
+			t.Fatalf("status pair = %#v, want %#v", status.Pair, wantPair)
+		}
 	}
 }
 

internal/playback/video_worker.go

@@ -72,8 +72,9 @@ func (s *stabilityVideoSink) ConsumeVideo(
 	return err
 }
 
-func (w *VideoWorker) emit(ctx context.Context, status Status) {
+func (w *VideoWorker) emit(ctx context.Context, config FeedConfig, status Status) {
 	status.Generation = generationFromContext(ctx)
+	status.Feed = config
 	if w.observer != nil {
 		w.observer(status)
 	}
@@ -100,7 +101,7 @@ func (w *VideoWorker) Run(
 		if attemptNumber > 1 {
 			state = StateReconnecting
 		}
-		w.emit(ctx, Status{
+		w.emit(ctx, config, Status{
 			Unit:    UnitVideo,
 			State:   state,
 			Attempt: attemptNumber,
@@ -109,7 +110,7 @@ func (w *VideoWorker) Run(
 		attemptSink := &stabilityVideoSink{
 			sink: w.sink,
 			onStable: func() {
-				w.emit(ctx, Status{
+				w.emit(ctx, config, Status{
 					Unit:    UnitVideo,
 					State:   StatePlaying,
 					Attempt: attemptNumber,
@@ -135,7 +136,7 @@ func (w *VideoWorker) Run(
 			return
 		}
 
-		w.emit(ctx, Status{
+		w.emit(ctx, config, Status{
 			Unit:           UnitVideo,
 			State:          StateReconnecting,
 			Attempt:        attemptNumber + 1,
@@ -155,11 +156,11 @@ func (w *VideoWorker) Run(
 	)
 
 	if ctx.Err() != nil {
-		w.emit(ctx, Status{
+		w.emit(ctx, config, Status{
 			Unit:  UnitVideo,
 			State: StateStopping,
 		})
-		w.emit(ctx, Status{
+		w.emit(ctx, config, Status{
 			Unit:  UnitVideo,
 			State: StateIdle,
 		})
@@ -167,7 +168,7 @@ func (w *VideoWorker) Run(
 	}
 
 	if err != nil {
-		w.emit(ctx, Status{
+		w.emit(ctx, config, Status{
 			Unit:           UnitVideo,
 			State:          StateFailed,
 			Attempt:        attemptNumber,
@@ -177,7 +178,7 @@ func (w *VideoWorker) Run(
 		return err
 	}
 
-	w.emit(ctx, Status{
+	w.emit(ctx, config, Status{
 		Unit:  UnitVideo,
 		State: StateIdle,
 	})

internal/playback/video_worker_test.go

@@ -162,7 +162,8 @@ func TestVideoWorkerStatusesInheritGeneration(t *testing.T) {
 		func(error) bool { return true },
 		func(status Status) { statuses = append(statuses, status) },
 	)
-	_ = worker.Run(withGeneration(context.Background(), 7), activeVideoConfig())
+	config := activeVideoConfig()
+	_ = worker.Run(withGeneration(context.Background(), 7), config)
 	if len(statuses) == 0 {
 		t.Fatal("no statuses emitted")
 	}
@@ -170,6 +171,9 @@ func TestVideoWorkerStatusesInheritGeneration(t *testing.T) {
 		if status.Generation != 7 {
 			t.Fatalf("status generation = %d, want 7: %+v", status.Generation, status)
 		}
+		if status.Feed != config {
+			t.Fatalf("status feed = %#v, want %#v", status.Feed, config)
+		}
 	}
 }