channel/timeout.go

package channel

import "time"

// CloseOnTimeout returns a channel which receives all values from the source.
// If no value was received in the given timeout duration, the returned channel will be closed
func CloseOnTimeout[T any](source <-chan T, timeout time.Duration) <-chan T {
	output := make(chan T, cap(source))

	go func() {
		defer close(output)

		for {
			timer := time.NewTimer(timeout)

			select {
			case value, ok := <-source:
				if !ok {
					return
				}
				output <- value
			case <-timer.C:
				return
			}
		}
	}()

	return output
}