itten/go-mxl-pattern-generator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

runner.go

Browse Source
This commit is contained in:
Dmitry Sergeev
2026-09-17 20:30:00 +03:00
parent d341d22611
commit d3f0b533e3
4 changed files with 125 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mxl-pattern/main.go
+8 -43
View File
@@ -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...)
}
cmd/mxl-pattern/main_test.go
-67
View File
@@ -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/runner.go
+43
View File
@@ -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
+74
View File
@@ -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")
}
}