and the video comes into chat

internal/source/source.go

@@ -0,0 +1,166 @@
+package source
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	mxl "github.com/qvest-digital/go-mxl/mxl"
+)
+
+type flowDef struct {
+	FrameWidth  int    `json:"frame_width"`
+	FrameHeight int    `json:"frame_height"`
+	MediaType   string `json:"media_type"`
+	Colorspace  string `json:"colorspace"`
+	GrainRate   struct {
+		Numerator   int64 `json:"numerator"`
+		Denominator int64 `json:"denominator"`
+	} `json:"grain_rate"`
+}
+
+type Frame struct {
+	Index   uint64
+	Width   uint32
+	Height  uint32
+	Stride  uint32
+	Size    uint32
+	Invalid bool
+	Payload []byte
+}
+
+type Source struct {
+	inst   *mxl.Instance
+	reader *mxl.Reader
+	info   mxl.FlowInfo
+	def    string
+	rate   mxl.Rational
+	stride uint32
+	width  uint32
+	height uint32
+	idx    uint64
+}
+
+func Open(domain, flowID string) (*Source, error) {
+	inst, err := mxl.NewInstance(domain, "")
+	if err != nil {
+		return nil, fmt.Errorf("NewInstance: %w", err)
+	}
+	r, err := inst.NewReader(flowID)
+	if err != nil {
+		inst.Close()
+		return nil, fmt.Errorf("NewReader: %w", err)
+	}
+	info, err := r.Info()
+	if err != nil {
+		r.Close()
+		inst.Close()
+		return nil, fmt.Errorf("Info: %w", err)
+	}
+	def, err := inst.FlowDef(flowID)
+	if err != nil {
+		r.Close()
+		inst.Close()
+		return nil, fmt.Errorf("FlowDef: %w", err)
+	}
+	var fd flowDef
+	if err := json.Unmarshal([]byte(def), &fd); err != nil {
+		r.Close()
+		inst.Close()
+		return nil, fmt.Errorf("parse flow def: %w", err)
+	}
+	rate := info.Config.Common.GrainRate
+	idx := mxl.CurrentIndex(rate)
+	if idx == mxl.UndefinedIndex {
+		r.Close()
+		inst.Close()
+		return nil, fmt.Errorf("invalid grain rate: %d/%d", rate.Num, rate.Den)
+	}
+	return &Source{
+		inst:   inst,
+		reader: r,
+		info:   info,
+		def:    def,
+		rate:   info.Config.Common.GrainRate,
+		stride: info.Config.Discrete.SliceSizes[0],
+		width:  uint32(fd.FrameWidth),
+		height: uint32(fd.FrameHeight),
+		idx:    idx,
+	}, nil
+}
+
+func (s *Source) Close() error {
+	_ = s.reader.Close()
+	return s.inst.Close()
+}
+
+func (s *Source) Next(timeout time.Duration) (Frame, error) {
+	for {
+		g, err := s.reader.GetGrain(s.idx, timeout)
+		switch {
+		case err == nil:
+			f := Frame{
+				Index:   g.Index,
+				Width:   s.width,
+				Height:  s.height,
+				Stride:  s.stride,
+				Size:    g.GrainSize,
+				Invalid: g.Invalid(),
+				Payload: g.Payload,
+			}
+			s.idx++
+			return f, nil
+		case errors.Is(err, mxl.ErrTimeout):
+			s.idx = mxl.CurrentIndex(s.rate)
+		case errors.Is(err, mxl.ErrOutOfRangeEarly):
+			time.Sleep(10 * time.Millisecond)
+		case errors.Is(err, mxl.ErrOutOfRangeLate):
+			s.idx = mxl.CurrentIndex(s.rate)
+		default:
+			return Frame{}, fmt.Errorf("GetGrain: %w", err)
+		}
+	}
+}
+
+func (s *Source) NextCtx(ctx context.Context, timeout time.Duration) (Frame, error) {
+	for {
+		select {
+		case <-ctx.Done():
+			return Frame{}, ctx.Err()
+		default:
+		}
+		g, err := s.reader.GetGrain(s.idx, timeout)
+		switch {
+		case err == nil:
+			f := Frame{
+				Index:   g.Index,
+				Width:   s.width,
+				Height:  s.height,
+				Stride:  s.stride,
+				Size:    g.GrainSize,
+				Invalid: g.Invalid(),
+				Payload: g.Payload,
+			}
+			s.idx++
+			return f, nil
+		case errors.Is(err, mxl.ErrTimeout):
+			s.idx = mxl.CurrentIndex(s.rate)
+		case errors.Is(err, mxl.ErrOutOfRangeEarly):
+			time.Sleep(10 * time.Millisecond)
+		case errors.Is(err, mxl.ErrOutOfRangeLate):
+			s.idx = mxl.CurrentIndex(s.rate)
+		default:
+			return Frame{}, fmt.Errorf("GetGrain: %w", err)
+		}
+	}
+}
+
+func (s *Source) FlowDef() string        { return s.def }
+func (s *Source) Rate() mxl.Rational     { return s.rate }
+func (s *Source) Stride() uint32         { return s.stride }
+func (s *Source) Width() uint32          { return s.width }
+func (s *Source) Height() uint32         { return s.height }
+func (s *Source) GrainCount() uint32     { return s.info.Config.Discrete.GrainCount }
+func (s *Source) Format() mxl.DataFormat { return s.info.Config.Common.Format }