Compare commits
5 Commits
b741b7d01f
...
e830da4a32
| Author | SHA1 | Date | |
|---|---|---|---|
 Dmitry Sergeev
|
e830da4a32 | ||
|
Dmitry Sergeev
|
19e2e076e9 | ||
|
Dmitry Sergeev
|
03ebe5bbd0 | ||
|
Dmitry Sergeev
|
ea8be34282 | ||
|
Dmitry Sergeev
|
a4cbed188a |
+68
-542
@@ -6,16 +6,12 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
mxladapter "mxl-player/internal/adapter/mxl"
|
||||
"mxl-player/internal/imgui"
|
||||
"mxl-player/internal/output"
|
||||
"mxl-player/internal/playback"
|
||||
"mxl-player/internal/renderer"
|
||||
"mxl-player/internal/sdl"
|
||||
"mxl-player/internal/source"
|
||||
"os"
|
||||
"runtime"
|
||||
"sync"
|
||||
"time"
|
||||
"unsafe"
|
||||
|
||||
@@ -170,22 +166,6 @@ func main() {
|
||||
if !args.ListAudio && !args.ListGPU {
|
||||
checkMXLargs(args)
|
||||
}
|
||||
if args.SyncRequested &&
|
||||
args.VideoFlowId != "" &&
|
||||
args.AudioFlowId != "" &&
|
||||
args.VideoDomain != args.AudioDomain {
|
||||
fmt.Fprintln(
|
||||
os.Stderr,
|
||||
"--sync currently requires video and audio to use the same MXL domain",
|
||||
)
|
||||
os.Exit(2)
|
||||
}
|
||||
// path selection
|
||||
useLegacySync := args.SyncRequested &&
|
||||
args.VideoFlowId != "" &&
|
||||
args.AudioFlowId != ""
|
||||
useIndependentSlots := !useLegacySync
|
||||
|
||||
runtime.LockOSThread()
|
||||
if err := sdl.Load(); err != nil {
|
||||
panic(err)
|
||||
@@ -289,77 +269,6 @@ func main() {
|
||||
}
|
||||
defer vkDevice.Destroy()
|
||||
|
||||
var (
|
||||
syncSrc *source.SyncSource
|
||||
videoSrc *source.Source
|
||||
audioSrc *source.AudioSource
|
||||
audioStream uintptr
|
||||
audioBatch uint64
|
||||
aChans uint64
|
||||
)
|
||||
|
||||
interleaveAudio := func(samples [][]byte) []byte {
|
||||
frameBytes := int(audioBatch) * int(aChans) * 4
|
||||
out := make([]byte, frameBytes)
|
||||
for ch := uint64(0); ch < aChans; ch++ {
|
||||
srcBytes := samples[ch]
|
||||
for i := uint64(0); i < audioBatch; i++ {
|
||||
srcOff := i * 4
|
||||
dstOff := (i*aChans + ch) * 4
|
||||
if srcOff+4 <= uint64(len(srcBytes)) {
|
||||
copy(out[dstOff:dstOff+4], srcBytes[srcOff:srcOff+4])
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
switch {
|
||||
case useLegacySync:
|
||||
syncSrc, err = source.OpenSameDomainSync(
|
||||
args.VideoDomain,
|
||||
args.VideoFlowId,
|
||||
args.AudioFlowId,
|
||||
)
|
||||
if err != nil {
|
||||
log.Fatalf("sync source: %v", err)
|
||||
}
|
||||
aChans = syncSrc.Channels()
|
||||
audioBatch = uint64(syncSrc.AudioRate().Num) / uint64(syncSrc.Rate().Num)
|
||||
if audioBatch == 0 {
|
||||
audioBatch = 1
|
||||
}
|
||||
audioStream = sdl.OpenAudioDeviceStream(sdlAudioDevice, sdl.AudioSpec{
|
||||
Format: sdl.AudioF32,
|
||||
Channels: int32(aChans),
|
||||
Freq: int32(syncSrc.AudioRate().Num / syncSrc.AudioRate().Den),
|
||||
})
|
||||
if audioStream == 0 {
|
||||
log.Fatalf("audio: %s", sdl.GetError())
|
||||
}
|
||||
sdl.ResumeAudioStreamDevice(audioStream)
|
||||
fmt.Printf("sync: video %dx%d audio %dch batch=%d\n",
|
||||
syncSrc.Width(), syncSrc.Height(), aChans, audioBatch)
|
||||
default:
|
||||
// Independent slots own their readers.
|
||||
// Empty startup opens nothing.
|
||||
}
|
||||
|
||||
defer func() {
|
||||
if syncSrc != nil {
|
||||
_ = syncSrc.Close()
|
||||
}
|
||||
if videoSrc != nil {
|
||||
_ = videoSrc.Close()
|
||||
}
|
||||
if audioSrc != nil {
|
||||
_ = audioSrc.Close()
|
||||
}
|
||||
}()
|
||||
if audioStream != 0 {
|
||||
defer sdl.DestroyAudioStream(audioStream)
|
||||
}
|
||||
|
||||
// Create renderer
|
||||
r, err := renderer.New(renderer.Config{
|
||||
PhysDevice: vkPhysDevice,
|
||||
@@ -414,419 +323,58 @@ func main() {
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
type reconnectParams struct {
|
||||
domain string
|
||||
video string
|
||||
audio string
|
||||
}
|
||||
// Reconnect requests from GUI or automatic retry
|
||||
control := make(chan reconnectParams, 1)
|
||||
videoBridge := playback.NewVideoBridge()
|
||||
statusStore := playback.NewStatusStore()
|
||||
|
||||
videoWorker, err := playback.NewVideoWorker(
|
||||
mxladapter.VideoFactory{},
|
||||
videoBridge,
|
||||
retryPolicy,
|
||||
mxladapter.ShouldRetry,
|
||||
func(status playback.Status) {
|
||||
statusStore.Observe(status)
|
||||
if status.Err != nil {
|
||||
log.Printf(
|
||||
"video: state=%v attempt=%d failed=%d: %v",
|
||||
status.State,
|
||||
status.Attempt,
|
||||
status.FailedAttempts,
|
||||
status.Err,
|
||||
)
|
||||
return
|
||||
}
|
||||
log.Printf(
|
||||
"video: state=%v attempt=%d failed=%d",
|
||||
status.State,
|
||||
status.Attempt,
|
||||
status.FailedAttempts,
|
||||
)
|
||||
},
|
||||
)
|
||||
player, err := newPlayerPlayback(sdlAudioDevice, retryPolicy)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
videoSlot, err := playback.NewVideoSlot(videoWorker)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
videoCommands := make(chan playback.FeedConfig, 1)
|
||||
videoBridge := player.Video
|
||||
statusStore := player.Status
|
||||
syncRequested := args.SyncRequested
|
||||
|
||||
audioOutput := output.NewSDLAudioSink(sdlAudioDevice)
|
||||
defer audioOutput.Close()
|
||||
audioWorker, err := playback.NewAudioWorker(
|
||||
mxladapter.AudioFactory{},
|
||||
audioOutput,
|
||||
retryPolicy,
|
||||
mxladapter.ShouldRetry,
|
||||
func(status playback.Status) {
|
||||
statusStore.Observe(status)
|
||||
|
||||
if status.Err != nil {
|
||||
log.Printf(
|
||||
"audio: state=%v attempt=%d failed=%d: %v",
|
||||
status.State,
|
||||
status.Attempt,
|
||||
status.FailedAttempts,
|
||||
status.Err,
|
||||
)
|
||||
return
|
||||
}
|
||||
|
||||
log.Printf(
|
||||
"audio: state=%v attempt=%d failed=%d",
|
||||
status.State,
|
||||
status.Attempt,
|
||||
status.FailedAttempts,
|
||||
)
|
||||
},
|
||||
)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
audioSlot, err := playback.NewAudioSlot(audioWorker)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
audioCommands := make(chan playback.FeedConfig, 1)
|
||||
|
||||
reopen := func(params reconnectParams) error {
|
||||
// Close current sources
|
||||
if syncSrc != nil {
|
||||
_ = syncSrc.Close()
|
||||
syncSrc = nil
|
||||
}
|
||||
if videoSrc != nil {
|
||||
_ = videoSrc.Close()
|
||||
videoSrc = nil
|
||||
}
|
||||
if audioSrc != nil {
|
||||
_ = audioSrc.Close()
|
||||
audioSrc = nil
|
||||
}
|
||||
// Try once. Return error if fails — caller loops back to select
|
||||
// and can pick up newer reconnect request.
|
||||
if params.video != "" && params.audio != "" {
|
||||
s, e := source.OpenSameDomainSync(params.domain, params.video, params.audio)
|
||||
if e == nil {
|
||||
syncSrc = s
|
||||
aChans = s.Channels()
|
||||
audioBatch = uint64(s.AudioRate().Num) / uint64(s.Rate().Num)
|
||||
if audioBatch == 0 {
|
||||
audioBatch = 1
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return e
|
||||
} else if params.video != "" {
|
||||
s, e := source.Open(params.domain, params.video)
|
||||
if e == nil {
|
||||
videoSrc = s
|
||||
return nil
|
||||
}
|
||||
return e
|
||||
} else if params.audio != "" {
|
||||
s, e := source.OpenAudio(params.domain, params.audio)
|
||||
if e == nil {
|
||||
audioSrc = s
|
||||
aChans = s.Channels()
|
||||
audioBatch = uint64(s.Rate().Num) / (100 * uint64(s.Rate().Den))
|
||||
if audioBatch == 0 {
|
||||
audioBatch = 1
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return e
|
||||
}
|
||||
return fmt.Errorf("reopen: no flow specified")
|
||||
}
|
||||
|
||||
enqueueVideoConfig := func(config playback.FeedConfig) {
|
||||
enqueueCommand := func(command playback.SessionCommand) {
|
||||
select {
|
||||
case <-videoCommands:
|
||||
default:
|
||||
}
|
||||
|
||||
select {
|
||||
case videoCommands <- config:
|
||||
case player.Commands <- command:
|
||||
default:
|
||||
log.Printf("playback command queue is full; ignoring command %d", command.Kind)
|
||||
}
|
||||
}
|
||||
enqueueAudioConfig := func(config playback.FeedConfig) {
|
||||
select {
|
||||
case <-audioCommands:
|
||||
default:
|
||||
}
|
||||
|
||||
select {
|
||||
case audioCommands <- config:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
doReconnect := func() {
|
||||
if useIndependentSlots {
|
||||
videoActive = videoStr != ""
|
||||
audioActive = audioStr != ""
|
||||
|
||||
videoConfig := playback.FeedConfig{}
|
||||
if videoStr != "" {
|
||||
videoConfig = playback.FeedConfig{
|
||||
if videoStr == "" {
|
||||
enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveVideo})
|
||||
} else {
|
||||
enqueueCommand(playback.SessionCommand{
|
||||
Kind: playback.CommandSetVideo,
|
||||
Config: playback.FeedConfig{
|
||||
Domain: videoDomainStr,
|
||||
UUID: videoStr,
|
||||
Active: true,
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
audioConfig := playback.FeedConfig{}
|
||||
if audioStr != "" {
|
||||
audioConfig = playback.FeedConfig{
|
||||
if audioStr == "" {
|
||||
enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveAudio})
|
||||
} else {
|
||||
enqueueCommand(playback.SessionCommand{
|
||||
Kind: playback.CommandSetAudio,
|
||||
Config: playback.FeedConfig{
|
||||
Domain: audioDomainStr,
|
||||
UUID: audioStr,
|
||||
Active: true,
|
||||
}
|
||||
}
|
||||
|
||||
enqueueVideoConfig(videoConfig)
|
||||
enqueueAudioConfig(audioConfig)
|
||||
return
|
||||
}
|
||||
// legacy
|
||||
if videoDomainStr != audioDomainStr {
|
||||
log.Printf(
|
||||
"sync reconnect rejected: video domain %q differs from audio domain %q",
|
||||
videoDomainStr,
|
||||
audioDomainStr,
|
||||
)
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-control:
|
||||
default:
|
||||
}
|
||||
control <- reconnectParams{domain: videoDomainStr, video: videoStr, audio: audioStr}
|
||||
}
|
||||
|
||||
playbackDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(playbackDone)
|
||||
|
||||
if useIndependentSlots {
|
||||
var slots sync.WaitGroup
|
||||
slots.Add(2)
|
||||
|
||||
go func() {
|
||||
defer slots.Done()
|
||||
|
||||
err := videoSlot.Run(
|
||||
ctx,
|
||||
playback.FeedConfig{
|
||||
Domain: args.VideoDomain,
|
||||
UUID: args.VideoFlowId,
|
||||
Active: args.VideoFlowId != "",
|
||||
},
|
||||
videoCommands,
|
||||
)
|
||||
if err != nil && !errors.Is(err, context.Canceled) {
|
||||
log.Printf("video slot: %v", err)
|
||||
})
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
playbackDone := make(chan error, 1)
|
||||
go func() {
|
||||
defer slots.Done()
|
||||
|
||||
err := audioSlot.Run(
|
||||
playbackDone <- player.Controller.Run(
|
||||
ctx,
|
||||
playback.FeedConfig{
|
||||
Domain: args.AudioDomain,
|
||||
UUID: args.AudioFlowId,
|
||||
Active: args.AudioFlowId != "",
|
||||
},
|
||||
audioCommands,
|
||||
args.playbackConfig(),
|
||||
player.Commands,
|
||||
)
|
||||
if err != nil && !errors.Is(err, context.Canceled) {
|
||||
log.Printf("audio slot: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
<-ctx.Done()
|
||||
slots.Wait()
|
||||
return
|
||||
}
|
||||
|
||||
// Audio-only mode: independent loop.
|
||||
if audioSrc != nil && syncSrc == nil && videoSrc == nil {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case params := <-control:
|
||||
if params.video != "" || params.audio != "" {
|
||||
if rerr := reopen(params); rerr != nil {
|
||||
if errors.Is(rerr, context.Canceled) {
|
||||
return
|
||||
}
|
||||
log.Printf("source: reopen failed: %v, retrying", rerr)
|
||||
select {
|
||||
case <-time.After(500 * time.Millisecond):
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
select {
|
||||
case control <- params:
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
continue
|
||||
default:
|
||||
}
|
||||
queued := sdl.GetAudioStreamQueued(audioStream)
|
||||
maxQueued := int32(audioBatch) * int32(aChans) * 4 * 20
|
||||
if queued > maxQueued {
|
||||
select {
|
||||
case <-time.After(10 * time.Millisecond):
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
continue
|
||||
}
|
||||
f, err := audioSrc.NextAudio(ctx, audioBatch, 20*time.Millisecond)
|
||||
if err != nil {
|
||||
if errors.Is(err, context.Canceled) {
|
||||
return
|
||||
}
|
||||
log.Printf("source: %v", err)
|
||||
params := reconnectParams{domain: videoDomainStr, video: videoStr, audio: audioStr}
|
||||
select {
|
||||
case <-control:
|
||||
default:
|
||||
}
|
||||
select {
|
||||
case <-time.After(500 * time.Millisecond):
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
select {
|
||||
case control <- params:
|
||||
default:
|
||||
}
|
||||
continue
|
||||
}
|
||||
if f.Samples != nil && audioStream != 0 {
|
||||
sdl.PutAudioStreamData(audioStream, interleaveAudio(f.Samples))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Video bridge provides backpressure: only one borrowed frame is in flight.
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
|
||||
case params := <-control:
|
||||
if rerr := reopen(params); rerr != nil {
|
||||
if errors.Is(rerr, context.Canceled) {
|
||||
return
|
||||
}
|
||||
log.Printf("source: reopen failed: %v, retrying", rerr)
|
||||
|
||||
select {
|
||||
case <-time.After(500 * time.Millisecond):
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
|
||||
select {
|
||||
case control <- params:
|
||||
default:
|
||||
// Preserve an already queued, potentially newer request.
|
||||
}
|
||||
}
|
||||
continue
|
||||
|
||||
default:
|
||||
}
|
||||
|
||||
var videoFrame playback.VideoFrame
|
||||
if syncSrc != nil {
|
||||
vFrame, aFrame, err := syncSrc.NextSync(ctx, audioBatch, 200*time.Millisecond)
|
||||
if err != nil {
|
||||
if errors.Is(err, context.Canceled) {
|
||||
return
|
||||
}
|
||||
log.Printf("source: %v", err)
|
||||
// Request a reconnect after the read failure.
|
||||
params := reconnectParams{
|
||||
domain: videoDomainStr,
|
||||
video: videoStr,
|
||||
audio: audioStr,
|
||||
}
|
||||
select {
|
||||
case control <- params:
|
||||
default:
|
||||
// Preserve an already queued, potentially newer request.
|
||||
}
|
||||
continue
|
||||
}
|
||||
videoFrame = playback.VideoFrame{
|
||||
Index: vFrame.Index,
|
||||
Width: vFrame.Width,
|
||||
Height: vFrame.Height,
|
||||
Stride: vFrame.Stride,
|
||||
Size: vFrame.Size,
|
||||
Invalid: vFrame.Invalid,
|
||||
Payload: vFrame.Payload,
|
||||
}
|
||||
if aFrame.Samples != nil && audioStream != 0 {
|
||||
sdl.PutAudioStreamData(audioStream, interleaveAudio(aFrame.Samples))
|
||||
}
|
||||
|
||||
} else if videoSrc != nil {
|
||||
f, err := videoSrc.NextCtx(ctx, 200*time.Millisecond)
|
||||
if err != nil {
|
||||
if errors.Is(err, context.Canceled) {
|
||||
return
|
||||
}
|
||||
log.Printf("source: %v", err)
|
||||
params := reconnectParams{
|
||||
domain: videoDomainStr,
|
||||
video: videoStr,
|
||||
audio: audioStr,
|
||||
}
|
||||
select {
|
||||
case control <- params:
|
||||
default:
|
||||
// Preserve an already queued, potentially newer request.
|
||||
}
|
||||
continue
|
||||
}
|
||||
videoFrame = playback.VideoFrame{
|
||||
Index: f.Index,
|
||||
Width: f.Width,
|
||||
Height: f.Height,
|
||||
Stride: f.Stride,
|
||||
Size: f.Size,
|
||||
Invalid: f.Invalid,
|
||||
Payload: f.Payload,
|
||||
}
|
||||
}
|
||||
|
||||
if err := videoBridge.ConsumeVideo(ctx, videoFrame); err != nil {
|
||||
if errors.Is(err, context.Canceled) {
|
||||
return
|
||||
}
|
||||
log.Printf("video output: %v", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
running := true
|
||||
@@ -979,41 +527,43 @@ func main() {
|
||||
cimgui.InputTextWithHint("Audio domain", "/dev/shm/mxl", &audioDomainStr, 0, nil)
|
||||
cimgui.InputTextWithHint("Video UUID", "", &videoStr, 0, nil)
|
||||
cimgui.InputTextWithHint("Audio UUID", "", &audioStr, 0, nil)
|
||||
if snapshot, ok := player.Controller.Snapshot(); ok {
|
||||
videoActive = snapshot.Desired.Video.Active
|
||||
audioActive = snapshot.Desired.Audio.Active
|
||||
syncRequested = snapshot.Desired.SyncRequested
|
||||
}
|
||||
if cimgui.Button("Connect") {
|
||||
doReconnect()
|
||||
}
|
||||
cimgui.SameLine()
|
||||
cimgui.Checkbox("Show stats", &showStats)
|
||||
if useIndependentSlots && videoActive {
|
||||
if cimgui.Checkbox("Synchronize", &syncRequested) {
|
||||
kind := playback.CommandDisableSync
|
||||
if syncRequested {
|
||||
kind = playback.CommandEnableSync
|
||||
}
|
||||
enqueueCommand(playback.SessionCommand{Kind: kind})
|
||||
}
|
||||
if videoActive {
|
||||
if cimgui.Button("Stop video") {
|
||||
videoActive = false
|
||||
enqueueVideoConfig(
|
||||
playback.FeedConfig{
|
||||
Domain: videoDomainStr,
|
||||
UUID: videoStr,
|
||||
Active: false,
|
||||
})
|
||||
enqueueCommand(playback.SessionCommand{Kind: playback.CommandStopVideo})
|
||||
}
|
||||
}
|
||||
if useIndependentSlots && !videoActive && videoStr != "" {
|
||||
if !videoActive && videoStr != "" {
|
||||
cimgui.SameLine()
|
||||
if cimgui.Button("Resume video") {
|
||||
videoActive = true
|
||||
enqueueVideoConfig(playback.FeedConfig{
|
||||
Domain: videoDomainStr,
|
||||
UUID: videoStr,
|
||||
Active: true,
|
||||
})
|
||||
enqueueCommand(playback.SessionCommand{Kind: playback.CommandResumeVideo})
|
||||
}
|
||||
}
|
||||
if useIndependentSlots && videoStr != "" {
|
||||
if videoStr != "" {
|
||||
if cimgui.Button("Remove video") {
|
||||
videoActive = false
|
||||
videoStr = ""
|
||||
enqueueVideoConfig(playback.FeedConfig{})
|
||||
enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveVideo})
|
||||
}
|
||||
}
|
||||
if useIndependentSlots {
|
||||
if videoActive {
|
||||
cimgui.Text("Video desired: active")
|
||||
} else if videoStr != "" {
|
||||
@@ -1023,61 +573,39 @@ func main() {
|
||||
}
|
||||
|
||||
if status, ok := statusStore.Snapshot(playback.UnitVideo); ok {
|
||||
cimgui.Text(fmt.Sprintf(
|
||||
"Video actual: %s",
|
||||
status.State,
|
||||
))
|
||||
cimgui.Text(fmt.Sprintf(
|
||||
"Attempt: %d, failed: %d",
|
||||
status.Attempt,
|
||||
status.FailedAttempts,
|
||||
))
|
||||
|
||||
cimgui.Text(fmt.Sprintf("Video actual: %s", status.State))
|
||||
cimgui.Text(fmt.Sprintf("Attempt: %d, failed: %d", status.Attempt, status.FailedAttempts))
|
||||
if status.RetryIn > 0 {
|
||||
cimgui.Text(fmt.Sprintf(
|
||||
"Retry in: %s",
|
||||
status.RetryIn.Round(time.Millisecond),
|
||||
))
|
||||
cimgui.Text(fmt.Sprintf("Retry in: %s", status.RetryIn.Round(time.Millisecond)))
|
||||
}
|
||||
|
||||
if status.Err != nil {
|
||||
cimgui.TextWrapped(status.Err.Error())
|
||||
}
|
||||
} else {
|
||||
cimgui.Text("Video actual: not started")
|
||||
}
|
||||
}
|
||||
if useIndependentSlots && audioActive {
|
||||
if audioActive {
|
||||
if cimgui.Button("Stop audio") {
|
||||
audioActive = false
|
||||
enqueueAudioConfig(playback.FeedConfig{
|
||||
Domain: audioDomainStr,
|
||||
UUID: audioStr,
|
||||
Active: false,
|
||||
})
|
||||
enqueueCommand(playback.SessionCommand{Kind: playback.CommandStopAudio})
|
||||
}
|
||||
}
|
||||
|
||||
if useIndependentSlots && !audioActive && audioStr != "" {
|
||||
if !audioActive && audioStr != "" {
|
||||
if cimgui.Button("Resume audio") {
|
||||
audioActive = true
|
||||
enqueueAudioConfig(playback.FeedConfig{
|
||||
Domain: audioDomainStr,
|
||||
UUID: audioStr,
|
||||
Active: true,
|
||||
})
|
||||
enqueueCommand(playback.SessionCommand{Kind: playback.CommandResumeAudio})
|
||||
}
|
||||
}
|
||||
|
||||
if useIndependentSlots && audioStr != "" {
|
||||
if audioStr != "" {
|
||||
if cimgui.Button("Remove audio") {
|
||||
audioActive = false
|
||||
audioStr = ""
|
||||
enqueueAudioConfig(playback.FeedConfig{})
|
||||
enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveAudio})
|
||||
}
|
||||
}
|
||||
|
||||
if useIndependentSlots {
|
||||
if audioActive {
|
||||
cimgui.Text("Audio desired: active")
|
||||
} else if audioStr != "" {
|
||||
@@ -1087,29 +615,22 @@ func main() {
|
||||
}
|
||||
|
||||
if status, ok := statusStore.Snapshot(playback.UnitAudio); ok {
|
||||
cimgui.Text(fmt.Sprintf(
|
||||
"Audio actual: %s",
|
||||
status.State,
|
||||
))
|
||||
cimgui.Text(fmt.Sprintf(
|
||||
"Attempt: %d, failed: %d",
|
||||
status.Attempt,
|
||||
status.FailedAttempts,
|
||||
))
|
||||
|
||||
cimgui.Text(fmt.Sprintf("Audio actual: %s", status.State))
|
||||
cimgui.Text(fmt.Sprintf("Attempt: %d, failed: %d", status.Attempt, status.FailedAttempts))
|
||||
if status.RetryIn > 0 {
|
||||
cimgui.Text(fmt.Sprintf(
|
||||
"Retry in: %s",
|
||||
status.RetryIn.Round(time.Millisecond),
|
||||
))
|
||||
cimgui.Text(fmt.Sprintf("Retry in: %s", status.RetryIn.Round(time.Millisecond)))
|
||||
}
|
||||
|
||||
if status.Err != nil {
|
||||
cimgui.TextWrapped(status.Err.Error())
|
||||
}
|
||||
} else {
|
||||
cimgui.Text("Audio actual: not started")
|
||||
}
|
||||
if status, ok := statusStore.Snapshot(playback.UnitSync); ok {
|
||||
cimgui.Text(fmt.Sprintf("Sync actual: %s", status.State))
|
||||
if status.Err != nil {
|
||||
cimgui.TextWrapped(status.Err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
cimgui.End()
|
||||
@@ -1141,5 +662,10 @@ func main() {
|
||||
}
|
||||
|
||||
cancel()
|
||||
<-playbackDone
|
||||
if err := <-playbackDone; err != nil && !errors.Is(err, context.Canceled) {
|
||||
log.Printf("playback controller: %v", err)
|
||||
}
|
||||
if err := player.Close(); err != nil {
|
||||
log.Printf("close playback: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"log"
|
||||
|
||||
mxladapter "mxl-player/internal/adapter/mxl"
|
||||
"mxl-player/internal/output"
|
||||
"mxl-player/internal/playback"
|
||||
)
|
||||
|
||||
type playerPlayback struct {
|
||||
Controller *playback.SessionController
|
||||
Commands chan playback.SessionCommand
|
||||
Video *playback.VideoBridge
|
||||
Status *playback.StatusStore
|
||||
Audio *output.SDLAudioSink
|
||||
}
|
||||
|
||||
func newPlayerPlayback(
|
||||
audioDevice uint32,
|
||||
retry playback.RetryPolicy,
|
||||
) (*playerPlayback, error) {
|
||||
videoBridge := playback.NewVideoBridge()
|
||||
statusStore := playback.NewStatusStore()
|
||||
audioSink := output.NewSDLAudioSink(audioDevice)
|
||||
|
||||
observe := func(status playback.Status) {
|
||||
statusStore.Observe(status)
|
||||
|
||||
if status.Err != nil {
|
||||
log.Printf(
|
||||
"%s: state=%v attempt=%d failed=%d: %v",
|
||||
status.Unit,
|
||||
status.State,
|
||||
status.Attempt,
|
||||
status.FailedAttempts,
|
||||
status.Err,
|
||||
)
|
||||
return
|
||||
}
|
||||
|
||||
log.Printf(
|
||||
"%s: state=%v attempt=%d failed=%d",
|
||||
status.Unit,
|
||||
status.State,
|
||||
status.Attempt,
|
||||
status.FailedAttempts,
|
||||
)
|
||||
}
|
||||
|
||||
videoWorker, err := playback.NewVideoWorker(
|
||||
mxladapter.VideoFactory{},
|
||||
videoBridge,
|
||||
retry,
|
||||
mxladapter.ShouldRetry,
|
||||
observe,
|
||||
)
|
||||
if err != nil {
|
||||
_ = audioSink.Close()
|
||||
return nil, err
|
||||
}
|
||||
videoSlot, err := playback.NewVideoSlot(videoWorker)
|
||||
if err != nil {
|
||||
_ = audioSink.Close()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
audioWorker, err := playback.NewAudioWorker(
|
||||
mxladapter.AudioFactory{},
|
||||
audioSink,
|
||||
retry,
|
||||
mxladapter.ShouldRetry,
|
||||
observe,
|
||||
)
|
||||
if err != nil {
|
||||
_ = audioSink.Close()
|
||||
return nil, err
|
||||
}
|
||||
audioSlot, err := playback.NewAudioSlot(audioWorker)
|
||||
if err != nil {
|
||||
_ = audioSink.Close()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
syncWorker, err := playback.NewSyncWorker(
|
||||
mxladapter.SyncFactory{},
|
||||
videoBridge,
|
||||
audioSink,
|
||||
retry,
|
||||
mxladapter.ShouldRetry,
|
||||
observe,
|
||||
)
|
||||
if err != nil {
|
||||
_ = audioSink.Close()
|
||||
return nil, err
|
||||
}
|
||||
syncSlot, err := playback.NewSyncSlot(syncWorker)
|
||||
if err != nil {
|
||||
_ = audioSink.Close()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
controller, err := playback.NewSessionController(
|
||||
videoSlot,
|
||||
audioSlot,
|
||||
syncSlot,
|
||||
func(video, audio playback.FeedConfig) bool {
|
||||
return video.Domain == audio.Domain
|
||||
},
|
||||
)
|
||||
if err != nil {
|
||||
_ = audioSink.Close()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &playerPlayback{
|
||||
Controller: controller,
|
||||
Commands: make(chan playback.SessionCommand, 32),
|
||||
Video: videoBridge,
|
||||
Status: statusStore,
|
||||
Audio: audioSink,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (p *playerPlayback) Close() error {
|
||||
return p.Audio.Close()
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"mxl-player/internal/playback"
|
||||
)
|
||||
|
||||
func validRuntimeRetryPolicy() playback.RetryPolicy {
|
||||
return playback.RetryPolicy{
|
||||
MaxAttempts: 3,
|
||||
InitialDelay: time.Millisecond,
|
||||
MaxDelay: time.Second,
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewPlayerPlaybackBuildsCompleteRuntime(t *testing.T) {
|
||||
runtime, err := newPlayerPlayback(123, validRuntimeRetryPolicy())
|
||||
if err != nil {
|
||||
t.Fatalf("newPlayerPlayback() error = %v", err)
|
||||
}
|
||||
if runtime.Controller == nil || runtime.Video == nil || runtime.Status == nil || runtime.Audio == nil {
|
||||
t.Fatalf("newPlayerPlayback() = %#v", runtime)
|
||||
}
|
||||
if runtime.Commands == nil || cap(runtime.Commands) != 32 {
|
||||
t.Fatalf("command channel = %#v, capacity = %d", runtime.Commands, cap(runtime.Commands))
|
||||
}
|
||||
if err := runtime.Close(); err != nil {
|
||||
t.Fatalf("Close() error = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewPlayerPlaybackRejectsInvalidRetryPolicy(t *testing.T) {
|
||||
runtime, err := newPlayerPlayback(123, playback.RetryPolicy{})
|
||||
if runtime != nil {
|
||||
t.Fatalf("newPlayerPlayback() runtime = %#v, want nil", runtime)
|
||||
}
|
||||
if err == nil {
|
||||
t.Fatal("newPlayerPlayback() error is nil")
|
||||
}
|
||||
}
|
||||
@@ -14,7 +14,7 @@ Size=200,200
|
||||
Collapsed=0
|
||||
|
||||
[Window][Connection]
|
||||
Pos=475,529
|
||||
Size=640,352
|
||||
Pos=322,387
|
||||
Size=618,275
|
||||
Collapsed=0
|
||||
|
||||
|
||||
@@ -14,7 +14,6 @@ import (
|
||||
|
||||
const (
|
||||
DefaultSyncReadTimeout = 200 * time.Millisecond
|
||||
DefaultSyncBatchDuration = 10 * time.Millisecond
|
||||
)
|
||||
|
||||
var ErrNativeSyncDifferentDomains = errors.New(
|
||||
@@ -23,14 +22,12 @@ var ErrNativeSyncDifferentDomains = errors.New(
|
||||
|
||||
type SyncFactory struct {
|
||||
ReadTimeout time.Duration
|
||||
BatchDuration time.Duration
|
||||
open func(string, string, string) (localSyncSource, error)
|
||||
}
|
||||
|
||||
type localSyncSource interface {
|
||||
NextSync(
|
||||
context.Context,
|
||||
uint64,
|
||||
time.Duration,
|
||||
) (source.Frame, source.AudioFrame, error)
|
||||
|
||||
@@ -41,7 +38,6 @@ type localSyncSource interface {
|
||||
type syncReader struct {
|
||||
source localSyncSource
|
||||
readTimeout time.Duration
|
||||
audioBatch uint64
|
||||
rateNumerator int64
|
||||
rateDenominator int64
|
||||
}
|
||||
@@ -118,25 +114,11 @@ func (f SyncFactory) OpenSync(
|
||||
if readTimeout <= 0 {
|
||||
readTimeout = DefaultSyncReadTimeout
|
||||
}
|
||||
batchDuration := f.BatchDuration
|
||||
if batchDuration <= 0 {
|
||||
batchDuration = DefaultSyncBatchDuration
|
||||
}
|
||||
audioRate := src.AudioRate()
|
||||
batch, err := audioBatchSize(audioRate.Num, audioRate.Den, batchDuration)
|
||||
if err != nil {
|
||||
_ = src.Close()
|
||||
return nil, &source.SourceError{
|
||||
Op: "calculate sync audio batch",
|
||||
Kind: source.ErrorKindInvalidConfig,
|
||||
Err: err,
|
||||
}
|
||||
}
|
||||
|
||||
return &syncReader{
|
||||
source: src,
|
||||
readTimeout: readTimeout,
|
||||
audioBatch: batch,
|
||||
rateNumerator: audioRate.Num,
|
||||
rateDenominator: audioRate.Den,
|
||||
}, nil
|
||||
@@ -145,7 +127,7 @@ func (f SyncFactory) OpenSync(
|
||||
func (r *syncReader) ReadSync(
|
||||
ctx context.Context,
|
||||
) (playback.SyncFrame, error) {
|
||||
video, audio, err := r.source.NextSync(ctx, r.audioBatch, r.readTimeout)
|
||||
video, audio, err := r.source.NextSync(ctx, r.readTimeout)
|
||||
if err != nil {
|
||||
return playback.SyncFrame{}, err
|
||||
}
|
||||
|
||||
@@ -17,7 +17,6 @@ type fakeLocalSyncSource struct {
|
||||
audio source.AudioFrame
|
||||
readErr error
|
||||
rate mxl.Rational
|
||||
batch uint64
|
||||
timeout time.Duration
|
||||
closed bool
|
||||
closeError error
|
||||
@@ -25,10 +24,8 @@ type fakeLocalSyncSource struct {
|
||||
|
||||
func (s *fakeLocalSyncSource) NextSync(
|
||||
_ context.Context,
|
||||
batch uint64,
|
||||
timeout time.Duration,
|
||||
) (source.Frame, source.AudioFrame, error) {
|
||||
s.batch = batch
|
||||
s.timeout = timeout
|
||||
return s.video, s.audio, s.readErr
|
||||
}
|
||||
@@ -79,23 +76,8 @@ func TestSyncFactoryUsesDefaultsAndForwardsFeeds(t *testing.T) {
|
||||
t.Fatalf("open args = %q %q %q", domain, videoUUID, audioUUID)
|
||||
}
|
||||
got := reader.(*syncReader)
|
||||
if got.readTimeout != DefaultSyncReadTimeout || got.audioBatch != 480 {
|
||||
t.Fatalf("reader timeout=%s batch=%d, want %s and 480", got.readTimeout, got.audioBatch, DefaultSyncReadTimeout)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncFactoryClosesSourceForInvalidAudioRate(t *testing.T) {
|
||||
fake := &fakeLocalSyncSource{rate: mxl.Rational{}}
|
||||
factory := SyncFactory{open: func(string, string, string) (localSyncSource, error) {
|
||||
return fake, nil
|
||||
}}
|
||||
video, audio := syncFeedConfigs()
|
||||
reader, err := factory.OpenSync(context.Background(), video, audio)
|
||||
if reader != nil {
|
||||
t.Fatal("OpenSync() reader is not nil")
|
||||
}
|
||||
if !errors.Is(err, ErrInvalidAudioBatch) || !fake.closed {
|
||||
t.Fatalf("OpenSync() error=%v closed=%t", err, fake.closed)
|
||||
if got.readTimeout != DefaultSyncReadTimeout {
|
||||
t.Fatalf("reader timeout=%s, want %s", got.readTimeout, DefaultSyncReadTimeout)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -108,7 +90,7 @@ func TestSyncReaderConvertsPairWithoutCopying(t *testing.T) {
|
||||
rate: mxl.Rational{Num: 48_000, Den: 1},
|
||||
}
|
||||
reader := &syncReader{
|
||||
source: fake, readTimeout: 7 * time.Millisecond, audioBatch: 12,
|
||||
source: fake, readTimeout: 7 * time.Millisecond,
|
||||
rateNumerator: 48_000, rateDenominator: 1,
|
||||
}
|
||||
|
||||
@@ -116,8 +98,8 @@ func TestSyncReaderConvertsPairWithoutCopying(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if fake.batch != 12 || fake.timeout != 7*time.Millisecond {
|
||||
t.Fatalf("NextSync() batch=%d timeout=%s", fake.batch, fake.timeout)
|
||||
if fake.timeout != 7*time.Millisecond {
|
||||
t.Fatalf("NextSync() timeout=%s", fake.timeout)
|
||||
}
|
||||
if frame.Video.Index != 10 || frame.Audio.Index != 40 || frame.Audio.SampleRateNumerator != 48_000 {
|
||||
t.Fatalf("frame = %+v", frame)
|
||||
|
||||
@@ -0,0 +1,344 @@
|
||||
package playback
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type VideoSlotRunner interface {
|
||||
Run(
|
||||
context.Context,
|
||||
FeedConfig,
|
||||
<-chan FeedConfig,
|
||||
) error
|
||||
}
|
||||
|
||||
type AudioSlotRunner interface {
|
||||
Run(
|
||||
context.Context,
|
||||
FeedConfig,
|
||||
<-chan FeedConfig,
|
||||
) error
|
||||
}
|
||||
|
||||
type SyncSlotRunner interface {
|
||||
Run(
|
||||
context.Context,
|
||||
SyncPairConfig,
|
||||
<-chan SyncPairConfig,
|
||||
) error
|
||||
}
|
||||
|
||||
var _ VideoSlotRunner = (*VideoSlot)(nil)
|
||||
var _ AudioSlotRunner = (*AudioSlot)(nil)
|
||||
var _ SyncSlotRunner = (*SyncSlot)(nil)
|
||||
|
||||
var (
|
||||
ErrVideoSlotRequired = errors.New("video slot is required")
|
||||
ErrAudioSlotRequired = errors.New("audio slot is required")
|
||||
ErrSyncSlotRequired = errors.New("sync slot is required")
|
||||
)
|
||||
|
||||
type SessionSnapshot struct {
|
||||
Desired SessionConfig
|
||||
Plan SessionPlan
|
||||
Generation uint64
|
||||
}
|
||||
|
||||
type SessionController struct {
|
||||
videoSlot VideoSlotRunner
|
||||
audioSlot AudioSlotRunner
|
||||
syncSlot SyncSlotRunner
|
||||
canSync SyncPredicate
|
||||
|
||||
mu sync.RWMutex
|
||||
snapshot SessionSnapshot
|
||||
hasSnapshot bool
|
||||
}
|
||||
|
||||
func NewSessionController(
|
||||
videoSlot VideoSlotRunner,
|
||||
audioSlot AudioSlotRunner,
|
||||
syncSlot SyncSlotRunner,
|
||||
canSync SyncPredicate,
|
||||
) (*SessionController, error) {
|
||||
if videoSlot == nil {
|
||||
return nil, ErrVideoSlotRequired
|
||||
}
|
||||
if audioSlot == nil {
|
||||
return nil, ErrAudioSlotRequired
|
||||
}
|
||||
if syncSlot == nil {
|
||||
return nil, ErrSyncSlotRequired
|
||||
}
|
||||
return &SessionController{
|
||||
videoSlot: videoSlot,
|
||||
audioSlot: audioSlot,
|
||||
syncSlot: syncSlot,
|
||||
canSync: canSync,
|
||||
}, nil
|
||||
}
|
||||
|
||||
type sessionRuntime struct {
|
||||
topology SessionTopology
|
||||
cancel context.CancelFunc
|
||||
done chan struct{}
|
||||
result error
|
||||
|
||||
videoCommands chan FeedConfig
|
||||
audioCommands chan FeedConfig
|
||||
syncCommands chan SyncPairConfig
|
||||
}
|
||||
|
||||
var (
|
||||
ErrSessionRuntimeStopped = errors.New(
|
||||
"session slot runtime stopped unexpectedly",
|
||||
)
|
||||
)
|
||||
|
||||
func (c *SessionController) Run(
|
||||
ctx context.Context,
|
||||
initial SessionConfig,
|
||||
commands <-chan SessionCommand,
|
||||
) error {
|
||||
plan, err := BuildSessionPlan(initial, c.canSync)
|
||||
if err != nil {
|
||||
return fmt.Errorf("build initial session plan: %w", err)
|
||||
}
|
||||
|
||||
desired := initial
|
||||
runtime := c.startSessionRuntime(ctx, plan)
|
||||
|
||||
generation := uint64(1)
|
||||
c.publish(SessionSnapshot{
|
||||
Desired: initial,
|
||||
Plan: plan,
|
||||
Generation: generation,
|
||||
})
|
||||
|
||||
for {
|
||||
var runtimeDone <-chan struct{}
|
||||
if runtime != nil {
|
||||
runtimeDone = runtime.done
|
||||
}
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
stopSessionRuntime(runtime)
|
||||
return ctx.Err()
|
||||
|
||||
case <-runtimeDone:
|
||||
return unexpectedSessionRuntimeError(runtime.result)
|
||||
|
||||
case command, ok := <-commands:
|
||||
if !ok {
|
||||
stopSessionRuntime(runtime)
|
||||
return nil
|
||||
}
|
||||
|
||||
nextDesired, err := ApplySessionCommand(desired, command)
|
||||
if err != nil {
|
||||
// Invalid commands must not disturb the current valid runtime.
|
||||
continue
|
||||
}
|
||||
nextPlan, err := BuildSessionPlan(nextDesired, c.canSync)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
nextRuntime, err := c.reconcileSessionRuntime(
|
||||
ctx,
|
||||
runtime,
|
||||
plan,
|
||||
nextPlan,
|
||||
)
|
||||
if err != nil {
|
||||
stopSessionRuntime(runtime)
|
||||
if ctx.Err() != nil {
|
||||
return ctx.Err()
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
if plan.Topology != nextPlan.Topology {
|
||||
generation++
|
||||
}
|
||||
desired = nextDesired
|
||||
plan = nextPlan
|
||||
runtime = nextRuntime
|
||||
c.publish(SessionSnapshot{
|
||||
Desired: desired,
|
||||
Plan: plan,
|
||||
Generation: generation,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *SessionController) startSessionRuntime(
|
||||
ctx context.Context,
|
||||
plan SessionPlan,
|
||||
) *sessionRuntime {
|
||||
if plan.Topology == TopologyIdle {
|
||||
return &sessionRuntime{topology: TopologyIdle}
|
||||
}
|
||||
|
||||
runtimeCtx, cancel := context.WithCancel(ctx)
|
||||
runtime := &sessionRuntime{
|
||||
topology: plan.Topology,
|
||||
cancel: cancel,
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
|
||||
switch plan.Topology {
|
||||
case TopologyIndependent:
|
||||
runtime.videoCommands = make(chan FeedConfig)
|
||||
runtime.audioCommands = make(chan FeedConfig)
|
||||
results := make(chan error, 2)
|
||||
|
||||
go func() {
|
||||
results <- c.videoSlot.Run(
|
||||
runtimeCtx,
|
||||
plan.Video,
|
||||
runtime.videoCommands,
|
||||
)
|
||||
}()
|
||||
go func() {
|
||||
results <- c.audioSlot.Run(
|
||||
runtimeCtx,
|
||||
plan.Audio,
|
||||
runtime.audioCommands,
|
||||
)
|
||||
}()
|
||||
go func() {
|
||||
first := <-results
|
||||
cancel()
|
||||
second := <-results
|
||||
runtime.result = errors.Join(first, second)
|
||||
close(runtime.done)
|
||||
}()
|
||||
|
||||
case TopologySynchronized:
|
||||
runtime.syncCommands = make(chan SyncPairConfig)
|
||||
go func() {
|
||||
runtime.result = c.syncSlot.Run(
|
||||
runtimeCtx,
|
||||
plan.Sync,
|
||||
runtime.syncCommands,
|
||||
)
|
||||
close(runtime.done)
|
||||
}()
|
||||
}
|
||||
|
||||
return runtime
|
||||
}
|
||||
|
||||
func stopSessionRuntime(runtime *sessionRuntime) {
|
||||
if runtime == nil || runtime.done == nil {
|
||||
return
|
||||
}
|
||||
runtime.cancel()
|
||||
<-runtime.done
|
||||
}
|
||||
|
||||
func unexpectedSessionRuntimeError(err error) error {
|
||||
if err == nil {
|
||||
return ErrSessionRuntimeStopped
|
||||
}
|
||||
return fmt.Errorf("%w: %v", ErrSessionRuntimeStopped, err)
|
||||
}
|
||||
|
||||
func (c *SessionController) reconcileSessionRuntime(
|
||||
ctx context.Context,
|
||||
runtime *sessionRuntime,
|
||||
current SessionPlan,
|
||||
next SessionPlan,
|
||||
) (*sessionRuntime, error) {
|
||||
if current.Topology != next.Topology {
|
||||
stopSessionRuntime(runtime)
|
||||
if err := ctx.Err(); err != nil {
|
||||
return runtime, err
|
||||
}
|
||||
return c.startSessionRuntime(ctx, next), nil
|
||||
}
|
||||
|
||||
switch next.Topology {
|
||||
case TopologyIndependent:
|
||||
if current.Video != next.Video {
|
||||
if !sendFeedConfig(ctx, runtime.done, runtime.videoCommands, next.Video) {
|
||||
return runtime, sessionRuntimeSendError(ctx, runtime)
|
||||
}
|
||||
}
|
||||
if current.Audio != next.Audio {
|
||||
if !sendFeedConfig(ctx, runtime.done, runtime.audioCommands, next.Audio) {
|
||||
return runtime, sessionRuntimeSendError(ctx, runtime)
|
||||
}
|
||||
}
|
||||
|
||||
case TopologySynchronized:
|
||||
if current.Sync != next.Sync {
|
||||
if !sendSyncPairConfig(ctx, runtime.done, runtime.syncCommands, next.Sync) {
|
||||
return runtime, sessionRuntimeSendError(ctx, runtime)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return runtime, nil
|
||||
}
|
||||
|
||||
func sendFeedConfig(
|
||||
ctx context.Context,
|
||||
done <-chan struct{},
|
||||
commands chan<- FeedConfig,
|
||||
config FeedConfig,
|
||||
) bool {
|
||||
select {
|
||||
case commands <- config:
|
||||
return true
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
case <-done:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func sendSyncPairConfig(
|
||||
ctx context.Context,
|
||||
done <-chan struct{},
|
||||
commands chan<- SyncPairConfig,
|
||||
config SyncPairConfig,
|
||||
) bool {
|
||||
select {
|
||||
case commands <- config:
|
||||
return true
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
case <-done:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func sessionRuntimeSendError(ctx context.Context, runtime *sessionRuntime) error {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
<-runtime.done
|
||||
return unexpectedSessionRuntimeError(runtime.result)
|
||||
}
|
||||
|
||||
func (c *SessionController) Snapshot() (SessionSnapshot, bool) {
|
||||
c.mu.RLock()
|
||||
snapshot := c.snapshot
|
||||
ok := c.hasSnapshot
|
||||
c.mu.RUnlock()
|
||||
return snapshot, ok
|
||||
}
|
||||
|
||||
func (c *SessionController) publish(snapshot SessionSnapshot) {
|
||||
c.mu.Lock()
|
||||
c.snapshot = snapshot
|
||||
c.hasSnapshot = true
|
||||
c.mu.Unlock()
|
||||
}
|
||||
@@ -0,0 +1,422 @@
|
||||
package playback
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
type stubVideoSlot struct{}
|
||||
|
||||
func (stubVideoSlot) Run(context.Context, FeedConfig, <-chan FeedConfig) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
type stubAudioSlot struct{}
|
||||
|
||||
func (stubAudioSlot) Run(context.Context, FeedConfig, <-chan FeedConfig) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
type stubSyncSlot struct{}
|
||||
|
||||
func (stubSyncSlot) Run(context.Context, SyncPairConfig, <-chan SyncPairConfig) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestNewSessionControllerValidatesSlots(t *testing.T) {
|
||||
video := stubVideoSlot{}
|
||||
audio := stubAudioSlot{}
|
||||
sync := stubSyncSlot{}
|
||||
tests := []struct {
|
||||
name string
|
||||
video VideoSlotRunner
|
||||
audio AudioSlotRunner
|
||||
sync SyncSlotRunner
|
||||
want error
|
||||
}{
|
||||
{"missing video", nil, audio, sync, ErrVideoSlotRequired},
|
||||
{"missing audio", video, nil, sync, ErrAudioSlotRequired},
|
||||
{"missing sync", video, audio, nil, ErrSyncSlotRequired},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
controller, err := NewSessionController(tt.video, tt.audio, tt.sync, nil)
|
||||
if controller != nil {
|
||||
t.Fatalf("NewSessionController() controller = %#v, want nil", controller)
|
||||
}
|
||||
if !errors.Is(err, tt.want) {
|
||||
t.Fatalf("NewSessionController() error = %v, want %v", err, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewSessionControllerAllowsUnavailableSynchronization(t *testing.T) {
|
||||
video := stubVideoSlot{}
|
||||
audio := stubAudioSlot{}
|
||||
sync := stubSyncSlot{}
|
||||
controller, err := NewSessionController(video, audio, sync, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("NewSessionController() error = %v", err)
|
||||
}
|
||||
if controller == nil {
|
||||
t.Fatal("NewSessionController() controller is nil")
|
||||
}
|
||||
if controller.videoSlot != video || controller.audioSlot != audio || controller.syncSlot != sync {
|
||||
t.Fatalf("NewSessionController() = %#v", controller)
|
||||
}
|
||||
if controller.canSync != nil {
|
||||
t.Fatal("nil sync predicate was not preserved")
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewSessionControllerStoresSyncPredicate(t *testing.T) {
|
||||
predicate := func(FeedConfig, FeedConfig) bool { return true }
|
||||
controller, err := NewSessionController(
|
||||
stubVideoSlot{}, stubAudioSlot{}, stubSyncSlot{}, predicate,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("NewSessionController() error = %v", err)
|
||||
}
|
||||
if controller.canSync == nil || !controller.canSync(FeedConfig{}, FeedConfig{}) {
|
||||
t.Fatal("sync predicate was not stored")
|
||||
}
|
||||
}
|
||||
|
||||
type controllerEvent struct {
|
||||
unit Unit
|
||||
action string
|
||||
feed FeedConfig
|
||||
pair SyncPairConfig
|
||||
}
|
||||
|
||||
type recordingVideoSlot struct{ events chan<- controllerEvent }
|
||||
|
||||
func (s recordingVideoSlot) Run(
|
||||
ctx context.Context,
|
||||
initial FeedConfig,
|
||||
commands <-chan FeedConfig,
|
||||
) error {
|
||||
s.events <- controllerEvent{unit: UnitVideo, action: "start", feed: initial}
|
||||
for {
|
||||
select {
|
||||
case config := <-commands:
|
||||
s.events <- controllerEvent{unit: UnitVideo, action: "command", feed: config}
|
||||
case <-ctx.Done():
|
||||
s.events <- controllerEvent{unit: UnitVideo, action: "stop"}
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type recordingAudioSlot struct{ events chan<- controllerEvent }
|
||||
|
||||
func (s recordingAudioSlot) Run(
|
||||
ctx context.Context,
|
||||
initial FeedConfig,
|
||||
commands <-chan FeedConfig,
|
||||
) error {
|
||||
s.events <- controllerEvent{unit: UnitAudio, action: "start", feed: initial}
|
||||
for {
|
||||
select {
|
||||
case config := <-commands:
|
||||
s.events <- controllerEvent{unit: UnitAudio, action: "command", feed: config}
|
||||
case <-ctx.Done():
|
||||
s.events <- controllerEvent{unit: UnitAudio, action: "stop"}
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type recordingSyncSlot struct{ events chan<- controllerEvent }
|
||||
|
||||
func (s recordingSyncSlot) Run(
|
||||
ctx context.Context,
|
||||
initial SyncPairConfig,
|
||||
commands <-chan SyncPairConfig,
|
||||
) error {
|
||||
s.events <- controllerEvent{unit: UnitSync, action: "start", pair: initial}
|
||||
for {
|
||||
select {
|
||||
case config := <-commands:
|
||||
s.events <- controllerEvent{unit: UnitSync, action: "command", pair: config}
|
||||
case <-ctx.Done():
|
||||
s.events <- controllerEvent{unit: UnitSync, action: "stop"}
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func newRecordingController(t *testing.T, events chan<- controllerEvent) *SessionController {
|
||||
t.Helper()
|
||||
controller, err := NewSessionController(
|
||||
recordingVideoSlot{events},
|
||||
recordingAudioSlot{events},
|
||||
recordingSyncSlot{events},
|
||||
func(FeedConfig, FeedConfig) bool { return true },
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return controller
|
||||
}
|
||||
|
||||
func receiveControllerEvent(t *testing.T, events <-chan controllerEvent) controllerEvent {
|
||||
t.Helper()
|
||||
select {
|
||||
case event := <-events:
|
||||
return event
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("controller event timed out")
|
||||
return controllerEvent{}
|
||||
}
|
||||
}
|
||||
|
||||
func receiveIndependentStarts(t *testing.T, events <-chan controllerEvent) {
|
||||
t.Helper()
|
||||
seen := map[Unit]bool{}
|
||||
for len(seen) < 2 {
|
||||
event := receiveControllerEvent(t, events)
|
||||
if event.action != "start" || (event.unit != UnitVideo && event.unit != UnitAudio) {
|
||||
t.Fatalf("unexpected initial event: %+v", event)
|
||||
}
|
||||
seen[event.unit] = true
|
||||
}
|
||||
}
|
||||
|
||||
func waitControllerSnapshot(
|
||||
t *testing.T,
|
||||
controller *SessionController,
|
||||
match func(SessionSnapshot) bool,
|
||||
) SessionSnapshot {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(time.Second)
|
||||
for {
|
||||
if snapshot, ok := controller.Snapshot(); ok && match(snapshot) {
|
||||
return snapshot
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
snapshot, ok := controller.Snapshot()
|
||||
t.Fatalf("snapshot timed out: %#v, available=%t", snapshot, ok)
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSessionControllerUpdatesOnlyChangedIndependentSlot(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)
|
||||
|
||||
want := FeedConfig{Domain: "/new-video", UUID: "new-video", Active: true}
|
||||
commands <- SessionCommand{Kind: CommandSetVideo, Config: want}
|
||||
event := receiveControllerEvent(t, events)
|
||||
if event.unit != UnitVideo || event.action != "command" || event.feed != want {
|
||||
t.Fatalf("replacement event = %+v", event)
|
||||
}
|
||||
select {
|
||||
case event := <-events:
|
||||
t.Fatalf("unchanged audio slot was disturbed: %+v", event)
|
||||
case <-time.After(20 * time.Millisecond):
|
||||
}
|
||||
|
||||
close(commands)
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
t.Fatalf("Run() error = %v, want nil", err)
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("Run() did not stop")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSessionControllerStopsIndependentSlotsBeforeStartingSync(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: CommandEnableSync}
|
||||
stopped := map[Unit]bool{}
|
||||
for {
|
||||
event := receiveControllerEvent(t, events)
|
||||
if event.unit == UnitSync && event.action == "start" {
|
||||
if !stopped[UnitVideo] || !stopped[UnitAudio] {
|
||||
t.Fatalf("sync started before both independent slots stopped: %v", stopped)
|
||||
}
|
||||
break
|
||||
}
|
||||
if event.action != "stop" || (event.unit != UnitVideo && event.unit != UnitAudio) {
|
||||
t.Fatalf("unexpected transition event: %+v", event)
|
||||
}
|
||||
stopped[event.unit] = true
|
||||
}
|
||||
|
||||
close(commands)
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
t.Fatalf("Run() error = %v, want nil", err)
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("Run() did not stop")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSessionControllerCancellationStopsAndJoinsRuntime(t *testing.T) {
|
||||
events := make(chan controllerEvent, 32)
|
||||
controller := newRecordingController(t, events)
|
||||
initial := validCommandSession()
|
||||
initial.SyncRequested = false
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- controller.Run(ctx, initial, make(chan SessionCommand)) }()
|
||||
receiveIndependentStarts(t, events)
|
||||
cancel()
|
||||
|
||||
select {
|
||||
case err := <-done:
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("Run() error = %v, want context.Canceled", err)
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("Run() did not stop after cancellation")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSessionControllerSnapshotTracksDesiredPlanAndGeneration(t *testing.T) {
|
||||
events := make(chan controllerEvent, 64)
|
||||
controller := newRecordingController(t, events)
|
||||
if snapshot, ok := controller.Snapshot(); ok {
|
||||
t.Fatalf("Snapshot() before Run = %#v, true; want unavailable", snapshot)
|
||||
}
|
||||
|
||||
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)
|
||||
|
||||
snapshot := waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
|
||||
return snapshot.Desired == initial
|
||||
})
|
||||
if snapshot.Generation != 1 || snapshot.Plan.Topology != TopologyIndependent {
|
||||
t.Fatalf("initial snapshot = %#v", snapshot)
|
||||
}
|
||||
|
||||
newVideo := FeedConfig{Domain: "/new-video", UUID: "new-video", Active: true}
|
||||
commands <- SessionCommand{Kind: CommandSetVideo, Config: newVideo}
|
||||
receiveControllerEvent(t, events)
|
||||
snapshot = waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
|
||||
return snapshot.Desired.Video == newVideo
|
||||
})
|
||||
if snapshot.Generation != 1 || snapshot.Plan.Topology != TopologyIndependent {
|
||||
t.Fatalf("same-topology snapshot = %#v", snapshot)
|
||||
}
|
||||
|
||||
commands <- SessionCommand{Kind: CommandEnableSync}
|
||||
for {
|
||||
if event := receiveControllerEvent(t, events); event.unit == UnitSync && event.action == "start" {
|
||||
break
|
||||
}
|
||||
}
|
||||
snapshot = waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
|
||||
return snapshot.Plan.Topology == TopologySynchronized
|
||||
})
|
||||
if snapshot.Generation != 2 || !snapshot.Desired.SyncRequested {
|
||||
t.Fatalf("sync snapshot = %#v", snapshot)
|
||||
}
|
||||
|
||||
commands <- SessionCommand{Kind: CommandDisableSync}
|
||||
for {
|
||||
event := receiveControllerEvent(t, events)
|
||||
if event.action == "start" && (event.unit == UnitVideo || event.unit == UnitAudio) {
|
||||
break
|
||||
}
|
||||
}
|
||||
snapshot = waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
|
||||
return snapshot.Plan.Topology == TopologyIndependent && !snapshot.Desired.SyncRequested
|
||||
})
|
||||
if snapshot.Generation != 3 {
|
||||
t.Fatalf("independent snapshot generation = %d, want 3", snapshot.Generation)
|
||||
}
|
||||
|
||||
close(commands)
|
||||
if err := <-done; err != nil {
|
||||
t.Fatalf("Run() error = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSessionControllerSnapshotRetainsUnavailableSyncRequest(t *testing.T) {
|
||||
events := make(chan controllerEvent, 32)
|
||||
controller, err := NewSessionController(
|
||||
recordingVideoSlot{events},
|
||||
recordingAudioSlot{events},
|
||||
recordingSyncSlot{events},
|
||||
nil,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
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: CommandEnableSync}
|
||||
|
||||
snapshot := waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
|
||||
return snapshot.Desired.SyncRequested
|
||||
})
|
||||
if snapshot.Plan.Topology != TopologyIndependent || snapshot.Generation != 1 {
|
||||
t.Fatalf("unsupported-sync snapshot = %#v", snapshot)
|
||||
}
|
||||
close(commands)
|
||||
if err := <-done; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSessionControllerSnapshotConcurrentReads(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)
|
||||
waitControllerSnapshot(t, controller, func(SessionSnapshot) bool { return true })
|
||||
|
||||
var readers sync.WaitGroup
|
||||
for range 8 {
|
||||
readers.Add(1)
|
||||
go func() {
|
||||
defer readers.Done()
|
||||
for range 1_000 {
|
||||
controller.Snapshot()
|
||||
}
|
||||
}()
|
||||
}
|
||||
readers.Wait()
|
||||
close(commands)
|
||||
if err := <-done; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
package playback
|
||||
|
||||
import "fmt"
|
||||
|
||||
type SessionTopology uint8
|
||||
|
||||
const (
|
||||
TopologyIdle SessionTopology = iota
|
||||
TopologyIndependent
|
||||
TopologySynchronized
|
||||
)
|
||||
|
||||
type SyncPredicate func(video, audio FeedConfig) bool
|
||||
|
||||
type SessionPlan struct {
|
||||
Topology SessionTopology
|
||||
|
||||
// Desired states for independent slots.
|
||||
Video FeedConfig
|
||||
Audio FeedConfig
|
||||
|
||||
// Desired state for the synchronized slot.
|
||||
Sync SyncPairConfig
|
||||
}
|
||||
|
||||
func BuildSessionPlan(
|
||||
desired SessionConfig,
|
||||
canSync SyncPredicate,
|
||||
) (SessionPlan, error) {
|
||||
if err := desired.Validate(); err != nil {
|
||||
return SessionPlan{}, fmt.Errorf("validate desired session: %w", err)
|
||||
}
|
||||
|
||||
if desired.SyncRequested &&
|
||||
desired.Video.Active &&
|
||||
desired.Audio.Active &&
|
||||
canSync != nil &&
|
||||
canSync(desired.Video, desired.Audio) {
|
||||
return SessionPlan{
|
||||
Topology: TopologySynchronized,
|
||||
Video: stoppedFeed(desired.Video),
|
||||
Audio: stoppedFeed(desired.Audio),
|
||||
Sync: SyncPairConfig{
|
||||
Video: desired.Video,
|
||||
Audio: desired.Audio,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
if desired.Video.Active || desired.Audio.Active {
|
||||
return SessionPlan{
|
||||
Topology: TopologyIndependent,
|
||||
Video: desired.Video,
|
||||
Audio: desired.Audio,
|
||||
Sync: SyncPairConfig{
|
||||
Video: stoppedFeed(desired.Video),
|
||||
Audio: stoppedFeed(desired.Audio),
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
return SessionPlan{
|
||||
Topology: TopologyIdle,
|
||||
Video: stoppedFeed(desired.Video),
|
||||
Audio: stoppedFeed(desired.Audio),
|
||||
Sync: SyncPairConfig{
|
||||
Video: stoppedFeed(desired.Video),
|
||||
Audio: stoppedFeed(desired.Audio),
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func stoppedFeed(config FeedConfig) FeedConfig {
|
||||
config.Active = false
|
||||
return config
|
||||
}
|
||||
@@ -0,0 +1,130 @@
|
||||
package playback
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestBuildSessionPlan(t *testing.T) {
|
||||
base := validCommandSession()
|
||||
stoppedVideo := stoppedFeed(base.Video)
|
||||
stoppedAudio := stoppedFeed(base.Audio)
|
||||
independent := func(video, audio FeedConfig) SessionPlan {
|
||||
return SessionPlan{
|
||||
Topology: TopologyIndependent,
|
||||
Video: video,
|
||||
Audio: audio,
|
||||
Sync: SyncPairConfig{Video: stoppedFeed(video), Audio: stoppedFeed(audio)},
|
||||
}
|
||||
}
|
||||
idle := func(video, audio FeedConfig) SessionPlan {
|
||||
return SessionPlan{
|
||||
Topology: TopologyIdle,
|
||||
Video: stoppedFeed(video),
|
||||
Audio: stoppedFeed(audio),
|
||||
Sync: SyncPairConfig{Video: stoppedFeed(video), Audio: stoppedFeed(audio)},
|
||||
}
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
desired SessionConfig
|
||||
canSync SyncPredicate
|
||||
want SessionPlan
|
||||
}{
|
||||
{
|
||||
name: "both active with sync disabled stay independent",
|
||||
desired: func() SessionConfig { c := base; c.SyncRequested = false; return c }(),
|
||||
canSync: func(FeedConfig, FeedConfig) bool { return true },
|
||||
want: independent(base.Video, base.Audio),
|
||||
},
|
||||
{
|
||||
name: "eligible requested pair is synchronized",
|
||||
desired: base,
|
||||
canSync: func(FeedConfig, FeedConfig) bool { return true },
|
||||
want: SessionPlan{
|
||||
Topology: TopologySynchronized,
|
||||
Video: stoppedVideo,
|
||||
Audio: stoppedAudio,
|
||||
Sync: SyncPairConfig{Video: base.Video, Audio: base.Audio},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "ineligible requested pair remains independent",
|
||||
desired: base,
|
||||
canSync: func(FeedConfig, FeedConfig) bool { return false },
|
||||
want: independent(base.Video, base.Audio),
|
||||
},
|
||||
{
|
||||
name: "nil capability remains independent",
|
||||
desired: base,
|
||||
want: independent(base.Video, base.Audio),
|
||||
},
|
||||
{
|
||||
name: "video only",
|
||||
desired: func() SessionConfig { c := base; c.Audio.Active = false; return c }(),
|
||||
canSync: func(FeedConfig, FeedConfig) bool { return true },
|
||||
want: independent(base.Video, stoppedAudio),
|
||||
},
|
||||
{
|
||||
name: "audio only",
|
||||
desired: func() SessionConfig { c := base; c.Video.Active = false; return c }(),
|
||||
canSync: func(FeedConfig, FeedConfig) bool { return true },
|
||||
want: independent(stoppedVideo, base.Audio),
|
||||
},
|
||||
{
|
||||
name: "stopped feeds are idle and retain configuration",
|
||||
desired: func() SessionConfig { c := base; c.Video.Active = false; c.Audio.Active = false; return c }(),
|
||||
canSync: func(FeedConfig, FeedConfig) bool { return true },
|
||||
want: idle(stoppedVideo, stoppedAudio),
|
||||
},
|
||||
{
|
||||
name: "empty session is idle",
|
||||
desired: func() SessionConfig { c := base; c.Video = FeedConfig{}; c.Audio = FeedConfig{}; return c }(),
|
||||
canSync: func(FeedConfig, FeedConfig) bool { return true },
|
||||
want: idle(FeedConfig{}, FeedConfig{}),
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got, err := BuildSessionPlan(tt.desired, tt.canSync)
|
||||
if err != nil {
|
||||
t.Fatalf("BuildSessionPlan() error = %v", err)
|
||||
}
|
||||
if got != tt.want {
|
||||
t.Fatalf("BuildSessionPlan() = %#v, want %#v", got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildSessionPlanRejectsInvalidSession(t *testing.T) {
|
||||
desired := validCommandSession()
|
||||
desired.Video = FeedConfig{UUID: "video", Active: true}
|
||||
got, err := BuildSessionPlan(desired, nil)
|
||||
if !errors.Is(err, ErrFeedDomainRequired) {
|
||||
t.Fatalf("BuildSessionPlan() error = %v, want %v", err, ErrFeedDomainRequired)
|
||||
}
|
||||
if got != (SessionPlan{}) {
|
||||
t.Fatalf("BuildSessionPlan() = %#v, want zero plan", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildSessionPlanPassesCompleteFeedsToPredicate(t *testing.T) {
|
||||
desired := validCommandSession()
|
||||
called := false
|
||||
_, err := BuildSessionPlan(desired, func(video, audio FeedConfig) bool {
|
||||
called = true
|
||||
if video != desired.Video || audio != desired.Audio {
|
||||
t.Fatalf("predicate feeds = %#v %#v", video, audio)
|
||||
}
|
||||
return true
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !called {
|
||||
t.Fatal("sync predicate was not called")
|
||||
}
|
||||
}
|
||||
@@ -598,8 +598,10 @@ func (s *SyncSource) Close() error {
|
||||
return s.inst.Close()
|
||||
}
|
||||
|
||||
// NextSync reads both at a synced timestamp. Returns video Frame + audio AudioFrame
|
||||
func (s *SyncSource) NextSync(ctx context.Context, audioBatch uint64, timeout time.Duration) (Frame, AudioFrame, error) {
|
||||
// NextSync reads one video frame and the audio interval between this video
|
||||
// timestamp and the next. Deriving the interval for every frame preserves
|
||||
// exact long-term timing for fractional video rates.
|
||||
func (s *SyncSource) NextSync(ctx context.Context, timeout time.Duration) (Frame, AudioFrame, error) {
|
||||
var timeouts int
|
||||
for {
|
||||
select {
|
||||
@@ -635,7 +637,30 @@ func (s *SyncSource) NextSync(ctx context.Context, audioBatch uint64, timeout ti
|
||||
}
|
||||
// read audio at the same timestamp
|
||||
aIdx := mxl.TimestampToIndex(s.aRate, ts)
|
||||
nextTimestamp := mxl.IndexToTimestamp(s.rate, s.idx+1)
|
||||
nextAudioIndex := mxl.TimestampToIndex(s.aRate, nextTimestamp)
|
||||
if nextAudioIndex <= aIdx {
|
||||
return Frame{}, AudioFrame{}, wrapError(
|
||||
"calculate synchronized audio interval",
|
||||
ErrorKindInvalidConfig,
|
||||
fmt.Errorf("invalid audio interval: %d..%d", aIdx, nextAudioIndex),
|
||||
)
|
||||
}
|
||||
audioBatch := nextAudioIndex - aIdx
|
||||
av, aerr := s.ar.GetSamples(aIdx, int(audioBatch), 50*time.Millisecond)
|
||||
if aerr != nil {
|
||||
kind := ErrorKindUnavailable
|
||||
if errors.Is(aerr, mxl.ErrTimeout) ||
|
||||
errors.Is(aerr, mxl.ErrOutOfRangeEarly) ||
|
||||
errors.Is(aerr, mxl.ErrOutOfRangeLate) {
|
||||
kind = ErrorKindTemporary
|
||||
}
|
||||
return Frame{}, AudioFrame{}, wrapError(
|
||||
"read synchronized audio",
|
||||
kind,
|
||||
aerr,
|
||||
)
|
||||
}
|
||||
vFrame := Frame{
|
||||
Index: g.Index, Width: s.width, Height: s.height,
|
||||
Stride: s.stride, Size: g.GrainSize,
|
||||
@@ -643,8 +668,6 @@ func (s *SyncSource) NextSync(ctx context.Context, audioBatch uint64, timeout ti
|
||||
}
|
||||
s.idx++
|
||||
|
||||
var aFrame AudioFrame
|
||||
if aerr == nil {
|
||||
samples := make([][]byte, s.chans)
|
||||
for ch := uint64(0); ch < s.chans; ch++ {
|
||||
f1, f2, _ := av.ChannelFragments(ch)
|
||||
@@ -654,12 +677,10 @@ func (s *SyncSource) NextSync(ctx context.Context, audioBatch uint64, timeout ti
|
||||
samples[ch] = f1
|
||||
}
|
||||
}
|
||||
aFrame = AudioFrame{
|
||||
aFrame := AudioFrame{
|
||||
Index: aIdx, SampleCount: audioBatch,
|
||||
Channels: s.chans, Samples: samples,
|
||||
}
|
||||
}
|
||||
// even if audio failed, video returns
|
||||
return vFrame, aFrame, nil
|
||||
case errors.Is(err, mxl.ErrTimeout), errors.Is(err, mxl.ErrOutOfRangeEarly), errors.Is(err, mxl.ErrOutOfRangeLate):
|
||||
timeouts++
|
||||
|
||||
Reference in New Issue
Block a user