package playback import ( "context" "errors" "fmt" ) var ( ErrVideoFactoryRequired = errors.New("video reader factory is required") ErrVideoSinkRequired = errors.New("video sink is required") ErrVideoRetryDeciderRequired = errors.New("video decider is required") ErrVideoFeedInactive = errors.New("video feed is not active") ) type VideoWorker struct { factory VideoReaderFactory sink VideoSink retry RetryPolicy shouldRetry retryDecider observer StatusObserver wait waitFunc } func NewVideoWorker( factory VideoReaderFactory, sink VideoSink, retry RetryPolicy, shouldRetry func(error) bool, observer StatusObserver, ) (*VideoWorker, error) { if factory == nil { return nil, ErrVideoFactoryRequired } if sink == nil { return nil, ErrVideoSinkRequired } if shouldRetry == nil { return nil, ErrVideoRetryDeciderRequired } if err := retry.Validate(); err != nil { return nil, fmt.Errorf("validate video retry policy: %w", err) } return &VideoWorker{ factory: factory, sink: sink, retry: retry, shouldRetry: shouldRetry, observer: observer, wait: waitForRetry, }, nil } type stabilityVideoSink struct { sink VideoSink onStable func() stable bool } func (s *stabilityVideoSink) ConsumeVideo( ctx context.Context, frame VideoFrame, ) error { err := s.sink.ConsumeVideo(ctx, frame) if err == nil && !s.stable { s.stable = true if s.onStable != nil { s.onStable() } } return err } func (w *VideoWorker) emit(ctx context.Context, config FeedConfig, status Status) { status.Generation = generationFromContext(ctx) status.Feed = config if w.observer != nil { w.observer(status) } } func (w *VideoWorker) Run( ctx context.Context, config FeedConfig, ) error { if err := config.Validate(); err != nil { return fmt.Errorf("validate video config: %w", err) } if !config.Active { return ErrVideoFeedInactive } attemptNumber := 0 var latestRetry retryEvent attempt := func(ctx context.Context) (bool, error) { attemptNumber++ state := StateConnecting if attemptNumber > 1 { state = StateReconnecting } w.emit(ctx, config, Status{ Unit: UnitVideo, State: state, Attempt: attemptNumber, }) attemptSink := &stabilityVideoSink{ sink: w.sink, onStable: func() { w.emit(ctx, config, Status{ Unit: UnitVideo, State: StatePlaying, Attempt: attemptNumber, }) }, } err := runVideoAttempt(ctx, w.factory, attemptSink, config) return attemptSink.stable, err } decide := func(err error) bool { var sinkErr *videoSinkError if errors.As(err, &sinkErr) { return false } return w.shouldRetry(err) } observeRetry := func(event retryEvent) { latestRetry = event if !event.WillRetry { return } w.emit(ctx, config, Status{ Unit: UnitVideo, State: StateReconnecting, Attempt: attemptNumber + 1, FailedAttempts: event.FailedAttempts, RetryIn: event.RetryIn, Err: event.Err, }) } err := runWithRetry( ctx, w.retry, attempt, decide, w.wait, observeRetry, ) if ctx.Err() != nil { w.emit(ctx, config, Status{ Unit: UnitVideo, State: StateStopping, }) w.emit(ctx, config, Status{ Unit: UnitVideo, State: StateIdle, }) return ctx.Err() } if err != nil { w.emit(ctx, config, Status{ Unit: UnitVideo, State: StateFailed, Attempt: attemptNumber, FailedAttempts: latestRetry.FailedAttempts, Err: err, }) return err } w.emit(ctx, config, Status{ Unit: UnitVideo, State: StateIdle, }) return nil }