mrqueue

Documentation

Index

• type BrokenStorage
• type Changer
• type Cleaner
• type CompletedStorage
• type Consumer
• type Producer
• type Storage

Types

type BrokenStorage

type BrokenStorage interface {
	Insert(ctx context.Context, rows []entity.ItemWithError) error
	InsertOne(ctx context.Context, row entity.ItemWithError) error
	Delete(ctx context.Context, expiry time.Duration, limit uint32) (rowsIDs []uint64, err error)
}

BrokenStorage - предоставляет доступ к хранилищу сломанных элементов.

type Changer

type Changer interface {
	ChangeProcessingToRetryByTimeout(ctx context.Context, limit uint32) (itemsIDs []uint64, err error)
	ChangeRetryToReady(ctx context.Context, limit uint32) (itemsIDs []uint64, err error)
}

Changer - изменяет статусы сломавшихся элементов, находящихся в очереди.

type Cleaner

type Cleaner interface {
	RemoveItemsWithoutAttempts(ctx context.Context, limit uint32) (itemsIDs []uint64, err error)
	RemoveCompletedItems(ctx context.Context, limit uint32) (itemsIDs []uint64, err error)
	RemoveBrokenItems(ctx context.Context, limit uint32) (itemsIDs []uint64, err error)
}

Cleaner - очищает очередь от обработанных/сломанных элементов.

type CompletedStorage

type CompletedStorage interface {
	Insert(ctx context.Context, rowID uint64) error
	Delete(ctx context.Context, expiry time.Duration, limit uint32) (rowsIDs []uint64, err error)
}

CompletedStorage - предоставляет доступ к хранилищу успешно обработанных элементов.

type Consumer

type Consumer interface {
	ReadItems(ctx context.Context, limit uint32) (itemsIDs []uint64, err error)
	CancelItems(ctx context.Context, itemsIDs []uint64) error
	Commit(ctx context.Context, itemID uint64) error
	Reject(ctx context.Context, itemID uint64, causeErr error) error
}

Consumer - читает элементы из очереди и информирует о статусе их обработки.

type Producer

type Producer interface {
	Append(ctx context.Context, item entity.Item) error
	Appends(ctx context.Context, items []entity.Item) error
}

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

type Storage

type Storage interface {
	Insert(ctx context.Context, rows []entity.Item) error
	FetchAndUpdateStatusReadyToProcessing(ctx context.Context, limit uint32) (rowsIDs []uint64, err error)
	UpdateStatusProcessingToReady(ctx context.Context, rowsIDs []uint64) error
	UpdateStatusProcessingToRetry(ctx context.Context, rowID uint64) error
	FetchAndUpdateStatusProcessingToRetryByTimeout(ctx context.Context, timeout time.Duration, limit uint32) (rowIDs []uint64, err error)
	FetchAndUpdateStatusRetryToReady(ctx context.Context, delayed time.Duration, limit uint32) (rowIDs []uint64, err error)
	Delete(ctx context.Context, rowID uint64, status enum.ItemStatus) error
	DeleteRetryWithoutAttempts(ctx context.Context, limit uint32) (rowsIDs []uint64, err error)
}

Storage - предоставляет доступ к хранилищу очереди элементов.