func (pub *Publisher) startAutoPublishingLoop() {
AutoPublishLoop:
	for {
		select {
		case stop := <-pub.autoStop:
			if stop {
				break AutoPublishLoop
			}
		default:
			break
		}
		if pub.deliverLetters() {
			break AutoPublishLoop
		}
	}
	pub.pubLock.Lock()
	pub.autoStarted = false
	pub.pubLock.Unlock()
}
StartAutoPublish starts auto-publishing letters queued up - is locking.
package tcr
import (
	"context"
	"fmt"
	"sync"
	"time"
	"github.com/streadway/amqp"
)
type Publisher struct {
	Config                 *RabbitSeasoning
	ConnectionPool         *ConnectionPool
	letters                chan *Letter
	autoStop               chan bool
	publishReceipts        chan *PublishReceipt
	autoStarted            bool
	autoPublishGroup       *sync.WaitGroup
	sleepOnIdleInterval    time.Duration
	sleepOnErrorInterval   time.Duration
	publishTimeOutDuration time.Duration
	pubLock                *sync.Mutex
	pubRWLock              *sync.RWMutex
}
func NewPublisherFromConfig(
	config *RabbitSeasoning,
	cp *ConnectionPool) *Publisher {
	if config.PublisherConfig.MaxRetryCount == 0 {
		config.PublisherConfig.MaxRetryCount = 5
	}
	return &Publisher{
		Config:                 config,
		ConnectionPool:         cp,
		letters:                make(chan *Letter, 1000),
		autoStop:               make(chan bool, 1),
		autoPublishGroup:       &sync.WaitGroup{},
		publishReceipts:        make(chan *PublishReceipt, 1000),
		sleepOnIdleInterval:    time.Duration(config.PublisherConfig.SleepOnIdleInterval) * time.Millisecond,
		sleepOnErrorInterval:   time.Duration(config.PublisherConfig.SleepOnErrorInterval) * time.Millisecond,
		publishTimeOutDuration: time.Duration(config.PublisherConfig.PublishTimeOutInterval) * time.Millisecond,
		pubLock:                &sync.Mutex{},
		pubRWLock:              &sync.RWMutex{},
		autoStarted:            false,
	}
}
func NewPublisher(
	cp *ConnectionPool,
	sleepOnIdleInterval time.Duration,
	sleepOnErrorInterval time.Duration,
	publishTimeOutDuration time.Duration) *Publisher {
	return &Publisher{
		ConnectionPool:         cp,
		letters:                make(chan *Letter, 1000),
		autoStop:               make(chan bool, 1),
		autoPublishGroup:       &sync.WaitGroup{},
		publishReceipts:        make(chan *PublishReceipt, 1000),
		sleepOnIdleInterval:    sleepOnIdleInterval,
		sleepOnErrorInterval:   sleepOnErrorInterval,
		publishTimeOutDuration: publishTimeOutDuration,
		pubLock:                &sync.Mutex{},
		pubRWLock:              &sync.RWMutex{},
		autoStarted:            false,
	}
}
func (pub *Publisher) Publish(letter *Letter, skipReceipt bool) {
	chanHost := pub.ConnectionPool.GetChannelFromPool()
	err := chanHost.Channel.Publish(
		letter.Envelope.Exchange,
		letter.Envelope.RoutingKey,
		letter.Envelope.Mandatory,
		letter.Envelope.Immediate,
		amqp.Publishing{
			ContentType:  letter.Envelope.ContentType,
			Body:         letter.Body,
			Headers:      letter.Envelope.Headers,
			DeliveryMode: letter.Envelope.DeliveryMode,
			Priority:     letter.Envelope.Priority,
			MessageId:    letter.LetterID.String(),
			Timestamp:    time.Now().UTC(),
			AppId:        pub.Config.PoolConfig.ApplicationName,
		},
	)
	if !skipReceipt {
		pub.publishReceipt(letter, err)
	}
	pub.ConnectionPool.ReturnChannel(chanHost, err != nil)
}
func (pub *Publisher) PublishWithTransient(letter *Letter) error {
	channel := pub.ConnectionPool.GetTransientChannel(false)
	defer func() {
		defer func() {
			_ = recover()
		}()
		channel.Close()
	}()
	return channel.Publish(
		letter.Envelope.Exchange,
		letter.Envelope.RoutingKey,
		letter.Envelope.Mandatory,
		letter.Envelope.Immediate,
		amqp.Publishing{
			ContentType:  letter.Envelope.ContentType,
			Body:         letter.Body,
			Headers:      letter.Envelope.Headers,
			DeliveryMode: letter.Envelope.DeliveryMode,
			Priority:     letter.Envelope.Priority,
			MessageId:    letter.LetterID.String(),
			Timestamp:    time.Now().UTC(),
			AppId:        pub.Config.PoolConfig.ApplicationName,
		},
	)
}
func (pub *Publisher) PublishWithConfirmation(letter *Letter, timeout time.Duration) {
	if timeout == 0 {
		timeout = pub.publishTimeOutDuration
	}
	for {
		chanHost := pub.ConnectionPool.GetChannelFromPool()
		chanHost.FlushConfirms()  
	Publish:
		timeoutAfter := time.After(timeout)  
		err := chanHost.Channel.Publish(
			letter.Envelope.Exchange,
			letter.Envelope.RoutingKey,
			letter.Envelope.Mandatory,
			letter.Envelope.Immediate,
			amqp.Publishing{
				ContentType:  letter.Envelope.ContentType,
				Body:         letter.Body,
				Headers:      letter.Envelope.Headers,
				DeliveryMode: letter.Envelope.DeliveryMode,
				Priority:     letter.Envelope.Priority,
				MessageId:    letter.LetterID.String(),
				Timestamp:    time.Now().UTC(),
				AppId:        pub.Config.PoolConfig.ApplicationName,
			},
		)
		if err != nil {
			pub.ConnectionPool.ReturnChannel(chanHost, true)
			continue  
		}
		for {
			select {
			case <-timeoutAfter:
				pub.publishReceipt(letter, fmt.Errorf("publish confirmation for LetterID: %s wasn't received in a timely manner - recommend retry/requeue", letter.LetterID.String()))
				pub.ConnectionPool.ReturnChannel(chanHost, false)  
				return
			case confirmation := <-chanHost.Confirmations:
				if !confirmation.Ack {
					goto Publish  
				}
				pub.publishReceipt(letter, nil)
				pub.ConnectionPool.ReturnChannel(chanHost, false)
				return
			default:
				time.Sleep(time.Duration(time.Millisecond * 1))  
			}
		}
	}
}
func (pub *Publisher) PublishWithConfirmationContext(ctx context.Context, letter *Letter) {
	for {
		chanHost := pub.ConnectionPool.GetChannelFromPool()
		chanHost.FlushConfirms()  
	Publish:
		err := chanHost.Channel.Publish(
			letter.Envelope.Exchange,
			letter.Envelope.RoutingKey,
			letter.Envelope.Mandatory,
			letter.Envelope.Immediate,
			amqp.Publishing{
				ContentType:  letter.Envelope.ContentType,
				Body:         letter.Body,
				Headers:      letter.Envelope.Headers,
				DeliveryMode: letter.Envelope.DeliveryMode,
				Priority:     letter.Envelope.Priority,
				MessageId:    letter.LetterID.String(),
				Timestamp:    time.Now().UTC(),
				AppId:        pub.Config.PoolConfig.ApplicationName,
			},
		)
		if err != nil {
			pub.ConnectionPool.ReturnChannel(chanHost, true)
			continue  
		}
		for {
			select {
			case <-ctx.Done():
				pub.publishReceipt(letter, fmt.Errorf("publish confirmation for LetterID: %s wasn't received before context expired - recommend retry/requeue", letter.LetterID.String()))
				pub.ConnectionPool.ReturnChannel(chanHost, false)  
				return
			case confirmation := <-chanHost.Confirmations:
				if !confirmation.Ack {
					goto Publish  
				}
				pub.publishReceipt(letter, nil)
				pub.ConnectionPool.ReturnChannel(chanHost, false)
				return
			default:
				time.Sleep(time.Duration(time.Millisecond * 1))  
			}
		}
	}
}
func (pub *Publisher) PublishWithConfirmationTransient(letter *Letter, timeout time.Duration) {
	if timeout == 0 {
		timeout = pub.publishTimeOutDuration
	}
	for {
		channel := pub.ConnectionPool.GetTransientChannel(true)
		confirms := make(chan amqp.Confirmation, 1)
		channel.NotifyPublish(confirms)
	Publish:
		timeoutAfter := time.After(timeout)
		err := channel.Publish(
			letter.Envelope.Exchange,
			letter.Envelope.RoutingKey,
			letter.Envelope.Mandatory,
			letter.Envelope.Immediate,
			amqp.Publishing{
				ContentType:  letter.Envelope.ContentType,
				Body:         letter.Body,
				Headers:      letter.Envelope.Headers,
				DeliveryMode: letter.Envelope.DeliveryMode,
				Priority:     letter.Envelope.Priority,
				MessageId:    letter.LetterID.String(),
				Timestamp:    time.Now().UTC(),
				AppId:        pub.Config.PoolConfig.ApplicationName,
			},
		)
		if err != nil {
			channel.Close()
			if pub.sleepOnErrorInterval < 0 {
				time.Sleep(pub.sleepOnErrorInterval)
			}
			continue  
		}
		for {
			select {
			case <-timeoutAfter:
				pub.publishReceipt(letter, fmt.Errorf("publish confirmation for LetterID: %s wasn't received in a timely manner (%dms) - recommend retry/requeue", letter.LetterID.String(), timeout))
				channel.Close()
				return
			case confirmation := <-confirms:
				if !confirmation.Ack {
					goto Publish  
				}
				pub.publishReceipt(letter, nil)
				channel.Close()
				return
			default:
				time.Sleep(time.Duration(time.Millisecond * 4))  
			}
		}
	}
}
func (pub *Publisher) PublishReceipts() <-chan *PublishReceipt {
	return pub.publishReceipts
}
func (pub *Publisher) StartAutoPublishing() {
	pub.pubLock.Lock()
	defer pub.pubLock.Unlock()
	if !pub.autoStarted {
		pub.autoStarted = true
		go pub.startAutoPublishingLoop()
	}
}