diff --git a/internal/adapter/mxl/video.go b/internal/adapter/mxl/video.go index 45f69e5..26e662d 100644 --- a/internal/adapter/mxl/video.go +++ b/internal/adapter/mxl/video.go @@ -10,15 +10,43 @@ import ( "mxl-player/internal/source" ) -const DefaultVideoReadTimeout = 200 * time.Millisecond +const ( + DefaultVideoReadTimeout = 200 * time.Millisecond + DefaultVideoUnavailableAfter = 2 * time.Second + DefaultTemporaryRetryDelay = 10 * time.Millisecond +) type VideoFactory struct { - ReadTimeout time.Duration + ReadTimeout time.Duration + UnavailableAfter time.Duration } type videoReader struct { - source *source.Source - timeout time.Duration + source localVideoSource + readTimeout time.Duration + unavailableAfter time.Duration + retryDelay time.Duration + now func() time.Time + wait temporaryWaitFunc +} + +type localVideoSource interface { + ReadOnceCtx(context.Context, time.Duration) (source.Frame, error) + Close() error +} + +type temporaryWaitFunc func(context.Context, time.Duration) error + +func waitForTemporaryRetry(ctx context.Context, delay time.Duration) error { + timer := time.NewTimer(delay) + defer timer.Stop() + + select { + case <-timer.C: + return nil + case <-ctx.Done(): + return ctx.Err() + } } var _ playback.VideoReaderFactory = VideoFactory{} @@ -56,34 +84,73 @@ func (f VideoFactory) OpenVideo( return nil, err } - timeout := f.ReadTimeout - if timeout <= 0 { - timeout = DefaultVideoReadTimeout + readTimeout := f.ReadTimeout + if readTimeout <= 0 { + readTimeout = DefaultVideoReadTimeout + } + + unavailableAfter := f.UnavailableAfter + if unavailableAfter <= 0 { + unavailableAfter = DefaultVideoUnavailableAfter } return &videoReader{ - source: src, - timeout: timeout, + source: src, + readTimeout: readTimeout, + unavailableAfter: unavailableAfter, + retryDelay: DefaultTemporaryRetryDelay, + now: time.Now, + wait: waitForTemporaryRetry, }, nil } func (r *videoReader) ReadVideo( ctx context.Context, ) (playback.VideoFrame, error) { - frame, err := r.source.NextCtx(ctx, r.timeout) - if err != nil { - return playback.VideoFrame{}, err - } + var unavailableSince time.Time - return playback.VideoFrame{ - Index: frame.Index, - Width: frame.Width, - Height: frame.Height, - Stride: frame.Stride, - Size: frame.Size, - Invalid: frame.Invalid, - Payload: frame.Payload, - }, nil + for { + frame, err := r.source.ReadOnceCtx(ctx, r.readTimeout) + if err == nil { + return playback.VideoFrame{ + Index: frame.Index, + Width: frame.Width, + Height: frame.Height, + Stride: frame.Stride, + Size: frame.Size, + Invalid: frame.Invalid, + Payload: frame.Payload, + }, nil + } + if ctx.Err() != nil { + return playback.VideoFrame{}, ctx.Err() + } + if source.KindOf(err) != source.ErrorKindTemporary { + return playback.VideoFrame{}, err + } + + now := r.now() + if unavailableSince.IsZero() { + unavailableSince = now + } else if now.Sub(unavailableSince) >= r.unavailableAfter { + return playback.VideoFrame{}, &source.SourceError{ + Op: "read local MXL video", + Kind: source.ErrorKindUnavailable, + Err: fmt.Errorf( + "no video data for %s: %w", + r.unavailableAfter, + err, + ), + } + } + + if err := r.wait(ctx, r.retryDelay); err != nil { + if ctx.Err() != nil { + return playback.VideoFrame{}, ctx.Err() + } + return playback.VideoFrame{}, err + } + } } func (r *videoReader) Close() error { diff --git a/internal/adapter/mxl/video_test.go b/internal/adapter/mxl/video_test.go index 94467a8..156e7c7 100644 --- a/internal/adapter/mxl/video_test.go +++ b/internal/adapter/mxl/video_test.go @@ -4,11 +4,51 @@ import ( "context" "errors" "testing" + "time" "mxl-player/internal/playback" "mxl-player/internal/source" ) +type localVideoReadResult struct { + frame source.Frame + err error +} + +type fakeLocalVideoSource struct { + results []localVideoReadResult + calls int + timeouts []time.Duration + closeErr error + closed bool +} + +func (s *fakeLocalVideoSource) ReadOnceCtx( + _ context.Context, + timeout time.Duration, +) (source.Frame, error) { + s.timeouts = append(s.timeouts, timeout) + if s.calls >= len(s.results) { + return source.Frame{}, errors.New("unexpected local video read") + } + result := s.results[s.calls] + s.calls++ + return result.frame, result.err +} + +func (s *fakeLocalVideoSource) Close() error { + s.closed = true + return s.closeErr +} + +func temporaryVideoError(cause error) error { + return &source.SourceError{ + Op: "read video", + Kind: source.ErrorKindTemporary, + Err: cause, + } +} + func TestVideoFactoryOpenVideoCanceled(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() @@ -73,3 +113,188 @@ func TestVideoFactoryOpenVideoRejectsInvalidConfig(t *testing.T) { }) } } + +func TestVideoReaderTemporaryFailureThenFrame(t *testing.T) { + temporaryErr := errors.New("video is early") + payload := []byte{1, 2, 3, 4} + wantFrame := source.Frame{ + Index: 42, + Width: 1920, + Height: 1080, + Stride: 5120, + Size: 5120 * 1080, + Invalid: false, + Payload: payload, + } + localSource := &fakeLocalVideoSource{ + results: []localVideoReadResult{ + {err: temporaryVideoError(temporaryErr)}, + {frame: wantFrame}, + }, + } + waits := 0 + reader := &videoReader{ + source: localSource, + readTimeout: 250 * time.Millisecond, + unavailableAfter: 2 * time.Second, + retryDelay: 10 * time.Millisecond, + now: func() time.Time { return time.Unix(100, 0) }, + wait: func(context.Context, time.Duration) error { + waits++ + return nil + }, + } + + got, err := reader.ReadVideo(context.Background()) + if err != nil { + t.Fatalf("ReadVideo() error = %v, want nil", err) + } + if localSource.calls != 2 { + t.Errorf("source read calls = %d, want 2", localSource.calls) + } + if waits != 1 { + t.Errorf("temporary wait calls = %d, want 1", waits) + } + if len(localSource.timeouts) != 2 || + localSource.timeouts[0] != 250*time.Millisecond || + localSource.timeouts[1] != 250*time.Millisecond { + t.Errorf("source timeouts = %v, want [250ms 250ms]", localSource.timeouts) + } + if got.Index != wantFrame.Index || + got.Width != wantFrame.Width || + got.Height != wantFrame.Height || + got.Stride != wantFrame.Stride || + got.Size != wantFrame.Size || + got.Invalid != wantFrame.Invalid { + t.Errorf("video frame = %+v, want metadata from %+v", got, wantFrame) + } + if len(got.Payload) == 0 || &got.Payload[0] != &payload[0] { + t.Fatal("video payload was copied") + } +} + +func TestVideoReaderProlongedTemporaryFailureBecomesUnavailable(t *testing.T) { + lastCause := errors.New("video timeout") + localSource := &fakeLocalVideoSource{ + results: []localVideoReadResult{ + {err: temporaryVideoError(errors.New("first timeout"))}, + {err: temporaryVideoError(errors.New("second timeout"))}, + {err: temporaryVideoError(lastCause)}, + }, + } + times := []time.Time{ + time.Unix(100, 0), + time.Unix(101, 0), + time.Unix(102, 0), + } + nowCall := 0 + waits := 0 + reader := &videoReader{ + source: localSource, + readTimeout: 200 * time.Millisecond, + unavailableAfter: 2 * time.Second, + retryDelay: 10 * time.Millisecond, + now: func() time.Time { + result := times[nowCall] + nowCall++ + return result + }, + wait: func(context.Context, time.Duration) error { + waits++ + return nil + }, + } + + _, err := reader.ReadVideo(context.Background()) + if err == nil { + t.Fatal("ReadVideo() error is nil after prolonged unavailability") + } + if got := source.KindOf(err); got != source.ErrorKindUnavailable { + t.Fatalf("source.KindOf(ReadVideo()) = %v, want %v", got, source.ErrorKindUnavailable) + } + if !errors.Is(err, lastCause) { + t.Errorf("ReadVideo() error = %v, want cause %v", err, lastCause) + } + if !ShouldRetry(err) { + t.Error("ShouldRetry(ReadVideo()) = false, want true") + } + if localSource.calls != 3 { + t.Errorf("source read calls = %d, want 3", localSource.calls) + } + if waits != 2 { + t.Errorf("temporary wait calls = %d, want 2", waits) + } +} + +func TestVideoReaderReturnsNonTemporaryErrorImmediately(t *testing.T) { + unavailableErr := &source.SourceError{ + Op: "read video", + Kind: source.ErrorKindUnavailable, + Err: errors.New("flow invalid"), + } + localSource := &fakeLocalVideoSource{ + results: []localVideoReadResult{{err: unavailableErr}}, + } + reader := &videoReader{ + source: localSource, + readTimeout: 200 * time.Millisecond, + unavailableAfter: 2 * time.Second, + retryDelay: 10 * time.Millisecond, + now: func() time.Time { + t.Fatal("clock called for non-temporary error") + return time.Time{} + }, + wait: func(context.Context, time.Duration) error { + t.Fatal("wait called for non-temporary error") + return nil + }, + } + + _, err := reader.ReadVideo(context.Background()) + if !errors.Is(err, unavailableErr) { + t.Fatalf("ReadVideo() error = %v, want %v", err, unavailableErr) + } + if localSource.calls != 1 { + t.Errorf("source read calls = %d, want 1", localSource.calls) + } +} + +func TestVideoReaderCancellationDuringTemporaryWait(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + localSource := &fakeLocalVideoSource{ + results: []localVideoReadResult{{err: temporaryVideoError(errors.New("early"))}}, + } + reader := &videoReader{ + source: localSource, + readTimeout: 200 * time.Millisecond, + unavailableAfter: 2 * time.Second, + retryDelay: 10 * time.Millisecond, + now: func() time.Time { return time.Unix(100, 0) }, + wait: func(ctx context.Context, _ time.Duration) error { + cancel() + return ctx.Err() + }, + } + + _, err := reader.ReadVideo(ctx) + if !errors.Is(err, context.Canceled) { + t.Fatalf("ReadVideo() error = %v, want context.Canceled", err) + } + if localSource.calls != 1 { + t.Errorf("source read calls = %d, want 1", localSource.calls) + } +} + +func TestVideoReaderCloseDelegatesToSource(t *testing.T) { + closeErr := errors.New("close failed") + localSource := &fakeLocalVideoSource{closeErr: closeErr} + reader := &videoReader{source: localSource} + + err := reader.Close() + if !errors.Is(err, closeErr) { + t.Fatalf("Close() error = %v, want %v", err, closeErr) + } + if !localSource.closed { + t.Fatal("local source was not closed") + } +}