postgres

Documentation

Index

• type Storage
  • func NewStorage(db *pgxpool.Pool) *Storage
  • func (s *Storage) CompleteJob(ctx context.Context, id uuid.UUID) error
  • func (s *Storage) DeleteColdJobs(ctx context.Context, cutoffDate time.Time, limit uint) (uint, error)
  • func (s *Storage) DeleteDeadJobs(ctx context.Context, cutoffDate time.Time, limit uint) (uint, error)
  • func (s *Storage) FailJob(ctx context.Context, id uuid.UUID, errorMessage string) error
  • func (s *Storage) InsertBatchJobs(ctx context.Context, batcher pqueue.BatchSender, ...) ([]*pqueue.Job, error)
  • func (s *Storage) InsertJob(ctx context.Context, queryer pqueue.QueryRower, id uuid.UUID, ...) (*pqueue.Job, error)
  • func (s *Storage) ListActiveJobs(ctx context.Context, queues []string, batchSize uint) ([]pqueue.Job, error)
  • func (s *Storage) ReScheduleJob(ctx context.Context, id uuid.UUID, scheduledAt time.Time, errorMessage string) error

Types

type Storage

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

Storage is a default Postgres-based implementation of pqueue interface.

func NewStorage

func NewStorage(db *pgxpool.Pool) *Storage

NewStorage creates a new Postgres storage.

func (*Storage) CompleteJob

func (s *Storage) CompleteJob(ctx context.Context, id uuid.UUID) error

CompleteJob marks a job in storage as completed.

func (*Storage) DeleteColdJobs

func (s *Storage) DeleteColdJobs(ctx context.Context, cutoffDate time.Time, limit uint) (uint, error)

DeleteColdJobs removes completed jobs.

func (*Storage) DeleteDeadJobs

func (s *Storage) DeleteDeadJobs(ctx context.Context, cutoffDate time.Time, limit uint) (uint, error)

DeleteDeadJobs removes dead jobs.

func (*Storage) FailJob

func (s *Storage) FailJob(ctx context.Context, id uuid.UUID, errorMessage string) error

FailJob marks a job in storage as failed.

func (*Storage) InsertBatchJobs

func (s *Storage) InsertBatchJobs(
	ctx context.Context,
	batcher pqueue.BatchSender,
	jobs []pqueue.PreparedBatchJob,
) ([]*pqueue.Job, error)

InsertBatchJobs inserts a batch of jobs into storage.

func (*Storage) InsertJob

func (s *Storage) InsertJob(
	ctx context.Context,
	queryer pqueue.QueryRower,
	id uuid.UUID,
	idempotencyKey uuid.UUID,
	queue string,
	payload []byte,
	options pqueue.JobOptions,
) (*pqueue.Job, error)

InsertJob inserts a new job into storage.

func (*Storage) ListActiveJobs

func (s *Storage) ListActiveJobs(ctx context.Context, queues []string, batchSize uint) ([]pqueue.Job, error)

ListActiveJobs fetches a batch of active jobs from storage for the specified queues. If no queues specified jobs for all queues will be fetched.

func (*Storage) ReScheduleJob

func (s *Storage) ReScheduleJob(
	ctx context.Context,
	id uuid.UUID,
	scheduledAt time.Time,
	errorMessage string,
) error

ReScheduleJob schedules a job for a further reprocessing.