func (con *Consumer) getDeliveryChannel() (<-chan amqp.Delivery, *pools.ChannelHost, error) {
	var chanHost *pools.ChannelHost
	var err error
	if con.autoAck {
		chanHost, err = con.channelPool.GetChannel()
	} else {
		chanHost, err = con.channelPool.GetAckableChannel()
	}
	if err != nil {
		con.handleError(err)
		return nil, nil, err
	}
	if con.qosCountOverride > 0 {
		err := chanHost.Channel.Qos(con.qosCountOverride, 0, false)
		if err != nil {
			return nil, nil, err
		}
	}
	deliveryChan, err := chanHost.Channel.Consume(con.QueueName, con.ConsumerName, con.autoAck, con.exclusive, false, con.noWait, nil)
	if err != nil {
		con.handleErrorAndChannel(err, chanHost)
		return nil, nil, err  
	}
	return deliveryChan, chanHost, nil
}
GetDeliveryChannel attempts to get the amqp.Delivery chan and a viable ChannelHost from the ChannelPool.
package consumer
import (
	"errors"
	"fmt"
	"sync"
	"time"
	"github.com/houseofcat/turbocookedrabbit/v1/pkg/models"
	"github.com/houseofcat/turbocookedrabbit/v1/pkg/pools"
	"github.com/streadway/amqp"
)
type Consumer struct {
	Config               *models.RabbitSeasoning
	channelPool          *pools.ChannelPool
	Enabled              bool
	QueueName            string
	ConsumerName         string
	errors               chan error
	sleepOnErrorInterval time.Duration
	sleepOnIdleInterval  time.Duration
	messageGroup         *sync.WaitGroup
	messages             chan *models.Message
	consumeStop          chan bool
	stopImmediate        bool
	started              bool
	autoAck              bool
	exclusive            bool
	noWait               bool
	args                 amqp.Table
	qosCountOverride     int
	conLock              *sync.Mutex
}
func NewConsumerFromConfig(
	config *models.ConsumerConfig,
	channelPool *pools.ChannelPool) (*Consumer, error) {
	if channelPool == nil {
		return nil, errors.New("can't start a consumer without a channel pool")
	} else if !channelPool.Initialized {
		err := channelPool.Initialize()
		if err != nil {
			return nil, err
		}
	}
	if config.MessageBuffer == 0 || config.ErrorBuffer == 0 {
		return nil, errors.New("message and/or error buffer in config can't be 0")
	}
	return &Consumer{
		Config:               nil,
		channelPool:          channelPool,
		Enabled:              config.Enabled,
		QueueName:            config.QueueName,
		ConsumerName:         config.ConsumerName,
		errors:               make(chan error, config.ErrorBuffer),
		sleepOnErrorInterval: time.Duration(config.SleepOnErrorInterval) * time.Millisecond,
		sleepOnIdleInterval:  time.Duration(config.SleepOnIdleInterval) * time.Millisecond,
		messageGroup:         &sync.WaitGroup{},
		messages:             make(chan *models.Message, config.MessageBuffer),
		consumeStop:          make(chan bool, 1),
		autoAck:              config.AutoAck,
		exclusive:            config.Exclusive,
		noWait:               config.NoWait,
		args:                 amqp.Table(config.Args),
		qosCountOverride:     config.QosCountOverride,
		conLock:              &sync.Mutex{},
	}, nil
}
func NewConsumer(
	config *models.RabbitSeasoning,
	channelPool *pools.ChannelPool,
	queuename string,
	consumerName string,
	autoAck bool,
	exclusive bool,
	noWait bool,
	args map[string]interface{},
	qosCountOverride int,  
	messageBuffer uint32,
	errorBuffer uint32,
	sleepOnErrorInterval uint32,
	sleepOnIdleInterval uint32) (*Consumer, error) {
	var err error
	if channelPool == nil {
		channelPool, err = pools.NewChannelPool(config.PoolConfig, nil, true)
		if err != nil {
			return nil, err
		}
	}
	if messageBuffer == 0 || errorBuffer == 0 {
		return nil, errors.New("message and/or error buffer can't be 0")
	}
	return &Consumer{
		Config:               config,
		channelPool:          channelPool,
		QueueName:            queuename,
		ConsumerName:         consumerName,
		errors:               make(chan error, errorBuffer),
		sleepOnErrorInterval: time.Duration(sleepOnErrorInterval) * time.Millisecond,
		sleepOnIdleInterval:  time.Duration(sleepOnIdleInterval) * time.Millisecond,
		messageGroup:         &sync.WaitGroup{},
		messages:             make(chan *models.Message, messageBuffer),
		consumeStop:          make(chan bool, 1),
		stopImmediate:        false,
		started:              false,
		autoAck:              autoAck,
		exclusive:            exclusive,
		noWait:               noWait,
		args:                 amqp.Table(args),
		qosCountOverride:     qosCountOverride,
		conLock:              &sync.Mutex{},
	}, nil
}
func (con *Consumer) Get(queueName string, autoAck bool) (*models.Message, error) {
	var chanHost *pools.ChannelHost
	var err error
	if autoAck {
		chanHost, err = con.channelPool.GetChannel()
	} else {
		chanHost, err = con.channelPool.GetAckableChannel()
	}
	if err != nil {
		return nil, err
	}
	amqpDelivery, ok, getErr := chanHost.Channel.Get(queueName, autoAck)
	if getErr != nil {
		con.channelPool.ReturnChannel(chanHost, true)
		return nil, getErr
	}
	if ok {
		return models.NewMessage(
			!autoAck,
			amqpDelivery.Body,
			amqpDelivery.DeliveryTag,
			chanHost.Channel), nil
	}
	con.channelPool.ReturnChannel(chanHost, false)
	return nil, nil
}
func (con *Consumer) GetBatch(queueName string, batchSize int, autoAck bool) ([]*models.Message, error) {
	if batchSize < 1 {
		return nil, errors.New("can't get a batch of messages whose size is less than 1")
	}
	var chanHost *pools.ChannelHost
	var err error
	if autoAck {
		chanHost, err = con.channelPool.GetChannel()
	} else {
		chanHost, err = con.channelPool.GetAckableChannel()
	}
	if err != nil {
		return nil, err
	}
	messages := make([]*models.Message, 0)
GetBatchLoop:
	for {
		if len(messages) == batchSize {
			break GetBatchLoop
		}
		amqpDelivery, ok, getErr := chanHost.Channel.Get(queueName, autoAck)
		if getErr != nil {
			con.channelPool.ReturnChannel(chanHost, true)
			return nil, getErr
		}
		if !ok {
			break GetBatchLoop
		}
		messages = append(messages, models.NewMessage(
			!autoAck,
			amqpDelivery.Body,
			amqpDelivery.DeliveryTag,
			chanHost.Channel))
	}
	return messages, nil
}
func (con *Consumer) StartConsuming() error {
	con.conLock.Lock()
	defer con.conLock.Unlock()
	if con.started {
		return errors.New("can't start an already started consumer")
	}
	if con.Enabled {
		con.FlushErrors()
		con.FlushStop()
		go con.startConsuming()
		con.started = true
	}
	return nil
}
func (con *Consumer) startConsuming() {
ConsumerOuterLoop:
	for {
		select {
		case stop := <-con.consumeStop:
			if stop {
				break ConsumerOuterLoop
			}
		default:
			break
		}
		deliveryChan, chanHost, err := con.getDeliveryChannel()
		if err != nil {
			continue  
		}
		if con.processDeliveries(deliveryChan, chanHost) {
			break ConsumerOuterLoop
		}
	}
	con.conLock.Lock()
	immediateStop := con.stopImmediate
	con.conLock.Unlock()
	if !immediateStop {
		con.messageGroup.Wait()  
	}
	con.conLock.Lock()
	con.started = false
	con.stopImmediate = false
	con.conLock.Unlock()
}