<file_sep>package consumer

import (
	"fmt"
	"time"
)

// Waiter ожидающий
type Waiter struct {
	*time.Ticker
}

// создает нового ожидающего
func newWaiter() *Waiter {
	waiter := &Waiter{time.NewTicker(time.Millisecond * 250)}
	go waiter.run()
	return waiter
}

// запускает нового ожидающего
func (w *Waiter) run() {
	commas := []string{
		".  ",
		" . ",
		"  .",
	}
	i := 0
	for {
		<-w.C
		fmt.Printf("\rgetting failure messages, please wait%s", commas[i])
		if i == 2 {
			i = 0
		} else {
			i++
		}
	}
}
<file_sep>package consumer

import (
	"encoding/json"
	"fmt"
	"regexp"
	"sync"

	"github.com/streadway/amqp"

	"github.com/Halfi/postmanq/common"
	"github.com/Halfi/postmanq/logger"
)

var (
	// обработчики результата отправки письма
	resultHandlers = map[common.SendEventResult]func(*Consumer, *amqp.Channel, *common.MailMessage){
		common.ErrorSendEventResult:     (*Consumer).handleErrorSend,
		common.DelaySendEventResult:     (*Consumer).handleDelaySend,
		common.OverlimitSendEventResult: (*Consumer).handleOverlimitSend,
	}
)

// получатель сообщений из очереди
type Consumer struct {
	id        int
	connector *amqpConnector
	binding   *Binding
}

// создает нового получателя
func NewConsumer(id int, connect *amqpConnector, binding *Binding) *Consumer {
	app := new(Consumer)
	app.id = id
	app.connector = connect
	app.binding = binding
	return app
}

// запускает получение сообщений из очереди в заданное количество потоков
func (c *Consumer) run() {
	for i := 0; i < c.binding.Handlers; i++ {
		go c.consume(i)
	}
}

// подключается к очереди для получения сообщений
func (c *Consumer) consume(id int) {
	channel, err := c.connector.GetConnect().Channel()
	if err != nil {
		logger.All().Warn("consumer#%d, handler#%d can't get channel %s", c.id, id, c.binding.Queue)
		return
	}

	// выбираем из очереди сообщения с запасом
	// это нужно для того, чтобы после отправки письма новое уже было готово к отправке
	// в тоже время нельзя выбираеть все сообщения из очереди разом, т.к. можно упереться в память
	err = channel.Qos(c.binding.PrefetchCount, 0, false)
	if err != nil {
		logger.All().Warn("consumer#%d, handler#%d can't set qos %s", c.id, id, c.binding.Queue)
		return
	}

	deliveries, err := channel.Consume(
		c.binding.Queue, // name
		"",              // consumerTag,
		false,           // noAck
		false,           // exclusive
		false,           // noLocal
		false,           // noWait
		nil,             // arguments
	)
	if err == nil {
		go c.consumeDeliveries(id, channel, deliveries)
	} else {
		logger.All().Warn("consumer#%d, handler#%d can't consume queue %s", c.id, id, c.binding.Queue)
	}
}

// получает сообщения из очереди и отправляет их другим сервисам
func (c *Consumer) consumeDeliveries(id int, channel *amqp.Channel, deliveries <-chan amqp.Delivery) {
	for delivery := range deliveries {
		message := new(common.MailMessage)
		err := json.Unmarshal(delivery.Body, message)
		if err == nil {
			// инициализируем параметры письма
			message.Init()
			logger.
				By(message.HostnameFrom).
				Info(
					"consumer#%d-%d, handler#%d send mail#%d: envelope - %s, recipient - %s to mailer",
					c.id,
					message.Id,
					id,
					message.Id,
					message.Envelope,
					message.Recipient,
				)

			event := common.NewSendEvent(message)
			logger.By(message.HostnameFrom).Debug("consumer#%d-%d send event", c.id, message.Id)
			event.Iterator.Next().(common.SendingService).Event(event)
			// ждем результата,
			// во время ожидания поток блокируется
			// если этого не сделать, тогда невозможно будет подтвердить получение сообщения из очереди
			if handler, ok := resultHandlers[<-event.Result]; ok {
				handler(c, channel, message)
			}
			message = nil
			event = nil
		} else {
			failureBinding := c.binding.failureBindings[TechnicalFailureBindingType]
			err := channel.Publish(
				failureBinding.Exchange,
				failureBinding.Routing,
				false,
				false,
				amqp.Publishing{
					ContentType:  "text/plain",
					Body:         delivery.Body,
					DeliveryMode: amqp.Transient,
				},
			)
			logger.All().WarnWithErr(err, "consumer#%d can't unmarshal delivery body, body should be json, %s given", c.id, string(delivery.Body))
		}
		// всегда подтверждаем получение сообщения
		// даже если во время отправки письма возникли ошибки,
		// мы уже положили это письмо в другую очередь
		_ = delivery.Ack(true)
	}
}

// обрабатывает письма, которые не удалось отправить
func (c *Consumer) handleErrorSend(channel *amqp.Channel, message *common.MailMessage) {
	// если есть ошибка при отправке, значит мы попали в серый список
	// https://ru.wikipedia.org/wiki/%D0%A1%D0%B5%D1%80%D1%8B%D0%B9_%D1%81%D0%BF%D0%B8%D1%81%D0%BE%D0%BA
	// или получили какую то ошибку от почтового сервиса, что он не может
	// отправить письмо указанному адресату или выполнить какую то команду
	var failureBinding *Binding
	// если ошибка связана с невозможностью отправить письмо адресату
	// перекладываем письмо в очередь для плохих писем
	// и пусть отправители сами с ними разбираются
	switch {
	case message.Error.Code >= 500 && message.Error.Code < 600:
		failureBinding = c.binding.failureBindings[errorSignsMap.BindingType(message)]
	case message.Error.Code == 450 || message.Error.Code == 451:
		failureBinding = delayedBindings[common.ThirtyMinutesDelayedBinding]
	default:
		failureBinding = c.binding.failureBindings[UnknownFailureBindingType]
	}
	jsonMessage, err := json.Marshal(message)
	if err == nil {
		// кладем в очередь
		err = channel.Publish(
			failureBinding.Exchange,
			failureBinding.Routing,
			false,
			false,
			amqp.Publishing{
				ContentType:  "text/plain",
				Body:         jsonMessage,
				DeliveryMode: amqp.Transient,
			},
		)
		if err == nil {
			logger.
				By(message.HostnameFrom).
				Debug(
					"consumer#%d-%d publish failure mail to queue %s, message: %s, code: %d",
					c.id,
					message.Id,
					failureBinding.Queue,
					message.Error.Message,
					message.Error.Code,
				)
		} else {
			logger.
				By(message.HostnameFrom).
				Debug(
					"consumer#%d-%d can't publish failure mail to queue %s, message: %s, code: %d, publish error %v",
					c.id,
					message.Id,
					failureBinding.Queue,
					message.Error.Message,
					message.Error.Code,
					err,
				)
			logger.By(message.HostnameFrom).WarnErr(err)
		}
	} else {
		logger.By(message.HostnameFrom).WarnErr(err)
	}
}

// обрабатывает письма, которые нужно отправить позже
func (c *Consumer) handleDelaySend(channel *amqp.Channel, message *common.MailMessage) {
	logger.
		By(message.HostnameFrom).
		Debug(
			"consumer%d-%d find dlx queue",
			c.id,
			message.Id,
		)
	bindingType := common.UnknownDelayedBinding
	if message.Error != nil {
		logger.
			By(message.HostnameFrom).
			Debug(
				"consumer%d-%d detect error, message: %s, code: %d",
				c.id,
				message.Id,
				message.Error.Message,
				message.Error.Code,
			)
	}
	logger.
		By(message.HostnameFrom).
		Debug(
			"consumer%d-%d detect old dlx queue type#%v",
			c.id,
			message.Id,
			message.BindingType,
		)
	// если нам просто не удалось отправить письмо, берем следующую очередь из цепочки
	if chainBinding, ok := bindingsChain[message.BindingType]; ok {
		bindingType = chainBinding
	}
	c.publishDelayedMessage(channel, bindingType, message)
}

// обрабатывает письма, которые превысили лимит отправки
func (c *Consumer) handleOverlimitSend(channel *amqp.Channel, message *common.MailMessage) {
	bindingType := common.UnknownDelayedBinding
	logger.By(message.HostnameFrom).Debug("consumer#%d-%d detect overlimit, find dlx queue", c.id, message.Id)
	for i := 0; i < limitBindingsLen; i++ {
		if limitBindings[i] == message.BindingType {
			bindingType = limitBindings[i]
			break
		}
	}
	c.publishDelayedMessage(channel, bindingType, message)
}

// кладет письмо обратно в одну из отложенных очередей
func (c *Consumer) publishDelayedMessage(channel *amqp.Channel, bindingType common.DelayedBindingType, message *common.MailMessage) {
	// получаем очередь, проверяем, что она реально есть
	// а что? а вдруг нет)
	if delayedBinding, ok := c.binding.delayedBindings[bindingType]; ok {
		message.BindingType = bindingType
		jsonMessage, err := json.Marshal(message)
		if err == nil {
			// кладем в очередь
			err = channel.Publish(
				delayedBinding.Exchange,
				delayedBinding.Routing,
				false,
				false,
				amqp.Publishing{
					ContentType:  "text/plain",
					Body:         jsonMessage,
					DeliveryMode: amqp.Transient,
				},
			)
			if err == nil {
				logger.By(message.HostnameFrom).Debug("consumer#%d-%d publish failure mail to queue %s", c.id, message.Id, delayedBinding.Queue)
			} else {
				logger.All().WarnWithErr(err, "consumer#%d-%d can't publish failure mail to queue %s", c.id, message.Id, delayedBinding.Queue)
			}
		} else {
			logger.All().Warn("consumer#%d-%d can't marshal mail to json", c.id, message.Id)
		}
	} else {
		logger.All().Warn("consumer#%d-%d unknow delayed type#%v", c.id, message.Id, bindingType)
	}
}

// получает письма из всех очередей с ошибками
func (c *Consumer) consumeFailureMessages(group *sync.WaitGroup) {
	channel, err := c.connector.GetConnect().Channel()
	if err == nil {
		for _, failureBinding := range c.binding.failureBindings {
			for {
				delivery, ok, _ := channel.Get(failureBinding.Queue, false)
				if ok {
					message := new(common.MailMessage)
					err = json.Unmarshal(delivery.Body, message)
					if err == nil {
						sendEvent := common.NewSendEvent(message)
						sendEvent.Iterator.Next().(common.ReportService).Event(sendEvent)
					}
				} else {
					break
				}
			}
		}
		group.Done()
	} else {
		logger.All().WarnErr(err)
	}
}

// получает сообщения из одной очереди и кладет их в другую
func (c *Consumer) consumeAndPublishMessages(event *common.ApplicationEvent, group *sync.WaitGroup) {
	channel, err := c.connector.GetConnect().Channel()
	if err == nil {
		var envelopeRegex, recipientRegex *regexp.Regexp
		srcBinding := c.findBindingByQueueName(event.GetStringArg("srcQueue"))
		if srcBinding == nil {
			fmt.Println("source queue should be defined")
			common.App.SendEvents(common.NewApplicationEvent(common.FinishApplicationEventKind))
		}
		destBinding := c.findBindingByQueueName(event.GetStringArg("destQueue"))
		if destBinding == nil {
			fmt.Println("destination queue should be defined")
			common.App.SendEvents(common.NewApplicationEvent(common.FinishApplicationEventKind))
		}
		if srcBinding == destBinding {
			fmt.Println("source and destination queue should be different")
			common.App.SendEvents(common.NewApplicationEvent(common.FinishApplicationEventKind))
		}
		if len(event.GetStringArg("envelope")) > 0 {
			envelopeRegex, _ = regexp.Compile(event.GetStringArg("envelope"))
		}
		if len(event.GetStringArg("recipient")) > 0 {
			recipientRegex, _ = regexp.Compile(event.GetStringArg("recipient"))
		}

		publishDeliveries := make([]amqp.Delivery, 0)
		for {
			delivery, ok, _ := channel.Get(srcBinding.Queue, false)
			if ok {
				message := new(common.MailMessage)
				err = json.Unmarshal(delivery.Body, message)
				if err == nil {
					var necessaryPublish bool
					if (event.GetIntArg("code") > common.InvalidInputInt && event.GetIntArg("code") == message.Error.Code) ||
						(envelopeRegex != nil && envelopeRegex.MatchString(message.Envelope)) ||
						(recipientRegex != nil && recipientRegex.MatchString(message.Recipient)) ||
						(event.GetIntArg("code") == common.InvalidInputInt && envelopeRegex == nil && recipientRegex == nil) {
						necessaryPublish = true
					}
					if necessaryPublish {
						fmt.Printf(
							"find mail#%d: envelope - %s, recipient - %s\n",
							message.Id,
							message.Envelope,
							message.Recipient,
						)
						publishDeliveries = append(publishDeliveries, delivery)
					}
				}
			} else {
				break
			}
		}

		for _, delivery := range publishDeliveries {
			err = channel.Publish(
				destBinding.Exchange,
				destBinding.Routing,
				false,
				false,
				amqp.Publishing{
					ContentType:  "text/plain",
					Body:         delivery.Body,
					DeliveryMode: amqp.Transient,
				},
			)
			if err == nil {
				err := delivery.Ack(true)
				if err != nil {
					logger.All().WarnWithErr(err, "can't acknowledge message")
				}
			} else {
				err := delivery.Nack(true, true)
				if err != nil {
					logger.All().WarnWithErr(err, "can't not acknowledge message")
				}
			}
		}
		group.Done()
	} else {
		logger.All().WarnErr(err)
	}
}

// ищет связку по имени
func (c *Consumer) findBindingByQueueName(queueName string) *Binding {
	var binding *Binding

	if c.binding.Queue == queueName {
		binding = c.binding
	}

	if binding == nil {
		for _, failureBinding := range c.binding.failureBindings {
			if failureBinding.Queue == queueName {
				binding = failureBinding
				break
			}
		}
	}

	if binding == nil {
		for _, delayedBinding := range c.binding.delayedBindings {
			if delayedBinding.Queue == queueName {
				binding = delayedBinding
				break
			}
		}
	}

	return binding
}
<file_sep>package common

import "sync"

// Queue потоко-безопасная очередь
type Queue struct {
	// флаг, сигнализирующий, что очередь пуста
	empty bool

	// элементы очереди
	items []interface{}

	// семафор
	mutex *sync.Mutex
}

// NewQueue создает новую очередь
func NewQueue() *Queue {
	return &Queue{
		empty: true,
		items: make([]interface{}, 0),
		mutex: new(sync.Mutex),
	}
}

// Push добавляет элемент в конец очереди
func (q *Queue) Push(item interface{}) {
	q.mutex.Lock()
	if q.empty {
		q.empty = false
	}
	q.items = append(q.items, item)
	q.mutex.Unlock()
}

// Pop достает первый элемент из очереди
func (q *Queue) Pop() interface{} {
	var item interface{}
	q.mutex.Lock()
	if !q.empty {
		oldItems := q.items
		oldItemsLen := len(oldItems)
		if oldItemsLen > 0 {
			item = oldItems[oldItemsLen-1]
			q.items = oldItems[0 : oldItemsLen-1]
		} else {
			q.empty = true
		}
	}
	q.mutex.Unlock()
	return item
}

// Empty сигнализирует, что очередь пуста
func (q *Queue) Empty() bool {
	var empty bool
	q.mutex.Lock()
	empty = q.empty
	q.mutex.Unlock()
	return empty
}

// Len возвращает длину очереди
func (q *Queue) Len() int {
	var itemsLen int
	q.mutex.Lock()
	itemsLen = len(q.items)
	q.mutex.Unlock()
	return itemsLen
}

// статус очереди
type queueStatus int

const (
	// лимитированная очередь
	limitedQueueStatus queueStatus = iota

	// безлимитная очередь
	unlimitedQueueStatus
)

// LimitedQueue лимитированная очередь, в ней будут храниться клиенты к почтовым сервисам
type LimitedQueue struct {
	*Queue

	// статус, говорящий заблокирована очередь или нет
	status queueStatus

	// максимальное количество элементов, которое было в очереди
	maxLen int
}

// NewLimitQueue создает новую лимитированную очередь
func NewLimitQueue() *LimitedQueue {
	return &LimitedQueue{
		Queue:  NewQueue(),
		status: unlimitedQueueStatus,
	}
}

// HasLimit сигнализирует, что очередь имеет лимит
func (l *LimitedQueue) HasLimit() bool {
	l.mutex.Lock()
	hasLimit := l.status == limitedQueueStatus
	l.mutex.Unlock()
	return hasLimit
}

// HasLimitOn устанавливает лимит очереди
func (l *LimitedQueue) HasLimitOn() {
	if l.MaxLen() > 0 && !l.HasLimit() {
		l.setStatus(limitedQueueStatus)
	}
}

// HasLimitOff снимает лимит очереди
func (l *LimitedQueue) HasLimitOff() {
	l.setStatus(unlimitedQueueStatus)
}

// устанавливает статус очереди
func (l *LimitedQueue) setStatus(status queueStatus) {
	l.mutex.Lock()
	l.status = status
	l.mutex.Unlock()
}

// MaxLen максимальная длина очереди до того момента, как был установлен лимит
func (l *LimitedQueue) MaxLen() int {
	l.mutex.Lock()
	maxLen := l.maxLen
	l.mutex.Unlock()
	return maxLen
}

// AddMaxLen увеличивает максимальную длину очереди
func (l *LimitedQueue) AddMaxLen() {
	l.mutex.Lock()
	l.maxLen++
	l.mutex.Unlock()
}
<file_sep>package mailer

import (
	"bytes"
	"fmt"
	"io"
	"strconv"
	"strings"

	"github.com/emersion/go-msgauth/dkim"

	"github.com/Halfi/postmanq/common"
	"github.com/Halfi/postmanq/logger"
)

// Mailer отправитель письма
type Mailer struct {
	// идентификатор для логов
	id int

	service *Service
}

// создает нового отправителя
func newMailer(id int, service *Service) {
	mailer := &Mailer{id: id, service: service}
	mailer.run()
}

// запускает отправителя
func (m *Mailer) run() {
	for event := range m.service.events {
		m.sendMail(event)
	}
}

// подписывает dkim и отправляет письмо
func (m *Mailer) sendMail(event *common.SendEvent) {
	message := event.Message
	if common.EmailRegexp.MatchString(message.Envelope) && common.EmailRegexp.MatchString(message.Recipient) {
		m.prepare(message)
		m.send(event)
	} else {
		ReturnMail(event, fmt.Errorf("511 service#%d can't send mail#%d, envelope or ricipient is invalid", m.id, message.Id))
	}
}

// подписывает dkim
func (m *Mailer) prepare(message *common.MailMessage) {
	options := &dkim.SignOptions{
		Domain:                 message.HostnameFrom,
		Identifier:             message.Envelope,
		Selector:               m.service.getDkimSelector(message.HostnameFrom),
		Signer:                 m.service.getPrivateKey(message.HostnameFrom),
		HeaderCanonicalization: dkim.CanonicalizationRelaxed,
		BodyCanonicalization:   dkim.CanonicalizationRelaxed,
		HeaderKeys: []string{
			"From",
			"To",
			"Subject",
		},
	}

	var b bytes.Buffer
	if err := dkim.Sign(&b, bytes.NewReader(message.Body), options); err != nil {
		logger.By(message.HostnameFrom).WarnWithErr(err, "mailer#%d-%d can't sign mail", m.id, message.Id)
		return
	}

	message.Body = b.Bytes()
	logger.By(message.HostnameFrom).Debug("mailer#%d-%d success sign mail", m.id, message.Id)
}

// отправляет письмо
func (m *Mailer) send(event *common.SendEvent) {
	message := event.Message
	worker := event.Client.Worker

	logger.By(event.Message.HostnameFrom).Info("mailer#%d-%d begin sending mail", m.id, message.Id)
	logger.By(message.HostnameFrom).Debug("mailer#%d-%d receive smtp client#%d", m.id, message.Id, event.Client.Id)

	success := false
	toErr := event.Client.SetTimeout(common.App.Timeout().Mail)
	if toErr != nil {
		logger.By(message.HostnameFrom).ErrErr(toErr)
	}

	err := worker.Mail(message.Envelope)
	if err == nil {
		logger.By(message.HostnameFrom).Debug("mailer#%d-%d send command MAIL FROM: %s", m.id, message.Id, message.Envelope)

		toErr := event.Client.SetTimeout(common.App.Timeout().Rcpt)
		if toErr != nil {
			logger.By(message.HostnameFrom).ErrErr(toErr)
		}

		err = worker.Rcpt(message.Recipient)
		if err == nil {
			logger.By(message.HostnameFrom).Debug("mailer#%d-%d send command RCPT TO: %s", m.id, message.Id, message.Recipient)

			toErr := event.Client.SetTimeout(common.App.Timeout().Data)
			if toErr != nil {
				logger.By(message.HostnameFrom).ErrErr(toErr)
			}

			var wc io.WriteCloser
			wc, err = worker.Data()
			if err == nil {
				logger.By(message.HostnameFrom).Debug("mailer#%d-%d send command DATA", m.id, message.Id)

				_, err = wc.Write(message.Body)
				if err == nil {
					_ = wc.Close()
					logger.By(message.HostnameFrom).Debug("%s", message.Body)
					logger.By(message.HostnameFrom).Debug("mailer#%d-%d send command .", m.id, message.Id)

					// стараемся слать письма через уже созданное соединение,
					// поэтому после отправки письма не закрываем соединение
					err = worker.Reset()
					if err == nil {
						logger.By(message.HostnameFrom).Debug("mailer#%d-%d send command RSET", m.id, message.Id)
						logger.By(event.Message.HostnameFrom).Info("mailer#%d-%d success send mail#%d", m.id, message.Id, message.Id)
						success = true
					}
				}
			}
		}
	}

	event.Client.Wait()
	event.Queue.Push(event.Client)

	if success {
		// отпускаем поток получателя сообщений из очереди
		event.Result <- common.SuccessSendEventResult
	} else {
		ReturnMail(event, err)
	}
}

// возвращает письмо обратно в очередь после ошибки во время отправки
func ReturnMail(event *common.SendEvent, err error) {
	// необходимо проверить сообщение на наличие кода ошибки
	// обычно код идет первым
	if err != nil {
		errorMessage := err.Error()
		parts := strings.Split(errorMessage, " ")
		if len(parts) > 0 {
			// пытаемся получить код
			code, e := strconv.Atoi(strings.TrimSpace(parts[0]))
			// и создать ошибку
			// письмо с ошибкой вернется в другую очередь, отличную от письмо без ошибки
			if e == nil {
				event.Message.Error = &common.MailError{Message: errorMessage, Code: code}
			}
		} else {
			logger.All().Err("can't get err code from error: %s", err)
		}
	}

	// если в событии уже создан клиент
	if event.Client != nil {
		if event.Client.Worker != nil {
			// сбрасываем цепочку команд к почтовому сервису
			err := event.Client.Worker.Reset()
			if err != nil {
				logger.All().WarnErr(err)
			}
		}
	}

	// отпускаем поток получателя сообщений из очереди
	if event.Message.Error == nil {
		event.Result <- common.DelaySendEventResult
		logger.All().Warn("message delayed")
	} else {
		if event.Message.Error.Code == 421 {
			logger.All().Warn("message delayed with error %s", event.Message.Error.Message)
			event.Result <- common.DelaySendEventResult
		} else {
			event.Result <- common.ErrorSendEventResult
		}
	}
}
<file_sep>package consumer

import (
	"strings"

	"github.com/Halfi/postmanq/common"
)

var (
	// карта признаков ошибок, используется для распределения неотправленных сообщений по очередям для ошибок
	errorSignsMap = ErrorSignsMap{
		501: ErrorSigns{
			ErrorSign{RecipientFailureBindingType, []string{
				"bad address syntax",
			}},
		},
		502: ErrorSigns{
			ErrorSign{TechnicalFailureBindingType, []string{
				"syntax error",
			}},
		},
		503: ErrorSigns{
			ErrorSign{TechnicalFailureBindingType, []string{
				"sender not yet given",
				"sender already",
				"bad sequence",
				"commands were rejected",
				"rcpt first",
				"rcpt command",
				"mail first",
				"mail command",
				"mail before",
			}},
			ErrorSign{RecipientFailureBindingType, []string{
				"account blocked",
				"user unknown",
			}},
		},
		504: ErrorSigns{
			ErrorSign{RecipientFailureBindingType, []string{
				"mailbox is disabled",
			}},
		},
		511: ErrorSigns{
			ErrorSign{RecipientFailureBindingType, []string{
				"can't lookup",
			}},
		},
		540: ErrorSigns{
			ErrorSign{RecipientFailureBindingType, []string{
				"recipient address rejected",
				"account has been suspended",
				"account deleted",
			}},
		},
		550: ErrorSigns{
			ErrorSign{TechnicalFailureBindingType, []string{
				"sender verify failed",
				"callout verification failed:",
				"relay",
				"verification failed",
				"unnecessary spaces",
				"host lookup failed",
				"client host rejected",
				"backresolv",
				"can't resolve hostname",
				"reverse",
				"authentication required",
				"bad commands",
				"double-checking",
				"system has detected that",
				"more information",
				"message has been blocked",
				"unsolicited mail",
				"blacklist",
				"black list",
				"not allowed to send",
				"dns operator",
			}},
			ErrorSign{RecipientFailureBindingType, []string{
				"unknown",
				"no such",
				"not exist",
				"disabled",
				"invalid mailbox",
				"not found",
				"mailbox unavailable",
				"has been suspended",
				"inactive",
				"account unavailable",
				"addresses failed",
				"mailbox is frozen",
				"address rejected",
				"administrative prohibition",
				"cannot deliver",
				"unrouteable address",
				"user banned",
				"policy rejection",
				"verify recipient",
				"mailbox locked",
				"blocked",
				"no mailbox",
				"bad destination mailbox",
				"not stored this user",
				"homo hominus",
			}},
			ErrorSign{ConnectionFailureBindingType, []string{
				"spam",
				"is full",
				"over quota",
				"quota exceeded",
				"message rejected",
				"was not accepted",
				"content denied",
				"timeout",
				"support.google.com",
			}},
		},
		552: ErrorSigns{
			ErrorSign{ConnectionFailureBindingType, []string{
				"receiving disabled",
				"is full",
				"over quot",
				"to big",
			}},
		},
		553: ErrorSigns{
			ErrorSign{RecipientFailureBindingType, []string{
				"list of allowed",
				"ecipient has been denied",
			}},
			ErrorSign{TechnicalFailureBindingType, []string{
				"relay",
			}},
			ErrorSign{ConnectionFailureBindingType, []string{
				"does not accept mail from",
			}},
		},
		554: ErrorSigns{
			ErrorSign{TechnicalFailureBindingType, []string{
				"relay access denied",
				"unresolvable address",
				"blocked using",
			}},
			ErrorSign{RecipientFailureBindingType, []string{
				"recipient address rejected",
				"user doesn't have",
				"no such user",
				"inactive user",
				"user unknown",
				"has been disabled",
				"should log in",
				"no mailbox here",
			}},
			ErrorSign{ConnectionFailureBindingType, []string{
				"spam message rejected",
				"suspicion of spam",
				"synchronization error",
				"refused",
			}},
		},
		571: ErrorSigns{
			ErrorSign{ConnectionFailureBindingType, []string{
				"relay",
			}},
		},
		578: ErrorSigns{
			ErrorSign{ConnectionFailureBindingType, []string{
				"address rejected with reverse-check",
			}},
		},
	}
)

// карта признаков ошибок, в качестве ключа используется код ошибки, полученной от почтового сервиса
type ErrorSignsMap map[int]ErrorSigns

// отдает идентификатор очереди, в которую необходимо положить письмо с ошибкой
func (e ErrorSignsMap) BindingType(message *common.MailMessage) FailureBindingType {
	if signs, ok := e[message.Error.Code]; ok {
		return signs.BindingType(message)
	} else {
		return UnknownFailureBindingType
	}
}

// признаки ошибок
type ErrorSigns []ErrorSign

// отдает идентификатор очереди, в которую необходимо положить письмо с ошибкой
func (e ErrorSigns) BindingType(message *common.MailMessage) FailureBindingType {
	bindingType := UnknownFailureBindingType
	for _, sign := range e {
		if sign.resemble(message) {
			bindingType = sign.bindingType
			break
		}
	}
	return bindingType
}

// признак ошибки
type ErrorSign struct {
	// идентификатор очереди
	bindingType FailureBindingType

	// возможные части сообщения, по которым ошибка соотносится с очередью для ошибок
	parts []string
}

// ищет возможные части сообщения в сообщении ошибки
func (e ErrorSign) resemble(message *common.MailMessage) bool {
	hasPart := false
	for _, part := range e.parts {
		if strings.Contains(message.Error.Message, part) {
			hasPart = true
			break
		}
	}
	return hasPart
}
<file_sep>module github.com/Halfi/postmanq

go 1.16

require (
	github.com/alexliesenfeld/health v0.6.0
	github.com/byorty/clitable v0.0.0-20150722055417-9f60651b8308
	github.com/emersion/go-msgauth v0.6.5
	github.com/kr/pretty v0.3.0 // indirect
	github.com/prometheus/client_golang v1.11.0
	github.com/rogpeppe/go-internal v1.8.0 // indirect
	github.com/rs/zerolog v1.23.0
	github.com/streadway/amqp v1.0.0
	golang.org/x/crypto v0.0.0-20210813211128-0a44fdfbc16e // indirect
	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
)
<file_sep>package connector

import (
	"errors"
	"fmt"
	"time"

	"github.com/Halfi/postmanq/common"
	"github.com/Halfi/postmanq/logger"
	"github.com/Halfi/postmanq/mailer"
)

// заготовщик, подготавливает событие соединения
type Preparer struct {
	// Идентификатор для логов
	id int

	events          chan *common.SendEvent
	connectorEvents chan *ConnectionEvent
	seekerEvents    chan *ConnectionEvent
}

// создает и запускает нового заготовщика
func newPreparer(id int, events chan *common.SendEvent, connectorEvents, seekerEvents chan *ConnectionEvent) *Preparer {
	return &Preparer{id: id, events: events, connectorEvents: connectorEvents, seekerEvents: seekerEvents}
}

// запускает прослушивание событий отправки писем
func (p *Preparer) run() {
	for event := range p.events {
		p.prepare(event)
	}
}

// подготавливает и запускает событие создание соединения
func (p *Preparer) prepare(event *common.SendEvent) {
	logger.By(event.Message.HostnameFrom).Info("preparer#%d-%d try create connection", p.id, event.Message.Id)

	connectionEvent := &ConnectionEvent{
		SendEvent:   event,
		servers:     make(chan *MailServer, 1),
		connectorId: p.id,
		address:     service.getAddress(event.Message.HostnameFrom, p.id),
	}
	goto connectToMailServer

connectToMailServer:
	// отправляем событие сбора информации о сервере
	p.seekerEvents <- connectionEvent
	server := <-connectionEvent.servers
	switch server.status {
	case LookupMailServerStatus:
		goto waitLookup
	case SuccessMailServerStatus:
		connectionEvent.server = server
		p.connectorEvents <- connectionEvent
	case ErrorMailServerStatus:
		mailer.ReturnMail(
			event,
			errors.New(fmt.Sprintf("511 preparer#%d-%d can't lookup %s", p.id, event.Message.Id, event.Message.HostnameTo)),
		)
	}

waitLookup:
	logger.By(event.Message.HostnameFrom).Debug("preparer#%d-%d wait ending look up mail server %s...", p.id, event.Message.Id, event.Message.HostnameTo)
	time.Sleep(common.App.Timeout().Sleep)
	goto connectToMailServer
}
<file_sep>package consumer

import (
	"fmt"
	"net/url"
	"sync"

	"github.com/streadway/amqp"
	"gopkg.in/yaml.v3"

	"github.com/Halfi/postmanq/common"
	"github.com/Halfi/postmanq/logger"
)

type amqpConnector struct {
	connect *amqp.Connection
	rwm     sync.RWMutex
}

func (ac *amqpConnector) SetConnect(connect *amqp.Connection) {
	ac.rwm.Lock()
	defer ac.rwm.Unlock()
	ac.connect = connect
}

func (ac *amqpConnector) GetConnect() *amqp.Connection {
	ac.rwm.RLock()
	defer ac.rwm.RUnlock()
	return ac.connect
}

// сервис получения сообщений
type Service struct {
	// настройка получателей сообщений
	Configs []*Config `yaml:"consumers"`

	// подключения к очередям
	connections map[string]*amqpConnector

	// получатели сообщений из очереди
	consumers map[string][]*Consumer

	assistants map[string][]*Assistant

	finish bool
}

// создает новый сервис получения сообщений
func Inst() common.SendingService {
	return &Service{
		connections: make(map[string]*amqpConnector),
		consumers:   make(map[string][]*Consumer),
		assistants:  make(map[string][]*Assistant),
	}
}

// OnInit инициализирует сервис
func (s *Service) OnInit(event *common.ApplicationEvent) {
	logger.All().Debug("init consumer service")
	// получаем настройки
	err := yaml.Unmarshal(event.Data, s)
	if err != nil {
		logger.All().ErrWithErr(err, "consumer service can't unmarshal config")
	}

	if len(s.Configs) == 0 {
		logger.All().FailExit("consumer config is empty")
		return
	}

	consumersCount := 0
	assistantsCount := 0
	for _, config := range s.Configs {
		connect, err := amqp.Dial(config.URI)
		if err != nil {
			logger.All().ErrWithErr(err, "consumer service can't connect to %s", config.URI)
			continue
		}

		connector := new(amqpConnector)
		connector.SetConnect(connect)

		channel, err := connector.GetConnect().Channel()
		if err != nil {
			logger.All().ErrWithErr(err, "consumer service can't get channel to %s", config.URI)
			continue
		}

		consumers := make([]*Consumer, len(config.Bindings))
		for i, binding := range config.Bindings {
			binding.init()
			// объявляем очередь
			binding.declare(channel)

			binding.delayedBindings = make(map[common.DelayedBindingType]*Binding)
			// объявляем отложенные очереди
			for delayedBindingType, delayedBinding := range delayedBindings {
				delayedBinding.declareDelayed(binding, channel)
				binding.delayedBindings[delayedBindingType] = delayedBinding
			}

			binding.failureBindings = make(map[FailureBindingType]*Binding)
			for failureBindingType, tplName := range failureBindingTypeTplNames {
				failureBinding := new(Binding)
				failureBinding.Exchange = fmt.Sprintf(tplName, binding.Exchange)
				failureBinding.Queue = fmt.Sprintf(tplName, binding.Queue)
				failureBinding.Type = binding.Type
				failureBinding.declare(channel)
				binding.failureBindings[failureBindingType] = failureBinding
			}

			consumersCount++
			consumers[i] = NewConsumer(consumersCount, connector, binding)
		}

		assistants := make([]*Assistant, len(config.Assistants))
		for i, assistantBinding := range config.Assistants {
			assistantBinding.Binding.init()
			// объявляем очередь
			assistantBinding.Binding.declare(channel)

			destBindings := make(map[string]*Binding)
			for domain, exchange := range assistantBinding.Dest {
				for _, consumer := range consumers {
					if consumer.binding.Exchange == exchange {
						destBindings[domain] = consumer.binding
						break
					}
				}
			}

			assistantsCount++
			assistants[i] = &Assistant{
				id:           assistantsCount,
				connector:    connector,
				srcBinding:   assistantBinding,
				destBindings: destBindings,
			}
		}

		s.connections[config.URI] = connector
		s.consumers[config.URI] = consumers
		s.assistants[config.URI] = assistants
		// слушаем закрытие соединения
		s.reconnect(connector, config)
	}
}

// объявляет слушателя закрытия соединения
func (s *Service) reconnect(connector *amqpConnector, config *Config) {
	closeErrors := connector.GetConnect().NotifyClose(make(chan *amqp.Error))
	go s.notifyCloseError(config, closeErrors)
}

// слушает закрытие соединения
func (s *Service) notifyCloseError(config *Config, closeErrors chan *amqp.Error) {
	for closeError := range closeErrors {
		if s.finish {
			return
		}

		logger.All().Warn("consumer service close connection %s with error - %v, restart...", config.URI, closeError)
		connect, err := amqp.Dial(config.URI)
		if err == nil {
			connector, ok := s.connections[config.URI]
			if !ok {
				s.connections[config.URI] = new(amqpConnector)
			}
			connector.SetConnect(connect)

			s.reconnect(connector, config)
			logger.All().Debug("consumer service reconnect to amqp server %s", config.URI)
		} else {
			logger.All().WarnWithErr(err, "consumer service can't reconnect to amqp server %s", config.URI)
		}
	}
}

// запускает сервис
func (s *Service) OnRun() {
	logger.All().Debug("run consumers...")
	for _, consumers := range s.consumers {
		s.runConsumers(consumers)
	}
	for _, assistants := range s.assistants {
		s.runAssistants(assistants)
	}
}

// запускает получателей
func (s *Service) runConsumers(consumers []*Consumer) {
	for _, consumer := range consumers {
		go consumer.run()
	}
}

func (s *Service) runAssistants(assistants []*Assistant) {
	for _, assistant := range assistants {
		go assistant.run()
	}
}

// останавливает получателей
func (s *Service) OnFinish() {
	logger.All().Debug("stop consumers...")
	s.finish = true
	for _, connector := range s.connections {
		if connector != nil && connector.GetConnect() != nil {
			err := connector.GetConnect().Close()
			if err != nil {
				logger.All().WarnErr(err)
			}
		}
	}

	s.connections = make(map[string]*amqpConnector)
	s.consumers = make(map[string][]*Consumer)
	s.assistants = make(map[string][]*Assistant)
}

// Event send event
func (s *Service) Event(_ *common.SendEvent) bool {
	return true
}

// запускает получение сообщений с ошибками и пересылает их другому сервису
func (s *Service) OnShowReport() {
	waiter := newWaiter()
	group := new(sync.WaitGroup)

	var delta int
	for _, apps := range s.consumers {
		for _, app := range apps {
			delta += app.binding.Handlers
		}
	}
	group.Add(delta)
	for _, apps := range s.consumers {
		go func(apps []*Consumer) {
			for _, app := range apps {
				for i := 0; i < app.binding.Handlers; i++ {
					go app.consumeFailureMessages(group)
				}
			}
		}(apps)
	}
	group.Wait()
	waiter.Stop()

	sendEvent := common.NewSendEvent(nil)
	sendEvent.Iterator.Next().(common.ReportService).Event(sendEvent)
}

// перекладывает сообщения из очереди в очередь
func (s *Service) OnPublish(event *common.ApplicationEvent) {
	group := new(sync.WaitGroup)
	delta := 0
	for uri, apps := range s.consumers {
		var necessaryPublish bool
		if len(event.GetStringArg("host")) > 0 {
			parsedUri, err := url.Parse(uri)
			if err == nil && parsedUri.Host == event.GetStringArg("host") {
				necessaryPublish = true
			} else {
				necessaryPublish = false
			}
		} else {
			necessaryPublish = true
		}
		if necessaryPublish {
			for _, app := range apps {
				delta += app.binding.Handlers
				for i := 0; i < app.binding.Handlers; i++ {
					go app.consumeAndPublishMessages(event, group)
				}
			}
		}
	}
	group.Add(delta)
	group.Wait()
	fmt.Println("done")
	common.App.SendEvents(common.NewApplicationEvent(common.FinishApplicationEventKind))
}

// получатель сообщений из очереди
type Config struct {
	URI        string              `yaml:"uri"`
	Assistants []*AssistantBinding `yaml:"assistants"`
	Bindings   []*Binding          `yaml:"bindings"`
}
