Слияние нескольких потоков в единый канал

Условие

Необходимо написать функцию, которая собирает данные из произвольного количества входных каналов и выводит их в один результирующий канал. Каждый из входных каналов может заполняться отдельными горутинами, а задача функции — корректно объединить все поступающие значения и закрыть выходной канал только после того, как завершится работа всех источников.

func fanIn(channels ...<-chan int) <-chan int {
    out
    // реализация объединения
    return out
}

Функция должна обеспечить безопасный доступ к общему потоку данных и корректно завершаться, когда все входные каналы закрыты.

Ответ

На каждый непустой вход запустите читателя, а выход закройте отдельной горутиной после WaitGroup. Отправка нескольких горутин в один канал безопасна без дополнительного mutex.

С импортом sync:

func fanIn(channels ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    for _, ch := range channels {
        if ch == nil { continue }
        wg.Add(1)
        go func(input <-chan int) {
            defer wg.Done()
            for value := range input { out <- value }
        }(ch)
    }
    go func() { wg.Wait(); close(out) }()
    return out
}

Nil-источники пропускаются; пустой набор даёт закрытый выход. Порядок между входами не гарантирован. Без context потребитель обязан дочитать поток, а отправители — закрыть входы, иначе возможны ожидающие горутины.