consume

Documentation

Index

• type MessageConsumer
  • func New(txManager mrstorage.DBTxManager, storage mrmailer.MessageStorage, ...) *MessageConsumer
  • func (co *MessageConsumer) CancelMessages(ctx context.Context, messages []any) error
  • func (co *MessageConsumer) Close() error
  • func (co *MessageConsumer) CommitMessage(ctx context.Context, message any, preCommit func(ctx context.Context) error) error
  • func (co *MessageConsumer) ReadMessages(ctx context.Context, limit uint32) ([]any, error)
  • func (co *MessageConsumer) RejectMessage(ctx context.Context, message any, causeErr error) error

Types

type MessageConsumer

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

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

func New

func New(
	txManager mrstorage.DBTxManager,
	storage mrmailer.MessageStorage,
	useCaseQueue mrqueue.Consumer,
	errorWrapper mrcore.UseCaseErrorWrapper,
) *MessageConsumer

New - создаёт объект MessageConsumer.

func (*MessageConsumer) CancelMessages

func (co *MessageConsumer) CancelMessages(ctx context.Context, messages []any) error

CancelMessages - отменяет обработку сообщений, которые были ранее считаны методом ReadMessages.

func (*MessageConsumer) Close

func (co *MessageConsumer) Close() error

Close - закрывает соединение консьюмера с источником данных.

func (*MessageConsumer) CommitMessage

func (co *MessageConsumer) CommitMessage(ctx context.Context, message any, preCommit func(ctx context.Context) error) error

CommitMessage - закрепляет результат обработки сообщения, которое было ранее считано методом ReadMessages. Внешняя функция preCommit работает вместе с фиксацией результата в рамках одной транзакции. При работе в рамках одной БД это позволяет коммитить изменения атомарно.

func (*MessageConsumer) ReadMessages

func (co *MessageConsumer) ReadMessages(ctx context.Context, limit uint32) ([]any, error)

ReadMessages - возвращает указанную порцию сообщений для их обработки.

func (*MessageConsumer) RejectMessage

func (co *MessageConsumer) RejectMessage(ctx context.Context, message any, causeErr error) error

RejectMessage - отклоняет результат обработки сообщения, если в процессе возникла ошибка.