code: func Merge(consumers ...Consumer) *Mx {
  	p := &Mx{
  		errc:    make(chan error),
  		msgc:    make(chan *proto.Message),
  		stop:    make(chan struct{}),
  		workers: len(consumers),
  	}
  
  	for _, consumer := range consumers {
  		go func(c Consumer) {
  			defer func() {
  				p.mu.Lock()
  				p.workers--
  				if p.workers == 0 && !p.closed {
  					close(p.stop)
  					p.closed = true
  				}
  				p.mu.Unlock()
  			}()
  
  			for {
  				msg, err := c.Consume()
  				if err != nil {
  					if err == ErrNoData {
  						return
  					}
  					select {
  					case p.errc <- err:
  					case <-p.stop:
  						return
  					}
  				} else {
  					select {
  					case p.msgc <- msg:
  					case <-p.stop:
  						return
  					}
  				}
  			}
  		}(consumer)
  	}
  
  	return p
  }
docs: // Merge is merging consume result of any number of consumers into single stream
  // and expose them through returned multiplexer.