code: func (mux *Multiplexer) propagateMessage(msg *client.ConsumerMessage) {
  	mux.rwlock.RLock()
  	defer mux.rwlock.RUnlock()
  
  	if msg == nil {
  		return
  	}
  
  	// Find subscribed topics. Note: topic can be subscribed for both dynamic and manual consuming
  	for _, subscription := range mux.mapping {
  		if msg.Topic == subscription.topic {
  			// Clustered mode - message is consumed only on right partition and offset
  			if subscription.manual {
  				if msg.Partition == subscription.partition && msg.Offset >= subscription.offset {
  					mux.Debug("offset ", msg.Offset, string(msg.Value), string(msg.Key), msg.Partition)
  					subscription.byteConsMsg(msg)
  				}
  			} else {
  				// Non-manual mode
  				// if we are not able to write into the channel we should skip the receiver
  				// and report an error to avoid deadlock
  				mux.Debug("offset ", msg.Offset, string(msg.Value), string(msg.Key), msg.Partition)
  				subscription.byteConsMsg(msg)
  			}
  		}
  	}
  }
docs: // Propagates incoming messages to respective channels.