Note - newbie in Go.
I've written a multiplexer that should merge the outputs of an array of channels into one. Happy with constructive criticism.
func Mux(channels []chan big.Int) chan big.Int {
// Count down as each channel closes. When hits zero - close ch.
n := len(channels)
// The channel to output to.
ch := make(chan big.Int, n)
// Make one go per channel.
for _, c := range channels {
go func() {
// Pump it.
for x := range c {
ch <- x
}
// It closed.
n -= 1
// Close output if all closed now.
if n == 0 {
close(ch)
}
}()
}
return ch
}
I am testing it with:
func fromTo(f, t int) chan big.Int {
ch := make(chan big.Int)
go func() {
for i := f; i < t; i++ {
fmt.Println("Feed:", i)
ch <- *big.NewInt(int64(i))
}
close(ch)
}()
return ch
}
func testMux() {
r := make([]chan big.Int, 10)
for i := 0; i < 10; i++ {
r[i] = fromTo(i*10, i*10+10)
}
all := Mux(r)
// Roll them out.
for l := range all {
fmt.Println(l)
}
}
but my output is very strange:
Feed: 0
Feed: 10
Feed: 20
Feed: 30
Feed: 40
Feed: 50
Feed: 60
Feed: 70
Feed: 80
Feed: 90
Feed: 91
Feed: 92
Feed: 93
Feed: 94
Feed: 95
Feed: 96
Feed: 97
Feed: 98
Feed: 99
{false [90]}
{false [91]}
{false [92]}
{false [93]}
{false [94]}
{false [95]}
{false [96]}
{false [97]}
{false [98]}
{false [99]}
So to my questions:
- Is there something I am doing wrong in Mux?
- Why am I only getting the last 10 from my output channel?
- Why does the feeding look so strange? (1st of each input channel, all of the last channel and then nothing)
- Is there a better way of doing this?
I need all of the input channels to have equal rights to the output channel - i.e. I cannot have all of the output from one channel and then all from the next etc.
For anyone interested - this was the final code after the fix and the correct (presumably) use of sync.WaitGroup
import (
"math/big"
"sync"
)
/*
Multiplex a number of channels into one.
*/
func Mux(channels []chan big.Int) chan big.Int {
// Count down as each channel closes. When hits zero - close ch.
var wg sync.WaitGroup
wg.Add(len(channels))
// The channel to output to.
ch := make(chan big.Int, len(channels))
// Make one go per channel.
for _, c := range channels {
go func(c <-chan big.Int) {
// Pump it.
for x := range c {
ch <- x
}
// It closed.
wg.Done()
}(c)
}
// Close the channel when the pumping is finished.
go func() {
// Wait for everyone to be done.
wg.Wait()
// Close.
close(ch)
}()
return ch
}
See Question&Answers more detail:
os 与恶龙缠斗过久,自身亦成为恶龙;凝视深渊过久,深渊将回以凝视…