conc_err_map.go

//go:generate pipeliner map(func(string) (string, error) concurrently as concErrMap into conc_err_map.go
func concErrMap(concurrency int, fn func(string) (string, error), in <-chan string) (<-chan string, <-chan error) {
	out := make(chan string)
	errc := make(chan error, 1)
	done := make(chan struct{})
	once := sync.Once{}
	go func() {
		defer close(out)
		var outerErr error
		defer func() {
			errc <- outerErr
		}()
		wg := sync.WaitGroup{}
		wg.Add(concurrency)
		for i := 0; i < concurrency; i++ {
			go func() {
				defer wg.Done()
				for {
					select {
					case <-done:
						return
					case item, ok := <-in:
						if !ok {
							return // end of channel
						}
						t, err := fn(item)
						if err != nil {
							once.Do(func() {
								outerErr = err
								close(done)
							})
							return
						}
						out <- t
					}
				}
			}()
		}
		wg.Wait()
	}()
	return out, errc
}