consumer

Documentation

Index

• Constants
• func Inst() common.SendingService
• type Assistant
• type AssistantBinding
• type Binding
• type Config
• type Consumer
  • func NewConsumer(id int, connect *amqp.Connection, binding *Binding) *Consumer
• type ErrorSign
• type ErrorSigns
  • func (e ErrorSigns) BindingType(message *common.MailMessage) FailureBindingType
• type ErrorSignsMap
  • func (e ErrorSignsMap) BindingType(message *common.MailMessage) FailureBindingType
• type ExchangeType
• type FailureBindingType
• type Service
  • func (s *Service) Events() chan *common.SendEvent
  • func (s *Service) OnFinish()
  • func (s *Service) OnInit(event *common.ApplicationEvent)
  • func (s *Service) OnPublish(event *common.ApplicationEvent)
  • func (s *Service) OnRun()
  • func (s *Service) OnShowReport()
• type Waiter

Constants

const (
	DirectExchangeType ExchangeType = "direct"
	FanoutExchangeType              = "fanout"
	TopicExchangeType               = "topic"
)

Functions

func Inst

func Inst() common.SendingService

создает новый сервис получения сообщений

Types

type Assistant

type Assistant struct {
	// contains filtered or unexported fields
}

type AssistantBinding

type AssistantBinding struct {
	Binding

	Dest map[string]string `yaml:"dest"`
	// contains filtered or unexported fields
}

type Binding

type Binding struct {
	// имя точки обмена и очереди
	Name string `yaml:"name"`

	// имя точки обмена
	Exchange string `yaml:"exchange"`

	// аргументы точки обмена
	ExchangeArgs amqp.Table

	// имя очереди
	Queue string `yaml:"queue"`

	// аргументы очереди
	QueueArgs amqp.Table

	// тип точки обмена
	Type ExchangeType `yaml:"type"`

	// ключ маршрутизации
	Routing string `yaml:"routing"`

	// количество потоков, разбирающих очередь
	Handlers int `yaml:"workers"`

	// количество сообщений, получаемых одновременно
	PrefetchCount int `yaml:"prefetchCount"`
	// contains filtered or unexported fields
}

связка точки обмена и очереди

type Config

type Config struct {
	URI        string              `yaml:"uri"`
	Assistants []*AssistantBinding `yaml:"assistants"`
	Bindings   []*Binding          `yaml:"bindings"`
}

получатель сообщений из очереди

type Consumer

type Consumer struct {
	// contains filtered or unexported fields
}

получатель сообщений из очереди

func NewConsumer

func NewConsumer(id int, connect *amqp.Connection, binding *Binding) *Consumer

создает нового получателя

type ErrorSign

type ErrorSign struct {
	// contains filtered or unexported fields
}

признак ошибки

type ErrorSigns

type ErrorSigns []ErrorSign

признаки ошибок

func (ErrorSigns) BindingType

func (e ErrorSigns) BindingType(message *common.MailMessage) FailureBindingType

отдает идентификатор очереди, в которую необходимо положить письмо с ошибкой

type ErrorSignsMap

type ErrorSignsMap map[int]ErrorSigns

карта признаков ошибок, в качестве ключа используется код ошибки, полученной от почтового сервиса

func (ErrorSignsMap) BindingType

func (e ErrorSignsMap) BindingType(message *common.MailMessage) FailureBindingType

отдает идентификатор очереди, в которую необходимо положить письмо с ошибкой

type ExchangeType

type ExchangeType string

тип точки обмена

type FailureBindingType

type FailureBindingType int

тип точки обмена для неотправленного письма

const (
	// проблемы с адресатом
	RecipientFailureBindingType FailureBindingType = iota

	// технические проблемы: неверная последовательность команд, косяки с dns
	TechnicalFailureBindingType

	// проблемы с подключеним к почтовому сервису
	ConnectionFailureBindingType

	// неизвестная проблема
	UnknownFailureBindingType
)

type Service

type Service struct {
	// настройка получателей сообщений
	Configs []*Config `yaml:"consumers"`
	// contains filtered or unexported fields
}

сервис получения сообщений

func (*Service) Events

func (s *Service) Events() chan *common.SendEvent

канал для приема событий отправки писем

func (*Service) OnFinish

func (s *Service) OnFinish()

останавливает получателей

func (*Service) OnInit

func (s *Service) OnInit(event *common.ApplicationEvent)

инициализирует сервис

func (*Service) OnPublish

func (s *Service) OnPublish(event *common.ApplicationEvent)

перекладывает сообщения из очереди в очередь

func (*Service) OnRun

func (s *Service) OnRun()

запускает сервис

func (*Service) OnShowReport

func (s *Service) OnShowReport()

запускает получение сообщений с ошибками и пересылает их другому сервису

type Waiter

type Waiter struct {
	*time.Ticker
}

ожидающий