consume

Documentation

Index

• type ConsumerWithTimeout
  • func NewConsumerWithTimeout(base mrworker.MessageConsumer, readTimeout, writeTimeout time.Duration) *ConsumerWithTimeout
  • func (t *ConsumerWithTimeout) CancelMessages(ctx context.Context, messages []any) error
  • func (t *ConsumerWithTimeout) CommitMessage(ctx context.Context, message any, preCommit func(ctx context.Context) error) error
  • func (t *ConsumerWithTimeout) ReadMessages(ctx context.Context, limit int) (messages []any, err error)
  • func (t *ConsumerWithTimeout) RejectMessage(ctx context.Context, message any, causeErr error) error
• 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 WithCaption(value string) Option
  • func WithCaptionPrefix(value string) Option
  • func WithConsumerTimeout(read, write time.Duration) Option
  • func WithHandlerTimeout(value time.Duration) Option
  • func WithQueueSize(value int) Option
  • func WithReadPeriod(value time.Duration) Option
  • func WithReadyTimeout(value time.Duration) Option
  • func WithSignalExecuteHandler(ch <-chan struct{}) Option
  • func WithWorkersCount(value int) Option

Types

type ConsumerWithTimeout

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

ConsumerWithTimeout - получатель сообщений с возможностью подтверждения их получения.

func NewConsumerWithTimeout

func NewConsumerWithTimeout(base mrworker.MessageConsumer, readTimeout, writeTimeout time.Duration) *ConsumerWithTimeout

NewConsumerWithTimeout - создаёт объект ConsumerWithTimeout.

func (*ConsumerWithTimeout) CancelMessages

func (t *ConsumerWithTimeout) CancelMessages(ctx context.Context, messages []any) error

CancelMessages - comment method.

func (*ConsumerWithTimeout) CommitMessage

func (t *ConsumerWithTimeout) CommitMessage(ctx context.Context, message any, preCommit func(ctx context.Context) error) error

CommitMessage - comment method.

func (*ConsumerWithTimeout) ReadMessages

func (t *ConsumerWithTimeout) ReadMessages(ctx context.Context, limit int) (messages []any, err error)

ReadMessages - comment method.

func (*ConsumerWithTimeout) RejectMessage

func (t *ConsumerWithTimeout) RejectMessage(ctx context.Context, message any, causeErr error) error

RejectMessage - comment method.

type MessageProcessor

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

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

func NewMessageProcessor

func NewMessageProcessor(
	consumer mrworker.MessageConsumer,
	handler mrworker.MessageHandler,
	errorHandler mrcore.ErrorHandler,
	logger mrlog.Logger,
	traceManager traceManager,
	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 - запуск сервиса обработки сообщений. Повторный запуск метода одно и того же объекта не предусмотрен, даже после вызова Shutdown.

type Option

type Option func(o *options)

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

func WithCaption

func WithCaption(value string) Option

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

func WithCaptionPrefix

func WithCaptionPrefix(value string) Option

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

func WithConsumerTimeout

func WithConsumerTimeout(read, write time.Duration) Option

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

func WithHandlerTimeout

func WithHandlerTimeout(value time.Duration) Option

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

func WithQueueSize

func WithQueueSize(value int) Option

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

func WithReadPeriod

func WithReadPeriod(value time.Duration) Option

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

func WithReadyTimeout

func WithReadyTimeout(value time.Duration) Option

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

func WithSignalExecuteHandler

func WithSignalExecuteHandler(ch <-chan struct{}) Option

WithSignalExecuteHandler - comment func.

func WithWorkersCount

func WithWorkersCount(value int) Option

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