diff --git a/main.go b/main.go index 5cacef1..baff566 100644 --- a/main.go +++ b/main.go @@ -36,22 +36,28 @@ func run(ctx context.Context, cfg Config) error { select { case <-ctx.Done(): return ctx.Err() - case err := <-processSlackToMessagePipeline(ctx, cfg): + case err := <-processPipelines(ctx, cfg.slackToMessagePipeline): return err case err := <-listenAndServe(ctx, cfg): return err } } -func processSlackToMessagePipeline(ctx context.Context, cfg Config) chan error { +func processPipelines(ctx context.Context, first Pipeline, pipelines ...Pipeline) chan error { + ctx, can := context.WithCancel(ctx) + defer can() + + pipelines = append(pipelines, first) errs := make(chan error) - go func() { - defer close(errs) - select { - case errs <- cfg.slackToMessagePipeline.Process(ctx): - case <-ctx.Done(): - } - }() + for i := range pipelines { + go func(i int) { + defer close(errs) + select { + case errs <- pipelines[i].Process(ctx): + case <-ctx.Done(): + } + }(i) + } return errs }