go-mxl-player/internal/playback/video_worker.go

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
}