go-mxl-player/internal/adapter/mxl/video.go

package mxladapter

import (
	"context"
	"errors"
	"fmt"
	"time"

	"mxl-player/internal/playback"
	"mxl-player/internal/source"
)

const (
	DefaultVideoReadTimeout      = 200 * time.Millisecond
	DefaultVideoUnavailableAfter = 2 * time.Second
	DefaultTemporaryRetryDelay   = 10 * time.Millisecond
)

type VideoFactory struct {
	ReadTimeout      time.Duration
	UnavailableAfter time.Duration
}

type videoReader struct {
	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{}
var _ playback.VideoReader = (*videoReader)(nil)

func (f VideoFactory) OpenVideo(
	ctx context.Context,
	config playback.FeedConfig,
) (playback.VideoReader, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if err := config.Validate(); err != nil {
		return nil, &source.SourceError{
			Op:   "validate video feed",
			Kind: source.ErrorKindInvalidConfig,
			Err:  err,
		}
	}
	if !config.IsConfigured() {
		return nil, &source.SourceError{
			Op:   "validate video feed",
			Kind: source.ErrorKindInvalidConfig,
			Err:  errors.New("video feed is not configured"),
		}
	}

	src, err := source.Open(config.Domain, config.UUID)
	if err != nil {
		return nil, fmt.Errorf("open local MXL video: %w", err)
	}

	if err := ctx.Err(); err != nil {
		_ = src.Close()
		return nil, err
	}

	readTimeout := f.ReadTimeout
	if readTimeout <= 0 {
		readTimeout = DefaultVideoReadTimeout
	}

	unavailableAfter := f.UnavailableAfter
	if unavailableAfter <= 0 {
		unavailableAfter = DefaultVideoUnavailableAfter
	}

	return &videoReader{
		source:           src,
		readTimeout:      readTimeout,
		unavailableAfter: unavailableAfter,
		retryDelay:       DefaultTemporaryRetryDelay,
		now:              time.Now,
		wait:             waitForTemporaryRetry,
	}, nil
}

func (r *videoReader) ReadVideo(
	ctx context.Context,
) (playback.VideoFrame, error) {
	var unavailableSince time.Time

	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,
				Label:                frame.Label,
				FrameRateNumerator:   frame.FrameRateNumerator,
				FrameRateDenominator: frame.FrameRateDenominator,
				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 {
	return r.source.Close()
}