channel/utils.go

package channel

func determineBufferSize[T any](channels []<-chan T) int {
	if len(channels) == 0 {
		return 0
	}

	maxBufSize := 0
	for _, ch := range channels {
		if cap(ch) > maxBufSize {
			maxBufSize = cap(ch)
		}
	}
	return maxBufSize
}

// Flush consumes all values and discards them immediately.
// It blocks until all sources are closed
func Flush[T any](sources ...<-chan T) {
	for range Merge(sources...) {
	}
}

// Each consumes all values and calls f for each of them.
// It blocks until source is closed
func Each[T any](source <-chan T, f func(T)) {
	for value := range source {
		f(value)
	}
}

// Tee returns 2 channels which both receive all values from source.
// It's basically a copy function for channels
func Tee[T any](source <-chan T) (<-chan T, <-chan T) {
	outs := TeeMany(source, 2)
	return outs[0], outs[1]
}

// TeeMany returns a given amount of channels which all receive all values from source.
// It's basically a copy function for channels
func TeeMany[T any](source <-chan T, amount int) []<-chan T {
	outputs := make([]chan T, amount)
	for i := range outputs {
		outputs[i] = make(chan T, cap(source))
	}

	go func() {
		defer func() {
			for _, out := range outputs {
				close(out)
			}
		}()

		for value := range source {
			for _, out := range outputs {
				out <- value
			}
		}
	}()

	readOnlyOutputs := make([]<-chan T, 0, amount)
	for _, out := range outputs {
		readOnlyOutputs = append(readOnlyOutputs, out)
	}
	return readOnlyOutputs
}