1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
package interactors
import (
"sync"
"github.com/pkg/errors"
)
type taskRunner struct {
errorChan chan error
sync.Mutex
sync.WaitGroup
}
func NewTaskRunner() *taskRunner {
return &taskRunner{}
}
func (t *taskRunner) ExecuteFunctionsAsync(functions []func() error) error {
t.Lock()
defer t.Unlock()
t.errorChan = make(chan error)
t.spawnTasksAsync(
t.syncErrorFnToAsync(functions),
)
go t.closeErrorChanOnComplete()
return t.breakOnError()
}
func (t *taskRunner) breakOnError() error {
return <-t.errorChan
}
func (t *taskRunner) syncErrorFnToAsync(functions []func() error) []func() {
transformed := make([]func(), len(functions))
for i, fn := range functions {
transformed[i] = t.redirectErrToChannel(fn)
}
return transformed
}
func (t *taskRunner) closeErrorChanOnComplete() {
t.Wait()
close(t.errorChan)
}
func (t *taskRunner) spawnTasksAsync(tasks []func()) {
t.Add(len(tasks))
for _, task := range tasks {
go task()
}
}
func (t *taskRunner) redirectErrToChannel(f func() error) func() {
return func() {
defer t.Done()
err := f()
if err != nil && t.errorChan != nil {
t.errorChan <- errors.Wrap(err, "Error redirected to channel")
}
}
}
|
