app run in app.go

cmd/mxl-pattern/main.go

@@ -5,7 +5,6 @@ package main
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"io"
 	"log"
@@ -19,14 +18,15 @@ import (
 	"github.com/qvest-digital/go-mxl/mxl"
 	"github.com/spf13/pflag"
 
+	"mxl-pattern-generator/internal/app"
 	"mxl-pattern-generator/internal/audio"
 	"mxl-pattern-generator/internal/flowdef"
 	"mxl-pattern-generator/internal/video"
 )
 
 const (
-	APP_NAME = "MXL pattern generator"
+	APP_NAME = app.Name
-	APP_VER  = "0.1.0"
+	APP_VER  = app.Version
 )
 
 type appArgs struct {
@@ -59,16 +59,6 @@ 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},
@@ -433,7 +423,7 @@ func main() {
 	}
 }
 
-func run(ctx context.Context, args appArgs) (runErr error) {
+func run(ctx context.Context, args appArgs) error {
 	videoCfg, err := buildVideoConfig(args)
 	if err != nil {
 		return fmt.Errorf("video configuration: %w", err)
@@ -443,69 +433,9 @@ func run(ctx context.Context, args appArgs) (runErr error) {
 		return fmt.Errorf("audio configuration: %w", err)
 	}
 
-	log.Printf("%s %s", APP_NAME, APP_VER)
+	return app.Run(ctx, app.Config{
-	log.Printf("Domain: %s", args.domain)
+		Domain: args.domain,
-	log.Printf("Video: %dx%d %d/%d",
+		Video:  videoCfg,
-		videoCfg.Width(), videoCfg.Height(), videoCfg.Rate().Num, videoCfg.Rate().Den)
+		Audio:  audioCfg,
-	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 {
-		return fmt.Errorf("initialize MXL domain %q: %w", args.domain, err)
-	}
-	defer func() {
-		if err := inst.Close(); err != nil {
-			runErr = errors.Join(runErr, fmt.Errorf("close MXL instance: %w", err))
-		}
-	}()
-
-	runners := []namedRunner{
-		{
-			name: "video",
-			run: func(ctx context.Context) error {
-				return video.Run(ctx, inst, videoCfg)
-			},
-		},
-	}
-	if audioCfg != nil {
-		runners = append(runners, namedRunner{
-			name: "audio",
-			run: func(ctx context.Context) error {
-				return audio.Run(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
 }

cmd/mxl-pattern/main_test.go

@@ -2,13 +2,10 @@ package main
 
 import (
 	"bytes"
-	"context"
 	"encoding/json"
-	"errors"
 	"os"
 	"strings"
 	"testing"
-	"time"
 
 	"mxl-pattern-generator/internal/audio"
 	"mxl-pattern-generator/internal/flowdef"
@@ -253,67 +250,3 @@ func TestValidateAudioArgsRejectsUnknownLevelForFlowDefinition(t *testing.T) {
 		t.Fatalf("error = %v, want unsupported audio level error", err)
 	}
 }
-
-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")
-	}
-}

internal/app/app.go

@@ -0,0 +1,66 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+
+	"mxl-pattern-generator/internal/audio"
+	"mxl-pattern-generator/internal/video"
+
+	"github.com/qvest-digital/go-mxl/mxl"
+)
+
+const (
+	Name    = "MXL pattern generator"
+	Version = "0.1.0"
+)
+
+type Config struct {
+	Domain string
+	Video  video.Config
+	Audio  *audio.Config
+}
+
+func Run(ctx context.Context, cfg Config) (runErr error) {
+	log.Printf("%s %s", Name, Version)
+	log.Printf("Domain: %s", cfg.Domain)
+	log.Printf("Video: %dx%d %d/%d",
+		cfg.Video.Width(), cfg.Video.Height(), cfg.Video.Rate().Num, cfg.Video.Rate().Den)
+	log.Printf("Video ID: %s", cfg.Video.ID())
+	if cfg.Audio != nil {
+		log.Printf("Audio: %d channels %d/%d Hz %.0f dBFS",
+			cfg.Audio.Channels(), cfg.Audio.Rate().Num, cfg.Audio.Rate().Den, cfg.Audio.LevelDBFS)
+		log.Printf("Audio ID: %s", cfg.Audio.ID())
+	}
+
+	inst, err := mxl.NewInstance(cfg.Domain, "")
+	if err != nil {
+		return fmt.Errorf("initialize MXL domain %q: %w", cfg.Domain, err)
+	}
+	defer func() {
+		if err := inst.Close(); err != nil {
+			runErr = errors.Join(runErr, fmt.Errorf("close MXL instance: %w", err))
+		}
+	}()
+
+	runners := []Runner{
+		{
+			Name: "video",
+			Run: func(ctx context.Context) error {
+				return video.Run(ctx, inst, cfg.Video)
+			},
+		},
+	}
+	if cfg.Audio != nil {
+		runners = append(runners, Runner{
+			Name: "audio",
+			Run: func(ctx context.Context) error {
+				return audio.Run(ctx, inst, *cfg.Audio)
+			},
+		})
+	}
+
+	return RunConcurrent(ctx, runners...)
+}

internal/app/runner.go

@@ -0,0 +1,43 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"fmt"
+)
+
+type Runner struct {
+	Name string
+	Run  func(context.Context) error
+}
+
+type runnerResult struct {
+	name string
+	err  error
+}
+
+func RunConcurrent(ctx context.Context, runners ...Runner) 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
+}

internal/app/runner_test.go

@@ -0,0 +1,74 @@
+package app_test
+
+import (
+	"context"
+	"errors"
+	"mxl-pattern-generator/internal/app"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestRunConcurrentCancelsSiblingAndWaitsForCleanup(t *testing.T) {
+	wantErr := errors.New("writer failed")
+	peerStarted := make(chan struct{})
+	peerStopped := make(chan struct{})
+
+	err := app.RunConcurrent(context.Background(),
+		app.Runner{
+			Name: "video",
+			Run: func(ctx context.Context) error {
+				<-peerStarted
+				return wantErr
+			},
+		},
+		app.Runner{
+			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 <- app.RunConcurrent(ctx, app.Runner{
+			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")
+	}
+}