mxl same domain sync adapter

internal/adapter/mxl/sync.go

@@ -0,0 +1,175 @@
+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
+	DefaultSyncBatchDuration = 10 * time.Millisecond
+)
+
+var ErrNativeSyncDifferentDomains = errors.New(
+	"native MXL synchronization requires matching domains",
+)
+
+type SyncFactory struct {
+	ReadTimeout   time.Duration
+	BatchDuration time.Duration
+	open          func(string, string, string) (localSyncSource, error)
+}
+
+type localSyncSource interface {
+	NextSync(
+		context.Context,
+		uint64,
+		time.Duration,
+	) (source.Frame, source.AudioFrame, error)
+
+	AudioRate() mxl.Rational
+	Close() error
+}
+
+type syncReader struct {
+	source          localSyncSource
+	readTimeout     time.Duration
+	audioBatch      uint64
+	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
+	}
+	batchDuration := f.BatchDuration
+	if batchDuration <= 0 {
+		batchDuration = DefaultSyncBatchDuration
+	}
+	audioRate := src.AudioRate()
+	batch, err := audioBatchSize(audioRate.Num, audioRate.Den, batchDuration)
+	if err != nil {
+		_ = src.Close()
+		return nil, &source.SourceError{
+			Op:   "calculate sync audio batch",
+			Kind: source.ErrorKindInvalidConfig,
+			Err:  err,
+		}
+	}
+
+	return &syncReader{
+		source:          src,
+		readTimeout:     readTimeout,
+		audioBatch:      batch,
+		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.audioBatch, 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,
+			Payload: video.Payload,
+		},
+		Audio: playback.AudioFrame{
+			Index:                 audio.Index,
+			SampleCount:           audio.SampleCount,
+			Channels:              audio.Channels,
+			SampleRateNumerator:   r.rateNumerator,
+			SampleRateDenominator: r.rateDenominator,
+			Samples:               audio.Samples,
+		},
+	}, nil
+}
+
+func (r *syncReader) Close() error {
+	return r.source.Close()
+}