fan_in_chan.go

package chanutil

import (
  "context"
  "sync"
)

func fanIn[T any](ctx context.Context, size int, chans ...<-chan T) <-chan T {
	var wg sync.WaitGroup
	copy := func(i <-chan T, o chan<- T) {
	COPY_LOOP:
		for {
			select {
			case v, ok := <-i:
				if !ok {
					break COPY_LOOP
				}
				select {
				case <-ctx.Done():
					break COPY_LOOP
				case o <- v:
				}
			case <-ctx.Done():
				break COPY_LOOP
			}
		}
		wg.Done()
	}
	out := make(chan T, size)
	for _, c := range chans {
		wg.Add(1)
		go copy(c, out)
	}
	go func() {
		wg.Wait()
		close(out)
	}()
	return out
}