code: func (b *Broker) consume(deliveries <-chan []byte, pool chan struct{}, concurrency int, taskProcessor iface.TaskProcessor) error {
  	errorsChan := make(chan error, concurrency*2)
  
  	for {
  		select {
  		case err := <-errorsChan:
  			return err
  		case d := <-deliveries:
  			b.processingWG.Add(1)
  
  			// Consume the task inside a goroutine so multiple tasks
  			// can be processed concurrently
  			go func() {
  				if err := b.consumeOne(d, taskProcessor); err != nil {
  					errorsChan <- err
  				}
  
  				b.processingWG.Done()
  			}()
  		case <-b.Broker.GetStopChan():
  			return nil
  		}
  	}
  }
docs: // consume takes delivered messages from the channel and manages a worker pool
  // to process tasks concurrently