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

Условие

Даны два входных канала in1 и in2, а также один выходной канал out. Требуется написать функцию merge, которая будет одновременно прослушивать оба входных канала и перенаправлять получаемые из них значения в единый выходной канал.

package main

import (
 "fmt"
 "math/rand"
 "time"
)

func merge(ch ...<-chan int) <-chan int {
 out := make(chan int)
 // Здесь необходимо реализовать логику, которая будет ждать данные из всех переданных каналов
 // и отправлять их в канал out
 return out
}

func source(sourceFunc func(int) int) <-chan int {
 ch := make(chan int)
 go func() {
  defer close(ch)
  for i := 0; i < 10; i++ {
   ch <- sourceFunc(i)
   time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
  }
 }()
 return ch
}

func main() {
 rand.Seed(time.Now().UnixMilli())

 in1 := source(func(_ int) int {
  return rand.Intn(7)
 })

 in2 := source(func(i int) int {
  return i
 })

 out := merge(in1, in2)

 for val := range out {
  fmt.Println("Value:", val)
 }
}

Ответ

Запустите читателя каждого источника и пересылайте значения в out. Когда оба исходных канала закроются, WaitGroup позволит закрыть выход и завершить range в main.

Добавьте sync:

func merge(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
}

Получится 20 значений, порядок смешивания зависит от готовности источников. Nil-входы пропускаются. Source закрывает свои каналы; merge закрывает только собственный выход. Для раннего прекращения чтения нужен согласованный сигнал отмены.