consume

Documentation

Index

• type MessageProcessor
  • func NewMessageProcessor(consumer mrworker.MessageConsumer, handler mrworker.MessageHandler, ...) *MessageProcessor
  • func (p *MessageProcessor) Caption() string
  • func (p *MessageProcessor) ReadyTimeout() time.Duration
  • func (p *MessageProcessor) Shutdown(ctx context.Context) error
  • func (p *MessageProcessor) Start(ctx context.Context, ready func()) error
• type Option
  • func WithBusyReadPeriod(value time.Duration) Option
  • func WithCancelReadTimeout(value time.Duration) Option
  • func WithCaption(value string) Option
  • func WithHandlerTimeout(value time.Duration) Option
  • func WithQueueSize(value uint32) Option
  • func WithReadPeriod(value time.Duration) Option
  • func WithReadyTimeout(value time.Duration) Option
  • func WithWorkersCount(value uint16) Option

Types

type MessageProcessor

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

MessageProcessor - многопоточный сервис обработки сообщений на основе консьюмера и обработчика.

func NewMessageProcessor

func NewMessageProcessor(
	consumer mrworker.MessageConsumer,
	handler mrworker.MessageHandler,
	errorHandler mrcore.ErrorHandler,
	opts ...Option,
) *MessageProcessor

NewMessageProcessor - создаёт объект MessageProcessor.

func (*MessageProcessor) Caption

func (p *MessageProcessor) Caption() string

Caption - возвращает название сервиса обработки сообщений.

func (*MessageProcessor) ReadyTimeout

func (p *MessageProcessor) ReadyTimeout() time.Duration

ReadyTimeout - возвращает максимальное время, за которое должен быть запущен сервис.

func (*MessageProcessor) Shutdown

func (p *MessageProcessor) Shutdown(ctx context.Context) error

Shutdown - корректная остановка сервиса обработки сообщений.

func (*MessageProcessor) Start

func (p *MessageProcessor) Start(ctx context.Context, ready func()) error

Start - запуск сервиса обработки сообщений.

type Option

type Option func(p *MessageProcessor)

Option - настройка объекта MessageProcessor.

func WithBusyReadPeriod

func WithBusyReadPeriod(value time.Duration) Option

WithBusyReadPeriod - устанавливает опцию периода чтения данных консьюмером, когда он в нагруженном состоянии.

func WithCancelReadTimeout

func WithCancelReadTimeout(value time.Duration) Option

WithCancelReadTimeout - устанавливает опцию таймаута на время отмены чтения данных консьюмером при неожиданном завершении работы воркеров.

func WithCaption

func WithCaption(value string) Option

WithCaption - устанавливает опцию caption для MessageProcessor.

func WithHandlerTimeout

func WithHandlerTimeout(value time.Duration) Option

WithHandlerTimeout - устанавливает опцию handlerTimeout выполнения обработчика сообщения.

func WithQueueSize

func WithQueueSize(value uint32) Option

WithQueueSize - устанавливает опцию размера очереди обработки сообщений.

func WithReadPeriod

func WithReadPeriod(value time.Duration) Option

WithReadPeriod - устанавливает опцию периода чтения данных консьюмером, когда он в состоянии простоя.

func WithReadyTimeout

func WithReadyTimeout(value time.Duration) Option

WithReadyTimeout - устанавливает опцию readyTimeout для MessageProcessor.

func WithWorkersCount

func WithWorkersCount(value uint16) Option

WithWorkersCount - устанавливает опцию количества воркеров обрабатывающих сообщения.