How to multiplex channel output in go

Question

I'm looking for a solution to multiplex some channel output in go.

I have a source of data which is a read from an io.Reader that I send to a single channel. On the other side I have a websocket request handler that reads from the channel. Now it happens that two clients create a websocket connection, both reading from the same channel but each of them only getting a part of the messages.

Code example (simplified):

func (b *Bootloader) ReadLog() (<-chan []byte, error) {
    if b.logCh != nil {
        logrus.Warn("ReadLog called while channel already exists!")
        return b.logCh, nil // This is where we get problems
    }

    b.logCh = make(chan []byte, 0)

    go func() {
        buf := make([]byte, 1024)
        for {
            n, err := b.p.Read(buf)

            if err == nil {
                msg := make([]byte, n)
                copy(msg, buf[:n])
                b.logCh <- msg
            } else {
                break
            }
        }

        close(b.logCh)
        b.logCh = nil
    }()

    return b.logCh, nil
}

Now when ReadLog() is called twice, the second call just returns the channel created in the first call, which leads to the problem explained above.

The question is: How to do proper multiplexing?

Is it better/easier/more ideomatic to care about the multiplexing on the sending or receiving site?

Should I hide the channel from the receiver and work with callbacks?

I'm a little stuck at the moment. Any hints are welcome.

Answer 1

Mutiplexing is pretty straightforward: make a slice of channels you want to multiplex to, start up a goroutine that reads from the original channel and copies each message to each channel in the slice:

// Really this should be in Bootloader but this is just an example
var consumers []chan []byte

func (b *Bootloader) multiplex() {
    // We'll use a sync.once to make sure we don't start a bunch of these.
    sync.Once(func(){ 
        go func() {
            // Every time a message comes over the channel...
            for v := range b.logCh {
                // Loop over the consumers...
                for _,cons := range consumers {
                    // Send each one the message
                    cons <- v
                }
            }
        }()
    })
}