From d3f0b533e30db4debcd0c6d6f685a6ebe3bb5d65 Mon Sep 17 00:00:00 2001 From: Dmitry Sergeev Date: Thu, 17 Sep 2026 20:30:00 +0300 Subject: [PATCH] runner.go --- cmd/mxl-pattern/main.go | 51 ++++--------------------- cmd/mxl-pattern/main_test.go | 67 -------------------------------- internal/app/runner.go | 43 +++++++++++++++++++++ internal/app/runner_test.go | 74 ++++++++++++++++++++++++++++++++++++ 4 files changed, 125 insertions(+), 110 deletions(-) create mode 100644 internal/app/runner.go create mode 100644 internal/app/runner_test.go diff --git a/cmd/mxl-pattern/main.go b/cmd/mxl-pattern/main.go index d562e34..04eb502 100644 --- a/cmd/mxl-pattern/main.go +++ b/cmd/mxl-pattern/main.go @@ -19,6 +19,7 @@ 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" @@ -59,16 +60,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}, @@ -464,48 +455,22 @@ func run(ctx context.Context, args appArgs) (runErr error) { } }() - runners := []namedRunner{ + runners := []app.Runner{ { - name: "video", - run: func(ctx context.Context) error { + 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 { + runners = append(runners, app.Runner{ + 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 + return app.RunConcurrent(ctx, runners...) } diff --git a/cmd/mxl-pattern/main_test.go b/cmd/mxl-pattern/main_test.go index 639b6c5..6b226cd 100644 --- a/cmd/mxl-pattern/main_test.go +++ b/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") - } -} diff --git a/internal/app/runner.go b/internal/app/runner.go new file mode 100644 index 0000000..ba577f8 --- /dev/null +++ b/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 +} diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go new file mode 100644 index 0000000..8f07529 --- /dev/null +++ b/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") + } +}