| 63 | } |
| 64 | |
| 65 | func MergeErrors(ctx context.Context, cs ...<-chan error) chan error { |
| 66 | var wg sync.WaitGroup |
| 67 | // We must ensure that the output channel has the capacity to |
| 68 | // hold as many errors |
| 69 | // as there are error channels. |
| 70 | // This will ensure that it never blocks, even |
| 71 | // if WaitForPipeline returns early. |
| 72 | out := make(chan error, len(cs)) |
| 73 | // Start an output goroutine for each input channel in cs. output |
| 74 | // copies values from c to out until c is closed, then calls |
| 75 | // wg.Done. |
| 76 | output := func(c <-chan error) { |
| 77 | for n := range c { |
| 78 | select { |
| 79 | case <-ctx.Done(): |
| 80 | return |
| 81 | case out <- n: |
| 82 | } |
| 83 | } |
| 84 | wg.Done() |
| 85 | } |
| 86 | wg.Add(len(cs)) |
| 87 | for _, c := range cs { |
| 88 | go output(c) |
| 89 | } |
| 90 | // Start a goroutine to close out once all the output goroutines |
| 91 | // are done. This must start after the wg.Add call. |
| 92 | go func() { |
| 93 | wg.Wait() |
| 94 | close(out) |
| 95 | }() |
| 96 | return out |
| 97 | } |