cmd/mxl-pattern/main.go

@@ -61,6 +61,16 @@ type parseResult struct {
 	shouldRun bool
 }
 
+type namedRunner struct {
+	name string
+	run  func(context.Context) error
+}
+
+type runnerResult struct {
+	name string
+	err  error
+}
+
 var frameRates = map[string]mxl.Rational{
 	"23.97": {Num: 24000, Den: 1001},
 	"24":    {Num: 24, Den: 1},
@@ -503,12 +513,21 @@ func run(ctx context.Context, args appArgs) (runErr error) {
 	if err != nil {
 		return fmt.Errorf("video configuration: %w", err)
 	}
+	audioCfg, err := buildAudioConfig(args)
+	if err != nil {
+		return fmt.Errorf("audio configuration: %w", err)
+	}
 
 	log.Printf("%s %s", APP_NAME, APP_VER)
 	log.Printf("Domain: %s", args.domain)
 	log.Printf("Video: %dx%d %d/%d",
 		videoCfg.Width(), videoCfg.Height(), videoCfg.Rate().Num, videoCfg.Rate().Den)
 	log.Printf("Video ID: %s", videoCfg.ID())
+	if audioCfg != nil {
+		log.Printf("Audio: %d channels %d/%d Hz %.0f dBFS",
+			audioCfg.Channels(), audioCfg.Rate().Num, audioCfg.Rate().Den, audioCfg.LevelDBFS)
+		log.Printf("Audio ID: %s", audioCfg.ID())
+	}
 
 	inst, err := mxl.NewInstance(args.domain, "")
 	if err != nil {
@@ -520,7 +539,50 @@ func run(ctx context.Context, args appArgs) (runErr error) {
 		}
 	}()
 
+	runners := []namedRunner{
+		{
+			name: "video",
+			run: func(ctx context.Context) error {
 				return runVideo(ctx, inst, videoCfg)
+			},
+		},
+	}
+	if audioCfg != nil {
+		runners = append(runners, namedRunner{
+			name: "audio",
+			run: func(ctx context.Context) error {
+				return runAudio(ctx, inst, *audioCfg)
+			},
+		})
+	}
+
+	return runConcurrent(ctx, runners...)
+}
+
+func runConcurrent(ctx context.Context, runners ...namedRunner) error {
+	if len(runners) == 0 {
+		return nil
+	}
+
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	results := make(chan runnerResult, len(runners))
+	for _, runner := range runners {
+		runner := runner
+		go func() {
+			results <- runnerResult{name: runner.name, err: runner.run(ctx)}
+		}()
+	}
+
+	var resultErr error
+	for range runners {
+		result := <-results
+		if result.err != nil {
+			resultErr = errors.Join(resultErr, fmt.Errorf("%s flow: %w", result.name, result.err))
+			cancel()
+		}
+	}
+	return resultErr
 }
 
 func runVideo(ctx context.Context, inst *mxl.Instance, cfg video.Config) (runErr error) {
@@ -602,6 +664,157 @@ func runVideo(ctx context.Context, inst *mxl.Instance, cfg video.Config) (runErr
 	}
 }
 
+func runAudio(
+	ctx context.Context,
+	inst *mxl.Instance,
+	cfg audio.Config,
+) (runErr error) {
+	flowJSON, err := json.Marshal(cfg.Definition)
+	if err != nil {
+		return fmt.Errorf("marshal audio flow definition: %w", err)
+	}
+
+	writer, isCreated, err := inst.NewWriter(string(flowJSON))
+	if err != nil {
+		return fmt.Errorf("create audio writer: %w", err)
+	}
+	defer func() {
+		if err := writer.Close(); err != nil {
+			runErr = errors.Join(
+				runErr,
+				fmt.Errorf("close audio writer: %w", err),
+			)
+		}
+	}()
+	if !isCreated {
+		log.Printf("reusing existing audio flow: %s", cfg.ID())
+	}
+
+	writerCfg := writer.Config()
+	if writerCfg.Common.Format != mxl.FormatAudio {
+		return fmt.Errorf(
+			"audio writer has format %s, want audio",
+			writerCfg.Common.Format,
+		)
+	}
+	if writerCfg.Continuous.ChannelCount != uint32(cfg.Channels()) {
+		return fmt.Errorf(
+			"audio writer has %d channels, configured generator expects %d",
+			writerCfg.Continuous.ChannelCount,
+			cfg.Channels(),
+		)
+	}
+
+	rate := writerCfg.Common.GrainRate
+	if rate != cfg.Rate() {
+		return fmt.Errorf(
+			"audio writer has sample rate %d/%d, configured generator expects %d/%d",
+			rate.Num,
+			rate.Den,
+			cfg.Rate().Num,
+			cfg.Rate().Den,
+		)
+	}
+
+	const baseFrequency = 1000.0
+
+	gen, err := audio.NewSineGenerator(cfg, baseFrequency)
+	if err != nil {
+		return fmt.Errorf("initialize audio generator: %w", err)
+	}
+
+	batch := audioBatchSize(rate)
+	maxBatch, err := writer.GetMaxWriteLengthSamples()
+	if err != nil {
+		return fmt.Errorf("get maximum audio write length: %w", err)
+	}
+	if maxBatch == 0 {
+		return fmt.Errorf("audio writer reported a maximum write length of zero samples")
+	}
+	if batch > maxBatch {
+		batch = maxBatch
+	}
+	index := mxl.CurrentIndex(rate)
+	if index < batch-1 {
+		return fmt.Errorf("current audio index %d is too small for batch size %d", index, batch)
+	}
+
+	log.Printf(
+		"writing audio flow sampleRate=%d/%d channels=%d batch=%d starting at idx=%d",
+		rate.Num,
+		rate.Den,
+		cfg.Channels(),
+		batch,
+		index,
+	)
+
+	var samplesWritten uint64
+
+	for {
+		select {
+		case <-ctx.Done():
+			log.Printf("stopping audio after %d samples", samplesWritten)
+			return nil
+		default:
+		}
+
+		access, err := writer.OpenSamples(index, int(batch))
+		if err != nil {
+			return fmt.Errorf(
+				"open %d audio samples at index %d: %w",
+				batch,
+				index,
+				err,
+			)
+		}
+
+		firstSample := index - batch + 1
+		for channel := uint64(0); channel < access.ChannelCount; channel++ {
+			first, second, err := access.ChannelFragments(channel)
+			if err != nil {
+				return cancelAudioSamples(
+					access,
+					fmt.Errorf(
+						"get fragments for audio channel %d at index %d: %w",
+						channel,
+						index,
+						err,
+					),
+				)
+			}
+
+			if err := gen.Generate(
+				uint(channel),
+				firstSample,
+				first,
+				second,
+			); err != nil {
+				return cancelAudioSamples(
+					access, fmt.Errorf(
+						"generate audio channel %d at index %d: %w",
+						channel,
+						index,
+						err,
+					),
+				)
+			}
+		}
+
+		if err := access.Commit(); err != nil {
+			return fmt.Errorf(
+				"commit %d audio samples at index %d: %w",
+				batch,
+				index,
+				err,
+			)
+		}
+
+		samplesWritten += batch
+		index += batch
+		mxl.SleepNs(mxl.NsUntilIndex(index, rate))
+	}
+}
+
 func buildTextOverlay(cfg video.Config) (overlay *generator.TextOverlay, resultErr error) {
 	if cfg.Overlay.Text == "" {
 		return nil, nil
@@ -643,3 +856,28 @@ func cancelVideoGrain(grain *mxl.GrainWriteAccess, cause error) error {
 	}
 	return cause
 }
+
+func audioBatchSize(rate mxl.Rational) uint64 {
+	if rate.Num <= 0 || rate.Den <= 0 {
+		return 1
+	}
+
+	samples := rate.Num / (100 * rate.Den)
+	if samples < 1 {
+		return 1
+	}
+	return uint64(samples)
+}
+
+func cancelAudioSamples(
+	access *mxl.SamplesWriteAccess,
+	cause error,
+) error {
+	if err := access.Cancel(); err != nil {
+		return errors.Join(
+			cause,
+			fmt.Errorf("cancel audio samples: %w", err),
+		)
+	}
+	return cause
+}

cmd/mxl-pattern/main_test.go

@@ -2,10 +2,15 @@ package main
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
+	"errors"
 	"os"
 	"strings"
 	"testing"
+	"time"
+
+	"github.com/qvest-digital/go-mxl/mxl"
 
 	"mxl-pattern-generator/internal/audio"
 	"mxl-pattern-generator/internal/flowdef"
@@ -250,3 +255,91 @@ func TestValidateAudioArgsRejectsUnknownLevelForFlowDefinition(t *testing.T) {
 		t.Fatalf("error = %v, want unsupported audio level error", err)
 	}
 }
+
+func TestAudioBatchSize(t *testing.T) {
+	tests := []struct {
+		name string
+		rate mxl.Rational
+		want uint64
+	}{
+		{name: "44.1 kHz", rate: mxl.Rational{Num: 44100, Den: 1}, want: 441},
+		{name: "48 kHz", rate: mxl.Rational{Num: 48000, Den: 1}, want: 480},
+		{name: "96 kHz", rate: mxl.Rational{Num: 96000, Den: 1}, want: 960},
+		{name: "192 kHz", rate: mxl.Rational{Num: 192000, Den: 1}, want: 1920},
+		{name: "minimum", rate: mxl.Rational{Num: 1, Den: 1}, want: 1},
+		{name: "zero numerator", rate: mxl.Rational{Num: 0, Den: 1}, want: 1},
+		{name: "zero denominator", rate: mxl.Rational{Num: 48000, Den: 0}, want: 1},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := audioBatchSize(tc.rate); got != tc.want {
+				t.Fatalf("audioBatchSize(%d/%d) = %d, want %d", tc.rate.Num, tc.rate.Den, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestRunConcurrentCancelsSiblingAndWaitsForCleanup(t *testing.T) {
+	wantErr := errors.New("writer failed")
+	peerStarted := make(chan struct{})
+	peerStopped := make(chan struct{})
+
+	err := runConcurrent(context.Background(),
+		namedRunner{
+			name: "video",
+			run: func(ctx context.Context) error {
+				<-peerStarted
+				return wantErr
+			},
+		},
+		namedRunner{
+			name: "audio",
+			run: func(ctx context.Context) error {
+				close(peerStarted)
+				<-ctx.Done()
+				close(peerStopped)
+				return nil
+			},
+		},
+	)
+
+	if !errors.Is(err, wantErr) {
+		t.Fatalf("error = %v, want wrapped %v", err, wantErr)
+	}
+	if !strings.Contains(err.Error(), "video flow") {
+		t.Fatalf("error = %q, want runner name", err)
+	}
+	select {
+	case <-peerStopped:
+	default:
+		t.Fatal("runConcurrent returned before the sibling completed cleanup")
+	}
+}
+
+func TestRunConcurrentParentCancellationIsGraceful(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	started := make(chan struct{})
+	done := make(chan error, 1)
+	go func() {
+		done <- runConcurrent(ctx, namedRunner{
+			name: "video",
+			run: func(ctx context.Context) error {
+				close(started)
+				<-ctx.Done()
+				return nil
+			},
+		})
+	}()
+
+	<-started
+	cancel()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("runConcurrent: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("runConcurrent did not stop after parent cancellation")
+	}
+}