repository

Documentation

Index

• type BrokenPostgres
  • func NewBrokenPostgres(client mrstorage.DBConnManager, table mrsql.DBTableInfo) *BrokenPostgres
  • func (re *BrokenPostgres) Delete(ctx context.Context, expiry time.Duration, limit uint32) (rowsIDs []uint64, err error)
  • func (re *BrokenPostgres) Insert(ctx context.Context, rows []entity.ItemWithError) error
  • func (re *BrokenPostgres) InsertOne(ctx context.Context, row entity.ItemWithError) error
• type CompletedPostgres
  • func NewCompletedPostgres(client mrstorage.DBConnManager, table mrsql.DBTableInfo) *CompletedPostgres
  • func (re *CompletedPostgres) Delete(ctx context.Context, expiry time.Duration, limit uint32) (rowsIDs []uint64, err error)
  • func (re *CompletedPostgres) Insert(ctx context.Context, rowID uint64) error
• type QueuePostgres
  • func NewQueuePostgres(client mrstorage.DBConnManager, table mrsql.DBTableInfo) *QueuePostgres
  • func (re *QueuePostgres) Delete(ctx context.Context, rowID uint64, status enum.ItemStatus) error
  • func (re *QueuePostgres) DeleteRetryWithoutAttempts(ctx context.Context, limit uint32) (rowsIDs []uint64, err error)
  • func (re *QueuePostgres) FetchAndUpdateStatusProcessingToRetryByTimeout(ctx context.Context, timeout time.Duration, limit uint32) (rowIDs []uint64, err error)
  • func (re *QueuePostgres) FetchAndUpdateStatusReadyToProcessing(ctx context.Context, limit uint32) (rowsIDs []uint64, err error)
  • func (re *QueuePostgres) FetchAndUpdateStatusRetryToReady(ctx context.Context, delayed time.Duration, limit uint32) (rowIDs []uint64, err error)
  • func (re *QueuePostgres) Insert(ctx context.Context, rows []entity.Item) error
  • func (re *QueuePostgres) UpdateStatusProcessingToReady(ctx context.Context, rowsIDs []uint64) error
  • func (re *QueuePostgres) UpdateStatusProcessingToRetry(ctx context.Context, rowID uint64) error

Types

type BrokenPostgres

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

BrokenPostgres - репозиторий для хранения ошибок записей, которые случились при их обработке.

func NewBrokenPostgres

func NewBrokenPostgres(client mrstorage.DBConnManager, table mrsql.DBTableInfo) *BrokenPostgres

NewBrokenPostgres - создаёт объект BrokenPostgres.

func (*BrokenPostgres) Delete

func (re *BrokenPostgres) Delete(ctx context.Context, expiry time.Duration, limit uint32) (rowsIDs []uint64, err error)

Delete - удаляет ограниченный список записей из журнала ошибок. Возвращает ID записей, которые были удалены.

func (*BrokenPostgres) Insert

func (re *BrokenPostgres) Insert(ctx context.Context, rows []entity.ItemWithError) error

Insert - добавляет указанный список записей в журнал ошибок.

func (*BrokenPostgres) InsertOne

func (re *BrokenPostgres) InsertOne(ctx context.Context, row entity.ItemWithError) error

InsertOne - добавляет указанную запись в журнал ошибок.

type CompletedPostgres

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

CompletedPostgres - репозиторий для хранения успешно обработанных записей.

func NewCompletedPostgres

func NewCompletedPostgres(client mrstorage.DBConnManager, table mrsql.DBTableInfo) *CompletedPostgres

NewCompletedPostgres - создаёт объект CompletedPostgres.

func (*CompletedPostgres) Delete

func (re *CompletedPostgres) Delete(ctx context.Context, expiry time.Duration, limit uint32) (rowsIDs []uint64, err error)

Delete - удаляет ограниченный список записей из успешно обработанных. Возвращает ID записей, которые были удалены.

func (*CompletedPostgres) Insert

func (re *CompletedPostgres) Insert(ctx context.Context, rowID uint64) error

Insert - добавляет указанную запись в список успешно обработанных.

type QueuePostgres

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

QueuePostgres - репозиторий для организации очереди и хранения в ней записей.

func NewQueuePostgres

func NewQueuePostgres(client mrstorage.DBConnManager, table mrsql.DBTableInfo) *QueuePostgres

NewQueuePostgres - создаёт объект QueuePostgres.

func (*QueuePostgres) Delete

func (re *QueuePostgres) Delete(ctx context.Context, rowID uint64, status enum.ItemStatus) error

Delete - удаляет запись из очереди по указанному ID и находящеюся в указанном статусе.

func (*QueuePostgres) DeleteRetryWithoutAttempts

func (re *QueuePostgres) DeleteRetryWithoutAttempts(ctx context.Context, limit uint32) (rowsIDs []uint64, err error)

DeleteRetryWithoutAttempts - удаляет из очереди ограниченный список записей находящихся в статусе RETRY и с нулевым кол-вом попыток в целях разгрузки очереди. Возвращает ID записей, которые были удалены.

func (*QueuePostgres) FetchAndUpdateStatusProcessingToRetryByTimeout

func (re *QueuePostgres) FetchAndUpdateStatusProcessingToRetryByTimeout(ctx context.Context, timeout time.Duration, limit uint32) (rowIDs []uint64, err error)

FetchAndUpdateStatusProcessingToRetryByTimeout - возвращает ограниченный список записей находящихся долгое время в статусе PROCESSING (например, в случае если обработка записи подвисла) предварительно переведя их в статус RETRY.

func (*QueuePostgres) FetchAndUpdateStatusReadyToProcessing

func (re *QueuePostgres) FetchAndUpdateStatusReadyToProcessing(ctx context.Context, limit uint32) (rowsIDs []uint64, err error)

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

func (*QueuePostgres) FetchAndUpdateStatusRetryToReady

func (re *QueuePostgres) FetchAndUpdateStatusRetryToReady(ctx context.Context, delayed time.Duration, limit uint32) (rowIDs []uint64, err error)

FetchAndUpdateStatusRetryToReady - переводит ограниченный список записей из статуса RETRY в статус READY учитывая указанную задержку нахождения записи в этом статусе и положительное кол-во попыток.

func (*QueuePostgres) Insert

func (re *QueuePostgres) Insert(ctx context.Context, rows []entity.Item) error

Insert - добавляет список записей в очередь со статусом READY. Если указано ReadyDelayed, то обработка записи откладывается на указанный период времени.

func (*QueuePostgres) UpdateStatusProcessingToReady

func (re *QueuePostgres) UpdateStatusProcessingToReady(ctx context.Context, rowsIDs []uint64) error

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

func (*QueuePostgres) UpdateStatusProcessingToRetry

func (re *QueuePostgres) UpdateStatusProcessingToRetry(ctx context.Context, rowID uint64) error

UpdateStatusProcessingToRetry - переводит указанную запись из статуса PROCESSING в статус RETRY, с уменьшением кол-ва попыток (например, в случае возникновения ошибки при обработке этой записи).