func (session *Session) Close() error {
	var err error
	log.Debug(session.logger, "start shutdown of rabbit session")
	session.mu.Lock()
	if !session.config.PublishOnly {
		<-session.done
	}
	session.config.ConsumerConnectionPool.ReturnChannel(session.consumerchannelhost, false)
	close(session.messages)
	close(session.done)
	session.consumerchannelhost = nil
	session.publisherchannelhost = nil
	session.mu.Unlock()
	log.Debug(session.logger, "...waiting for shutdown of rabbit session")
	log.Debug(session.logger, "rabbit session closed")
	return err
}
Close will cleanly shutdown the channel and connection.
package rabbitmqsmartshovel
import (
	"context"
	"errors"
	"github.com/pinpt/go-common/v10/rabbitmq"
	"strings"
	"sync"
	"time"
	"github.com/pinpt/go-common/v10/log"
	pstrings "github.com/pinpt/go-common/v10/strings"
	"github.com/streadway/amqp"
)
type Config struct {
	Name                        string
	ID                          string
	Exchange                    string
	IngestExchange              string
	SubscriptionExchange        string
	IsShovel                    bool
	IsSubscriptionQueueListener bool
	ConsumerConnectionPool      *rabbitmq.ConnectionPool
	PublisherConnectionPool     *rabbitmq.ConnectionPool
	AutoAck                     bool
	DurableQueue                bool
	DeleteUnused                bool
	Exclusive                   bool
	Args                        amqp.Table
	Qos                         int
	PublishOnly                 bool
	Context                     context.Context
}
type Session struct {
	messages             chan amqp.Delivery
	logger               log.Logger
	config               *Config
	name                 string
	consumerchannelhost  *rabbitmq.ChannelHost
	publisherchannelhost *rabbitmq.ChannelHost
	done                 chan bool
	mu                   sync.RWMutex
	bindings             []string
	autocommit           bool
	inflightmessages     map[uint64]bool
}
const (
	reconnectDelay = 2 * time.Second
	reInitDelay = 10 * time.Millisecond
	sendTimeout = 5 * time.Second
)
var (
	ErrNotConnected = errors.New("not connected to a server")
	ErrAlreadyClosed = errors.New("already closed: not connected to the server")
	ErrShutdown = errors.New("session is shutting down")
	ErrNack = errors.New("message was not sent")
	ErrServerBusy = errors.New("server busy; message was not sent")
	ErrPublishOnly = errors.New("channel is publish only")
	ErrTimedOut = errors.New("confirmation timed out")
)
func New(logger log.Logger, config Config) *Session {
	if config.Name == "" {
		config.Name = pstrings.NewUUIDV4()
	}
	if config.Qos <= 0 {
		config.Qos = 1
	}
	session := Session{
		logger:   log.With(logger, "pkg", "rabbitmq", "name", config.Name),
		config:   &config,
		done:     make(chan bool),
		messages: make(chan amqp.Delivery, 1),
	}
	session.ensureConsumerChannel()
	return &session
}
func (session *Session) ensureQueueAndBindings() error {
	if session.config.PublishOnly {
		return ErrPublishOnly
	}
	channel, err := session.config.ConsumerConnectionPool.GetTransientChannel(false)
	if err != nil {
		log.Error(session.logger, "error getting channel to create queue", "err", err)
	}
	log.Info(session.logger, "staring queue create", "queuename", session.config.Name)
	if _, err := channel.QueueDeclare(
		session.config.Name,
		session.config.DurableQueue,  
		session.config.DeleteUnused,  
		session.config.Exclusive,     
		false,                        
		session.config.Args,          
	); err != nil {
		log.Error(session.logger, "error declaring queue", "err", err)
		channel.Close()
		return err
	}
	exchange := session.config.Exchange
	if session.config.IsShovel {
		exchange = session.config.IngestExchange
	} else if session.config.IsSubscriptionQueueListener {
		exchange = session.config.SubscriptionExchange
	}
	if err := session.consumerchannelhost.Channel.QueueBind(
		session.config.Name,  
		session.config.Name,  
		exchange,             
		false,
		nil,
	); err != nil {
		log.Error(session.logger, "error binding routing key", "err", err, "routingKey", session.config.Name, "exchange", session.config.Exchange)
		channel.Close()
		return err
	}
	log.Info(session.logger, "setup and ready")
	channel.Close()
	return nil
}
func (session *Session) ensurePublisherChannel() *rabbitmq.ChannelHost {
	for {
		chanHost, err := session.config.PublisherConnectionPool.GetChannel()
		log.Debug(session.logger, "getting channel from pool to publish")
		if err != nil {
			log.Error(session.logger, "error with getting channel...retrying", "err", err)
			session.config.PublisherConnectionPool.ReturnChannel(chanHost, true)  
			chanHost = nil
			time.Sleep(reInitDelay)
			continue
		} else {
			session.mu.Lock()
			session.publisherchannelhost = chanHost
			session.mu.Unlock()
			return chanHost
		}
	}
}
func (session *Session) ensureConsumerChannel() {
	for {
		chanHost, err := session.config.ConsumerConnectionPool.GetChannel()
		log.Debug(session.logger, "getting channel from pool to consume")
		if err != nil {
			log.Error(session.logger, "error with getting channel...retrying", "err", err)
			session.config.ConsumerConnectionPool.ReturnChannel(chanHost, true)
			chanHost = nil
			time.Sleep(reInitDelay)
			continue
		} else {
			session.mu.Lock()
			session.consumerchannelhost = chanHost
			prefetch := session.config.Qos
			session.mu.Unlock()
			session.consumerchannelhost.Channel.Qos(prefetch, 0, false)
			session.ensureQueueAndBindings()
			return
		}
	}
}
func (session *Session) Push(routingKey string, data amqp.Publishing) error {
	exchange := session.config.IngestExchange
	if session.config.IsShovel {
		exchange = session.config.Exchange
	} else if session.config.IsSubscriptionQueueListener {
		exchange = session.config.SubscriptionExchange
		routingKey = session.config.SubscriptionExchange
	} else {
		routingKey = session.config.IngestExchange
	}
	return session.config.PublisherConnectionPool.Push(exchange, routingKey, data)
}
func (session *Session) Stream(consumergroup string, autoAck bool, exclusive bool) (chan amqp.Delivery, error) {
	if session.config.PublishOnly {
		return nil, ErrPublishOnly
	}
	session.mu.Lock()
	session.autocommit = autoAck
	session.mu.Unlock()
	go session.startConsumeLoop(consumergroup, autoAck, exclusive)
	return session.messages, nil
}
func (session *Session) startConsumeLoop(consumergroup string, autoAck bool, exclusive bool) {
	log.Info(session.logger, "starting consume loop for", "consumer", consumergroup)
	for {
		log.Info(session.logger, "starting connection for", "consumer", consumergroup)
		deliveryChan, err := session.consumerchannelhost.Channel.Consume(
			session.config.Name,
			consumergroup,  
			autoAck,        
			exclusive,      
			false,          
			false,          
			nil,            
		)
		log.Info(session.logger, "consuming", "consumer", consumergroup)
		if err != nil {
			if strings.Contains(err.Error(), "PRECONDITION_FAILED") {
				ch, _ := session.config.ConsumerConnectionPool.GetTransientChannel(false)
				log.Warn(session.logger, "queue precondition failed, will try and delete and try again", "err", err, "queue", session.config.Name)
				ch.QueueDelete(session.config.Name, false, true, false)
				ch.Close()
				continue
			}
			session.config.ConsumerConnectionPool.ReturnChannel(session.consumerchannelhost, true)
			session.ensureConsumerChannel()
			log.Error(session.logger, "consumer's current channel closed", "err", err)
			continue
		}
		if session.processDeliveries(deliveryChan, session.consumerchannelhost) {
			log.Debug(session.logger, "process deliveries exited")
			err := session.consumerchannelhost.Channel.Cancel(
				consumergroup,  
				true,           
			)
			if err != nil {
				log.Error(session.logger, "error closing channel on exit", "err", err)
			}
			session.done <- true
			return
		}
		log.Error(session.logger, "error occurred processing deliveries, attempting to reconnect")
	}
}
func (session *Session) processDeliveries(deliveryChan <-chan amqp.Delivery, chanHost *rabbitmq.ChannelHost) bool {
	for {
		select {
		case errorMessage := <-session.consumerchannelhost.Errors:
			if errorMessage != nil {
				log.Error(session.logger, "consumer's current channel errored...starting reconnect", "err", errorMessage, "reason", errorMessage.Reason, "code", errorMessage.Code)
				session.config.ConsumerConnectionPool.ReturnChannel(chanHost, true)
				session.ensureConsumerChannel()
				return false
			}
		case <-session.config.Context.Done():
			log.Debug(session.logger, "consumer context cancelled")
			return true
		case delivery := <-deliveryChan:
			session.messages <- delivery
		}
	}
}
func (session *Session) Ack(tag uint64) error {
	log.Debug(session.logger, "ack", "tag", tag, "autocommit", session.autocommit)
	session.mu.RLock()
	if session.autocommit {
		session.mu.RUnlock()
		return nil
	}
	session.mu.RUnlock()
	err := session.consumerchannelhost.Channel.Ack(tag, false)
	return err
}
func (session *Session) Nack(tag uint64) error {
	log.Debug(session.logger, "nack", "tag", tag, "autocommit", session.autocommit)
	session.mu.RLock()
	if session.autocommit {
		session.mu.RUnlock()
		return nil
	}
	session.mu.RUnlock()
	err := session.consumerchannelhost.Channel.Nack(tag, false, true)
	return err
}