code: func (b *Broker) consume(deliveries <-chan amqp.Delivery, concurrency int, taskProcessor iface.TaskProcessor, amqpCloseChan <-chan *amqp.Error) error {
  	pool := make(chan struct{}, concurrency)
  
  	// initialize worker pool with maxWorkers workers
  	go func() {
  		for i := 0; i < concurrency; i++ {
  			pool <- struct{}{}
  		}
  	}()
  
  	errorsChan := make(chan error)
  
  	for {
  		select {
  		case amqpErr := <-amqpCloseChan:
  			return amqpErr
  		case err := <-errorsChan:
  			return err
  		case d := <-deliveries:
  			if concurrency > 0 {
  				// get worker from pool (blocks until one is available)
  				<-pool
  			}
  
  			b.processingWG.Add(1)
  
  			// Consume the task inside a gotourine so multiple tasks
  			// can be processed concurrently
  			go func() {
  				if err := b.consumeOne(d, taskProcessor); err != nil {
  					errorsChan <- err
  				}
  
  				b.processingWG.Done()
  
  				if concurrency > 0 {
  					// give worker back to pool
  					pool <- struct{}{}
  				}
  			}()
  		case <-b.GetStopChan():
  			return nil
  		}
  	}
  }
docs: // consume takes delivered messages from the channel and manages a worker pool
  // to process tasks concurrently