consume

Documentation

Index

• type Option
  • func WithStorageBroken(value mrqueue.BrokenStorage) Option
  • func WithStorageCompleted(value mrqueue.CompletedStorage) Option
• type QueueConsumer
  • func New(txManager mrstorage.DBTxManager, storage mrqueue.Storage, ...) *QueueConsumer
  • func (co *QueueConsumer) CancelItems(ctx context.Context, itemsIDs []uint64) error
  • func (co *QueueConsumer) Commit(ctx context.Context, itemID uint64) error
  • func (co *QueueConsumer) ReadItems(ctx context.Context, limit uint32) (itemsIDs []uint64, err error)
  • func (co *QueueConsumer) Reject(ctx context.Context, itemID uint64, causeErr error) error

Types

type Option

type Option func(co *QueueConsumer)

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

func WithStorageBroken

func WithStorageBroken(value mrqueue.BrokenStorage) Option

WithStorageBroken - устанавливает опцию storageBroken для QueueConsumer.

func WithStorageCompleted

func WithStorageCompleted(value mrqueue.CompletedStorage) Option

WithStorageCompleted - устанавливает опцию storageCompleted для QueueConsumer.

type QueueConsumer

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

QueueConsumer - объект для чтения элементов из очереди и информирования о статусе их обработки.

func New

func New(
	txManager mrstorage.DBTxManager,
	storage mrqueue.Storage,
	eventEmitter mrsender.EventEmitter,
	errorWrapper mrcore.UseCaseErrorWrapper,
	opts ...Option,
) *QueueConsumer

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

func (*QueueConsumer) CancelItems

func (co *QueueConsumer) CancelItems(ctx context.Context, itemsIDs []uint64) error

CancelItems - возвращает указанные элементы в статус READY, но только если они находятся в статусе PROCESSING (при этом попытки не отнимаются).

func (*QueueConsumer) Commit

func (co *QueueConsumer) Commit(ctx context.Context, itemID uint64) error

Commit - фиксирует успешный результат обработки указанного элемента очереди. При этом элемент удаляется из очереди и добавляется в список выполненных.

func (*QueueConsumer) ReadItems

func (co *QueueConsumer) ReadItems(ctx context.Context, limit uint32) (itemsIDs []uint64, err error)

ReadItems - читает ограниченный список элементов из очереди находящихся в статусе READY в порядке их добавления и переводит эти элементы в статус PROCESSING.

func (*QueueConsumer) Reject

func (co *QueueConsumer) Reject(ctx context.Context, itemID uint64, causeErr error) error

Reject - отклоняет результат обработки указанного элемента очереди с указанием причины. При этом элемент переводится в статус RETRY с фиксацией ошибки в журнале.