Compare commits
3 Commits
aac8c664d8
...
b741b7d01f
| Author | SHA1 | Date | |
|---|---|---|---|
 Dmitry Sergeev
|
b741b7d01f | ||
|
Dmitry Sergeev
|
dcc56ac18c | ||
|
Dmitry Sergeev
|
cd7e239891 |
@@ -14,7 +14,7 @@ Size=200,200
|
||||
Collapsed=0
|
||||
|
||||
[Window][Connection]
|
||||
Pos=177,324
|
||||
Size=640,354
|
||||
Pos=475,529
|
||||
Size=640,352
|
||||
Collapsed=0
|
||||
|
||||
|
||||
@@ -0,0 +1,175 @@
|
||||
package mxladapter
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"mxl-player/internal/playback"
|
||||
"mxl-player/internal/source"
|
||||
|
||||
mxl "github.com/qvest-digital/go-mxl/mxl"
|
||||
)
|
||||
|
||||
const (
|
||||
DefaultSyncReadTimeout = 200 * time.Millisecond
|
||||
DefaultSyncBatchDuration = 10 * time.Millisecond
|
||||
)
|
||||
|
||||
var ErrNativeSyncDifferentDomains = errors.New(
|
||||
"native MXL synchronization requires matching domains",
|
||||
)
|
||||
|
||||
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)
|
||||
|
||||
AudioRate() mxl.Rational
|
||||
Close() error
|
||||
}
|
||||
|
||||
type syncReader struct {
|
||||
source localSyncSource
|
||||
readTimeout time.Duration
|
||||
audioBatch uint64
|
||||
rateNumerator int64
|
||||
rateDenominator int64
|
||||
}
|
||||
|
||||
var _ playback.SyncReaderFactory = SyncFactory{}
|
||||
var _ playback.SyncReader = (*syncReader)(nil)
|
||||
|
||||
func (f SyncFactory) OpenSync(
|
||||
ctx context.Context,
|
||||
videoConfig playback.FeedConfig,
|
||||
audioConfig playback.FeedConfig,
|
||||
) (playback.SyncReader, error) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := videoConfig.Validate(); err != nil {
|
||||
return nil, &source.SourceError{
|
||||
Op: "validate sync video feed",
|
||||
Kind: source.ErrorKindInvalidConfig,
|
||||
Err: err,
|
||||
}
|
||||
}
|
||||
if err := audioConfig.Validate(); err != nil {
|
||||
return nil, &source.SourceError{
|
||||
Op: "validate sync audio feed",
|
||||
Kind: source.ErrorKindInvalidConfig,
|
||||
Err: err,
|
||||
}
|
||||
}
|
||||
if !videoConfig.IsConfigured() {
|
||||
return nil, &source.SourceError{
|
||||
Op: "validate sync video feed",
|
||||
Kind: source.ErrorKindInvalidConfig,
|
||||
Err: errors.New("sync video feed is not configured"),
|
||||
}
|
||||
}
|
||||
if !audioConfig.IsConfigured() {
|
||||
return nil, &source.SourceError{
|
||||
Op: "validate sync audio feed",
|
||||
Kind: source.ErrorKindInvalidConfig,
|
||||
Err: errors.New("sync audio feed is not configured"),
|
||||
}
|
||||
}
|
||||
|
||||
if videoConfig.Domain != audioConfig.Domain {
|
||||
return nil, &source.SourceError{
|
||||
Op: "validate native MXL sync group",
|
||||
Kind: source.ErrorKindInvalidConfig,
|
||||
Err: ErrNativeSyncDifferentDomains,
|
||||
}
|
||||
}
|
||||
|
||||
open := f.open
|
||||
if open == nil {
|
||||
open = func(domain, videoUUID, audioUUID string) (localSyncSource, error) {
|
||||
return source.OpenSameDomainSync(domain, videoUUID, audioUUID)
|
||||
}
|
||||
}
|
||||
src, err := open(
|
||||
videoConfig.Domain,
|
||||
videoConfig.UUID,
|
||||
audioConfig.UUID,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("open native MXL sync group: %w", err)
|
||||
}
|
||||
|
||||
if err := ctx.Err(); err != nil {
|
||||
_ = src.Close()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
readTimeout := f.ReadTimeout
|
||||
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
|
||||
}
|
||||
|
||||
func (r *syncReader) ReadSync(
|
||||
ctx context.Context,
|
||||
) (playback.SyncFrame, error) {
|
||||
video, audio, err := r.source.NextSync(ctx, r.audioBatch, r.readTimeout)
|
||||
if err != nil {
|
||||
return playback.SyncFrame{}, err
|
||||
}
|
||||
return playback.SyncFrame{
|
||||
Video: playback.VideoFrame{
|
||||
Index: video.Index,
|
||||
Width: video.Width,
|
||||
Height: video.Height,
|
||||
Stride: video.Stride,
|
||||
Size: video.Size,
|
||||
Invalid: video.Invalid,
|
||||
Payload: video.Payload,
|
||||
},
|
||||
Audio: playback.AudioFrame{
|
||||
Index: audio.Index,
|
||||
SampleCount: audio.SampleCount,
|
||||
Channels: audio.Channels,
|
||||
SampleRateNumerator: r.rateNumerator,
|
||||
SampleRateDenominator: r.rateDenominator,
|
||||
Samples: audio.Samples,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (r *syncReader) Close() error {
|
||||
return r.source.Close()
|
||||
}
|
||||
@@ -0,0 +1,143 @@
|
||||
package mxladapter
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"mxl-player/internal/playback"
|
||||
"mxl-player/internal/source"
|
||||
|
||||
mxl "github.com/qvest-digital/go-mxl/mxl"
|
||||
)
|
||||
|
||||
type fakeLocalSyncSource struct {
|
||||
video source.Frame
|
||||
audio source.AudioFrame
|
||||
readErr error
|
||||
rate mxl.Rational
|
||||
batch uint64
|
||||
timeout time.Duration
|
||||
closed bool
|
||||
closeError error
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
func (s *fakeLocalSyncSource) AudioRate() mxl.Rational { return s.rate }
|
||||
func (s *fakeLocalSyncSource) Close() error {
|
||||
s.closed = true
|
||||
return s.closeError
|
||||
}
|
||||
|
||||
func syncFeedConfigs() (playback.FeedConfig, playback.FeedConfig) {
|
||||
return playback.FeedConfig{Domain: "/mxl", UUID: "video", Active: true},
|
||||
playback.FeedConfig{Domain: "/mxl", UUID: "audio", Active: true}
|
||||
}
|
||||
|
||||
func TestSyncFactoryRejectsDifferentDomains(t *testing.T) {
|
||||
video, audio := syncFeedConfigs()
|
||||
audio.Domain = "/other"
|
||||
reader, err := (SyncFactory{}).OpenSync(context.Background(), video, audio)
|
||||
if reader != nil {
|
||||
t.Fatal("OpenSync() reader is not nil")
|
||||
}
|
||||
if !errors.Is(err, ErrNativeSyncDifferentDomains) {
|
||||
t.Fatalf("OpenSync() error = %v, want %v", err, ErrNativeSyncDifferentDomains)
|
||||
}
|
||||
if source.KindOf(err) != source.ErrorKindInvalidConfig {
|
||||
t.Fatalf("error kind = %v, want invalid config", source.KindOf(err))
|
||||
}
|
||||
if ShouldRetry(err) {
|
||||
t.Fatal("ShouldRetry() = true for different domains")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncFactoryUsesDefaultsAndForwardsFeeds(t *testing.T) {
|
||||
fake := &fakeLocalSyncSource{rate: mxl.Rational{Num: 48_000, Den: 1}}
|
||||
var domain, videoUUID, audioUUID string
|
||||
factory := SyncFactory{open: func(d, v, a string) (localSyncSource, error) {
|
||||
domain, videoUUID, audioUUID = d, v, a
|
||||
return fake, nil
|
||||
}}
|
||||
video, audio := syncFeedConfigs()
|
||||
|
||||
reader, err := factory.OpenSync(context.Background(), video, audio)
|
||||
if err != nil {
|
||||
t.Fatalf("OpenSync() error = %v", err)
|
||||
}
|
||||
if domain != video.Domain || videoUUID != video.UUID || audioUUID != audio.UUID {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncReaderConvertsPairWithoutCopying(t *testing.T) {
|
||||
videoPayload := []byte{1, 2, 3}
|
||||
audioSamples := [][]byte{{4, 5, 6, 7}}
|
||||
fake := &fakeLocalSyncSource{
|
||||
video: source.Frame{Index: 10, Width: 20, Height: 30, Payload: videoPayload},
|
||||
audio: source.AudioFrame{Index: 40, SampleCount: 1, Channels: 1, Samples: audioSamples},
|
||||
rate: mxl.Rational{Num: 48_000, Den: 1},
|
||||
}
|
||||
reader := &syncReader{
|
||||
source: fake, readTimeout: 7 * time.Millisecond, audioBatch: 12,
|
||||
rateNumerator: 48_000, rateDenominator: 1,
|
||||
}
|
||||
|
||||
frame, err := reader.ReadSync(context.Background())
|
||||
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 frame.Video.Index != 10 || frame.Audio.Index != 40 || frame.Audio.SampleRateNumerator != 48_000 {
|
||||
t.Fatalf("frame = %+v", frame)
|
||||
}
|
||||
if &frame.Video.Payload[0] != &videoPayload[0] || &frame.Audio.Samples[0][0] != &audioSamples[0][0] {
|
||||
t.Fatal("sync payload was copied")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncFactoryReturnsPreCanceledContextWithoutOpening(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
opened := false
|
||||
factory := SyncFactory{open: func(string, string, string) (localSyncSource, error) {
|
||||
opened = true
|
||||
return nil, nil
|
||||
}}
|
||||
video, audio := syncFeedConfigs()
|
||||
reader, err := factory.OpenSync(ctx, video, audio)
|
||||
if reader != nil || !errors.Is(err, context.Canceled) || opened {
|
||||
t.Fatalf("reader=%v error=%v opened=%t", reader, err, opened)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
package playback
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
type SessionCommandKind uint8
|
||||
|
||||
const (
|
||||
CommandSetVideo SessionCommandKind = iota + 1
|
||||
CommandSetAudio
|
||||
CommandStopVideo
|
||||
CommandStopAudio
|
||||
CommandStopAll
|
||||
CommandResumeVideo
|
||||
CommandResumeAudio
|
||||
CommandResumeAll
|
||||
CommandRemoveVideo
|
||||
CommandRemoveAudio
|
||||
CommandEnableSync
|
||||
CommandDisableSync
|
||||
)
|
||||
|
||||
type SessionCommand struct {
|
||||
Kind SessionCommandKind
|
||||
Config FeedConfig // Used only by SetVideo and SetAudio.
|
||||
}
|
||||
|
||||
var (
|
||||
ErrUnknownSessionCommand = errors.New("unknown session command")
|
||||
ErrVideoNotConfigured = errors.New("video feed is not configured")
|
||||
ErrAudioNotConfigured = errors.New("audio feed is not configured")
|
||||
)
|
||||
|
||||
func ApplySessionCommand(
|
||||
current SessionConfig,
|
||||
command SessionCommand,
|
||||
) (SessionConfig, error) {
|
||||
next := current
|
||||
switch command.Kind {
|
||||
case CommandSetVideo:
|
||||
if !command.Config.IsConfigured() {
|
||||
return current, ErrVideoNotConfigured
|
||||
}
|
||||
if err := command.Config.Validate(); err != nil {
|
||||
return current, fmt.Errorf("validate video command: %w", err)
|
||||
}
|
||||
next.Video = command.Config
|
||||
|
||||
case CommandSetAudio:
|
||||
if !command.Config.IsConfigured() {
|
||||
return current, ErrAudioNotConfigured
|
||||
}
|
||||
if err := command.Config.Validate(); err != nil {
|
||||
return current, fmt.Errorf("validate audio command: %w", err)
|
||||
}
|
||||
next.Audio = command.Config
|
||||
|
||||
case CommandStopVideo:
|
||||
next.Video.Active = false
|
||||
next.SyncRequested = false
|
||||
|
||||
case CommandStopAudio:
|
||||
next.Audio.Active = false
|
||||
next.SyncRequested = false
|
||||
|
||||
case CommandStopAll:
|
||||
next.Video.Active = false
|
||||
next.Audio.Active = false
|
||||
|
||||
case CommandResumeVideo:
|
||||
if !next.Video.IsConfigured() {
|
||||
return current, ErrVideoNotConfigured
|
||||
}
|
||||
next.Video.Active = true
|
||||
|
||||
case CommandResumeAudio:
|
||||
if !next.Audio.IsConfigured() {
|
||||
return current, ErrAudioNotConfigured
|
||||
}
|
||||
next.Audio.Active = true
|
||||
|
||||
case CommandResumeAll:
|
||||
next.Video.Active = next.Video.IsConfigured()
|
||||
next.Audio.Active = next.Audio.IsConfigured()
|
||||
|
||||
case CommandRemoveVideo:
|
||||
next.Video = FeedConfig{}
|
||||
next.SyncRequested = false
|
||||
|
||||
case CommandRemoveAudio:
|
||||
next.Audio = FeedConfig{}
|
||||
next.SyncRequested = false
|
||||
|
||||
case CommandEnableSync:
|
||||
next.SyncRequested = true
|
||||
|
||||
case CommandDisableSync:
|
||||
next.SyncRequested = false
|
||||
|
||||
default:
|
||||
return current, ErrUnknownSessionCommand
|
||||
}
|
||||
|
||||
if err := next.Validate(); err != nil {
|
||||
return current, fmt.Errorf("validate desired session: %w", err)
|
||||
}
|
||||
return next, nil
|
||||
}
|
||||
@@ -0,0 +1,185 @@
|
||||
package playback
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func validCommandSession() SessionConfig {
|
||||
return SessionConfig{
|
||||
Video: FeedConfig{Domain: "/video", UUID: "video", Active: true},
|
||||
Audio: FeedConfig{Domain: "/audio", UUID: "audio", Active: true},
|
||||
SyncRequested: true,
|
||||
Retry: RetryPolicy{
|
||||
MaxAttempts: 3,
|
||||
InitialDelay: time.Millisecond,
|
||||
MaxDelay: time.Second,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplySessionCommand(t *testing.T) {
|
||||
base := validCommandSession()
|
||||
newVideo := FeedConfig{Domain: "/new-video", UUID: "new-video", Active: false}
|
||||
newAudio := FeedConfig{Domain: "/new-audio", UUID: "new-audio", Active: true}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
current SessionConfig
|
||||
command SessionCommand
|
||||
want SessionConfig
|
||||
}{
|
||||
{
|
||||
name: "set video replaces complete config",
|
||||
current: base,
|
||||
command: SessionCommand{Kind: CommandSetVideo, Config: newVideo},
|
||||
want: func() SessionConfig { c := base; c.Video = newVideo; return c }(),
|
||||
},
|
||||
{
|
||||
name: "set audio replaces complete config",
|
||||
current: base,
|
||||
command: SessionCommand{Kind: CommandSetAudio, Config: newAudio},
|
||||
want: func() SessionConfig { c := base; c.Audio = newAudio; return c }(),
|
||||
},
|
||||
{
|
||||
name: "stop video disables sync",
|
||||
current: base,
|
||||
command: SessionCommand{Kind: CommandStopVideo},
|
||||
want: func() SessionConfig { c := base; c.Video.Active = false; c.SyncRequested = false; return c }(),
|
||||
},
|
||||
{
|
||||
name: "stop audio disables sync",
|
||||
current: base,
|
||||
command: SessionCommand{Kind: CommandStopAudio},
|
||||
want: func() SessionConfig { c := base; c.Audio.Active = false; c.SyncRequested = false; return c }(),
|
||||
},
|
||||
{
|
||||
name: "stop all preserves sync request",
|
||||
current: base,
|
||||
command: SessionCommand{Kind: CommandStopAll},
|
||||
want: func() SessionConfig { c := base; c.Video.Active = false; c.Audio.Active = false; return c }(),
|
||||
},
|
||||
{
|
||||
name: "resume video",
|
||||
current: func() SessionConfig { c := base; c.Video.Active = false; return c }(),
|
||||
command: SessionCommand{Kind: CommandResumeVideo},
|
||||
want: base,
|
||||
},
|
||||
{
|
||||
name: "resume audio",
|
||||
current: func() SessionConfig { c := base; c.Audio.Active = false; return c }(),
|
||||
command: SessionCommand{Kind: CommandResumeAudio},
|
||||
want: base,
|
||||
},
|
||||
{
|
||||
name: "resume all activates only configured feeds",
|
||||
current: func() SessionConfig { c := base; c.Video.Active = false; c.Audio = FeedConfig{}; return c }(),
|
||||
command: SessionCommand{Kind: CommandResumeAll},
|
||||
want: func() SessionConfig { c := base; c.Audio = FeedConfig{}; return c }(),
|
||||
},
|
||||
{
|
||||
name: "remove video clears config and disables sync",
|
||||
current: base,
|
||||
command: SessionCommand{Kind: CommandRemoveVideo},
|
||||
want: func() SessionConfig { c := base; c.Video = FeedConfig{}; c.SyncRequested = false; return c }(),
|
||||
},
|
||||
{
|
||||
name: "remove audio clears config and disables sync",
|
||||
current: base,
|
||||
command: SessionCommand{Kind: CommandRemoveAudio},
|
||||
want: func() SessionConfig { c := base; c.Audio = FeedConfig{}; c.SyncRequested = false; return c }(),
|
||||
},
|
||||
{
|
||||
name: "enable sync without feeds records request",
|
||||
current: func() SessionConfig {
|
||||
c := base
|
||||
c.Video = FeedConfig{}
|
||||
c.Audio = FeedConfig{}
|
||||
c.SyncRequested = false
|
||||
return c
|
||||
}(),
|
||||
command: SessionCommand{Kind: CommandEnableSync},
|
||||
want: func() SessionConfig {
|
||||
c := base
|
||||
c.Video = FeedConfig{}
|
||||
c.Audio = FeedConfig{}
|
||||
c.SyncRequested = true
|
||||
return c
|
||||
}(),
|
||||
},
|
||||
{
|
||||
name: "disable sync",
|
||||
current: base,
|
||||
command: SessionCommand{Kind: CommandDisableSync},
|
||||
want: func() SessionConfig { c := base; c.SyncRequested = false; return c }(),
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got, err := ApplySessionCommand(tt.current, tt.command)
|
||||
if err != nil {
|
||||
t.Fatalf("ApplySessionCommand() error = %v", err)
|
||||
}
|
||||
if got != tt.want {
|
||||
t.Fatalf("ApplySessionCommand() = %#v, want %#v", got, tt.want)
|
||||
}
|
||||
if got.Retry != tt.current.Retry {
|
||||
t.Fatalf("retry changed from %#v to %#v", tt.current.Retry, got.Retry)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplySessionCommandFailurePreservesState(t *testing.T) {
|
||||
base := validCommandSession()
|
||||
tests := []struct {
|
||||
name string
|
||||
current SessionConfig
|
||||
command SessionCommand
|
||||
wantErr error
|
||||
}{
|
||||
{"set empty video", base, SessionCommand{Kind: CommandSetVideo}, ErrVideoNotConfigured},
|
||||
{"set empty audio", base, SessionCommand{Kind: CommandSetAudio}, ErrAudioNotConfigured},
|
||||
{"invalid video", base, SessionCommand{Kind: CommandSetVideo, Config: FeedConfig{UUID: "video"}}, ErrFeedDomainRequired},
|
||||
{"invalid audio", base, SessionCommand{Kind: CommandSetAudio, Config: FeedConfig{UUID: "audio"}}, ErrFeedDomainRequired},
|
||||
{
|
||||
"resume missing video",
|
||||
func() SessionConfig { c := base; c.Video = FeedConfig{}; return c }(),
|
||||
SessionCommand{Kind: CommandResumeVideo},
|
||||
ErrVideoNotConfigured,
|
||||
},
|
||||
{
|
||||
"resume missing audio",
|
||||
func() SessionConfig { c := base; c.Audio = FeedConfig{}; return c }(),
|
||||
SessionCommand{Kind: CommandResumeAudio},
|
||||
ErrAudioNotConfigured,
|
||||
},
|
||||
{"unknown command", base, SessionCommand{Kind: 255}, ErrUnknownSessionCommand},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got, err := ApplySessionCommand(tt.current, tt.command)
|
||||
if !errors.Is(err, tt.wantErr) {
|
||||
t.Fatalf("ApplySessionCommand() error = %v, want %v", err, tt.wantErr)
|
||||
}
|
||||
if got != tt.current {
|
||||
t.Fatalf("failed command changed state from %#v to %#v", tt.current, got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplySessionCommandRejectsInvalidResult(t *testing.T) {
|
||||
current := validCommandSession()
|
||||
current.Retry = RetryPolicy{}
|
||||
got, err := ApplySessionCommand(current, SessionCommand{Kind: CommandDisableSync})
|
||||
if !errors.Is(err, ErrInvalidRetryDelay) {
|
||||
t.Fatalf("ApplySessionCommand() error = %v, want %v", err, ErrInvalidRetryDelay)
|
||||
}
|
||||
if got != current {
|
||||
t.Fatalf("failed command changed state from %#v to %#v", current, got)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
package playback
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
type SyncPairConfig struct {
|
||||
Video FeedConfig
|
||||
Audio FeedConfig
|
||||
}
|
||||
|
||||
var (
|
||||
ErrSyncWorkerRequired = errors.New("sync worker is required")
|
||||
ErrSyncActivityMismatch = errors.New(
|
||||
"synchronized video and audio must have matching active states",
|
||||
)
|
||||
)
|
||||
|
||||
func (c SyncPairConfig) Validate() error {
|
||||
if err := c.Video.Validate(); err != nil {
|
||||
return fmt.Errorf("video: %w", err)
|
||||
}
|
||||
if err := c.Audio.Validate(); err != nil {
|
||||
return fmt.Errorf("audio: %w", err)
|
||||
}
|
||||
if c.Video.Active != c.Audio.Active {
|
||||
return ErrSyncActivityMismatch
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c SyncPairConfig) Active() bool {
|
||||
return c.Video.Active && c.Audio.Active
|
||||
}
|
||||
|
||||
type SyncSlot struct {
|
||||
worker *SyncWorker
|
||||
}
|
||||
|
||||
func NewSyncSlot(worker *SyncWorker) (*SyncSlot, error) {
|
||||
if worker == nil {
|
||||
return nil, ErrSyncWorkerRequired
|
||||
}
|
||||
return &SyncSlot{
|
||||
worker: worker,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *SyncSlot) Run(
|
||||
ctx context.Context,
|
||||
initial SyncPairConfig,
|
||||
commands <-chan SyncPairConfig,
|
||||
) error {
|
||||
if err := initial.Validate(); err != nil {
|
||||
return fmt.Errorf("validate initial sync config: %w", err)
|
||||
}
|
||||
|
||||
var (
|
||||
workerCancel context.CancelFunc
|
||||
workerDone chan error
|
||||
)
|
||||
|
||||
start := func(config SyncPairConfig) {
|
||||
workerCtx, cancel := context.WithCancel(ctx)
|
||||
done := make(chan error, 1)
|
||||
|
||||
workerCancel = cancel
|
||||
workerDone = done
|
||||
|
||||
go func() {
|
||||
done <- s.worker.Run(workerCtx, config.Video, config.Audio)
|
||||
}()
|
||||
}
|
||||
|
||||
stop := func() {
|
||||
if workerCancel == nil {
|
||||
return
|
||||
}
|
||||
|
||||
workerCancel()
|
||||
<-workerDone
|
||||
|
||||
workerCancel = nil
|
||||
workerDone = nil
|
||||
}
|
||||
|
||||
if initial.Active() {
|
||||
start(initial)
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
stop()
|
||||
return ctx.Err()
|
||||
|
||||
case config, ok := <-commands:
|
||||
if !ok {
|
||||
stop()
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := config.Validate(); err != nil {
|
||||
// Ignore invalid commands without disturbing the current worker.
|
||||
continue
|
||||
}
|
||||
|
||||
stop()
|
||||
if config.Active() {
|
||||
start(config)
|
||||
}
|
||||
|
||||
case <-workerDone:
|
||||
// The worker stopped naturally or exhausted its retries.
|
||||
workerCancel()
|
||||
workerCancel = nil
|
||||
workerDone = nil
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,231 @@
|
||||
package playback
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
type slotSyncFactory struct {
|
||||
opened chan SyncPairConfig
|
||||
|
||||
mu sync.Mutex
|
||||
active int
|
||||
maxActive int
|
||||
closeCount int
|
||||
}
|
||||
|
||||
func newSlotSyncFactory() *slotSyncFactory {
|
||||
return &slotSyncFactory{opened: make(chan SyncPairConfig, 8)}
|
||||
}
|
||||
|
||||
func (f *slotSyncFactory) OpenSync(
|
||||
_ context.Context,
|
||||
video FeedConfig,
|
||||
audio FeedConfig,
|
||||
) (SyncReader, error) {
|
||||
f.mu.Lock()
|
||||
f.active++
|
||||
if f.active > f.maxActive {
|
||||
f.maxActive = f.active
|
||||
}
|
||||
f.mu.Unlock()
|
||||
f.opened <- SyncPairConfig{Video: video, Audio: audio}
|
||||
return &slotSyncReader{factory: f}, nil
|
||||
}
|
||||
|
||||
func (f *slotSyncFactory) counts() (active, maxActive, closeCount int) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.active, f.maxActive, f.closeCount
|
||||
}
|
||||
|
||||
type slotSyncReader struct {
|
||||
factory *slotSyncFactory
|
||||
}
|
||||
|
||||
func (r *slotSyncReader) ReadSync(ctx context.Context) (SyncFrame, error) {
|
||||
<-ctx.Done()
|
||||
return SyncFrame{}, ctx.Err()
|
||||
}
|
||||
|
||||
func (r *slotSyncReader) Close() error {
|
||||
r.factory.mu.Lock()
|
||||
defer r.factory.mu.Unlock()
|
||||
r.factory.active--
|
||||
r.factory.closeCount++
|
||||
return nil
|
||||
}
|
||||
|
||||
func newSlotTestSyncWorker(t *testing.T, factory SyncReaderFactory) *SyncWorker {
|
||||
t.Helper()
|
||||
worker, err := NewSyncWorker(
|
||||
factory,
|
||||
&fakeVideoSink{},
|
||||
&fakeAudioSink{},
|
||||
testRetryPolicy(1),
|
||||
func(error) bool { return false },
|
||||
nil,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("NewSyncWorker() error = %v", err)
|
||||
}
|
||||
return worker
|
||||
}
|
||||
|
||||
func receiveSyncSlotOpen(t *testing.T, opened <-chan SyncPairConfig) SyncPairConfig {
|
||||
t.Helper()
|
||||
select {
|
||||
case config := <-opened:
|
||||
return config
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("sync worker did not open")
|
||||
return SyncPairConfig{}
|
||||
}
|
||||
}
|
||||
|
||||
func testSyncPair(name string, active bool) SyncPairConfig {
|
||||
return SyncPairConfig{
|
||||
Video: FeedConfig{Domain: "/video", UUID: name + "-video", Active: active},
|
||||
Audio: FeedConfig{Domain: "/audio", UUID: name + "-audio", Active: active},
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncPairConfigRejectsActivityMismatch(t *testing.T) {
|
||||
config := testSyncPair("pair", true)
|
||||
config.Audio.Active = false
|
||||
if err := config.Validate(); !errors.Is(err, ErrSyncActivityMismatch) {
|
||||
t.Fatalf("Validate() error = %v, want %v", err, ErrSyncActivityMismatch)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewSyncSlotRequiresWorker(t *testing.T) {
|
||||
slot, err := NewSyncSlot(nil)
|
||||
if slot != nil {
|
||||
t.Fatalf("NewSyncSlot(nil) slot = %#v, want nil", slot)
|
||||
}
|
||||
if !errors.Is(err, ErrSyncWorkerRequired) {
|
||||
t.Fatalf("NewSyncSlot(nil) error = %v, want %v", err, ErrSyncWorkerRequired)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncSlotStartsInitialActivePair(t *testing.T) {
|
||||
factory := newSlotSyncFactory()
|
||||
slot, err := NewSyncSlot(newSlotTestSyncWorker(t, factory))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan error, 1)
|
||||
want := testSyncPair("first", true)
|
||||
go func() { done <- slot.Run(ctx, want, make(chan SyncPairConfig)) }()
|
||||
|
||||
if got := receiveSyncSlotOpen(t, factory.opened); got != want {
|
||||
t.Fatalf("opened config = %#v, want %#v", got, want)
|
||||
}
|
||||
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")
|
||||
}
|
||||
active, _, closed := factory.counts()
|
||||
if active != 0 || closed != 1 {
|
||||
t.Fatalf("reader counts = active %d, closed %d; want 0, 1", active, closed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncSlotReplacesWithoutOverlappingWorkers(t *testing.T) {
|
||||
factory := newSlotSyncFactory()
|
||||
slot, err := NewSyncSlot(newSlotTestSyncWorker(t, factory))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
commands := make(chan SyncPairConfig)
|
||||
done := make(chan error, 1)
|
||||
first := testSyncPair("first", true)
|
||||
second := testSyncPair("second", true)
|
||||
go func() { done <- slot.Run(ctx, first, commands) }()
|
||||
receiveSyncSlotOpen(t, factory.opened)
|
||||
commands <- second
|
||||
if got := receiveSyncSlotOpen(t, factory.opened); got != second {
|
||||
t.Fatalf("replacement = %#v, want %#v", got, second)
|
||||
}
|
||||
cancel()
|
||||
<-done
|
||||
active, maxActive, closed := factory.counts()
|
||||
if active != 0 || maxActive != 1 || closed != 2 {
|
||||
t.Fatalf("counts = active %d, maximum %d, closed %d; want 0, 1, 2", active, maxActive, closed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncSlotIgnoresInvalidCommand(t *testing.T) {
|
||||
factory := newSlotSyncFactory()
|
||||
slot, err := NewSyncSlot(newSlotTestSyncWorker(t, factory))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
commands := make(chan SyncPairConfig)
|
||||
done := make(chan error, 1)
|
||||
initial := testSyncPair("first", true)
|
||||
go func() { done <- slot.Run(ctx, initial, commands) }()
|
||||
receiveSyncSlotOpen(t, factory.opened)
|
||||
|
||||
invalid := testSyncPair("invalid", true)
|
||||
invalid.Audio.Active = false
|
||||
commands <- invalid
|
||||
select {
|
||||
case config := <-factory.opened:
|
||||
t.Fatalf("invalid command opened config %#v", config)
|
||||
case <-time.After(20 * time.Millisecond):
|
||||
}
|
||||
active, _, closed := factory.counts()
|
||||
if active != 1 || closed != 0 {
|
||||
t.Fatalf("invalid command disturbed reader: active %d, closed %d", active, closed)
|
||||
}
|
||||
cancel()
|
||||
<-done
|
||||
}
|
||||
|
||||
func TestSyncSlotInactivePairStopsWithoutRestart(t *testing.T) {
|
||||
factory := newSlotSyncFactory()
|
||||
slot, err := NewSyncSlot(newSlotTestSyncWorker(t, factory))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
commands := make(chan SyncPairConfig)
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- slot.Run(ctx, testSyncPair("first", true), commands) }()
|
||||
receiveSyncSlotOpen(t, factory.opened)
|
||||
commands <- testSyncPair("first", false)
|
||||
|
||||
deadline := time.Now().Add(time.Second)
|
||||
for {
|
||||
active, _, closed := factory.counts()
|
||||
if active == 0 && closed == 1 {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("inactive pair did not stop reader")
|
||||
}
|
||||
time.Sleep(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 after commands closed")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user