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

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
)

var ErrNativeSyncDifferentDomains = errors.New(
	"native MXL synchronization requires matching domains",
)

type SyncFactory struct {
	ReadTimeout time.Duration
	open        func(string, string, string) (localSyncSource, error)
}

type localSyncSource interface {
	NextSync(
		context.Context,
		time.Duration,
	) (source.Frame, source.AudioFrame, error)

	AudioRate() mxl.Rational
	Close() error
}

type syncReader struct {
	source          localSyncSource
	readTimeout     time.Duration
	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
	}
	audioRate := src.AudioRate()

	return &syncReader{
		source:          src,
		readTimeout:     readTimeout,
		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.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,
			Label:                video.Label,
			FrameRateNumerator:   video.FrameRateNumerator,
			FrameRateDenominator: video.FrameRateDenominator,
			Payload:              video.Payload,
		},
		Audio: playback.AudioFrame{
			Index:                 audio.Index,
			SampleCount:           audio.SampleCount,
			Channels:              audio.Channels,
			Label:                 audio.Label,
			SampleRateNumerator:   r.rateNumerator,
			SampleRateDenominator: r.rateDenominator,
			Samples:               audio.Samples,
		},
	}, nil
}

func (r *syncReader) Close() error {
	return r.source.Close()
}