postgres

Documentation

Index

• func TruncateForTest(t *testing.T, s *Store)
• type Option
  • func WithDLQRetention(maxAge time.Duration, maxDepth int) Option
  • func WithDeliveredRetention(maxAge time.Duration) Option
  • func WithNowFunc(now func() time.Time) Option
  • func WithPollInterval(d time.Duration) Option
  • func WithQueueLimits(maxDepth int, dropPolicy string) Option
  • func WithRetention(maxAge, pruneInterval time.Duration) Option
• type Store
  • func NewStore(dsn string, opts ...Option) (*Store, error)
  • func (s *Store) Ack(leaseID string) error
  • func (s *Store) AckBatch(leaseIDs []string) (queue.LeaseBatchResult, error)
  • func (s *Store) CancelMessages(req queue.MessageCancelRequest) (queue.MessageCancelResponse, error)
  • func (s *Store) CancelMessagesByFilter(req queue.MessageManageFilterRequest) (queue.MessageCancelResponse, error)
  • func (s *Store) CaptureBacklogTrendSample(at time.Time) error
  • func (s *Store) Close() error
  • func (s *Store) Delete(poolName, id string) (bool, error)
  • func (s *Store) DeleteDead(req queue.DeadDeleteRequest) (queue.DeadDeleteResponse, error)
  • func (s *Store) Dequeue(req queue.DequeueRequest) (queue.DequeueResponse, error)
  • func (s *Store) Enqueue(env queue.Envelope) error
  • func (s *Store) Extend(leaseID string, extendBy time.Duration) error
  • func (s *Store) ListAll() ([]secrets.Record, error)
  • func (s *Store) ListAttempts(req queue.AttemptListRequest) (queue.AttemptListResponse, error)
  • func (s *Store) ListBacklogTrend(req queue.BacklogTrendListRequest) (queue.BacklogTrendListResponse, error)
  • func (s *Store) ListByPool(poolName string) ([]secrets.Record, error)
  • func (s *Store) ListDead(req queue.DeadListRequest) (queue.DeadListResponse, error)
  • func (s *Store) ListMessages(req queue.MessageListRequest) (queue.MessageListResponse, error)
  • func (s *Store) LookupMessages(req queue.MessageLookupRequest) (queue.MessageLookupResponse, error)
  • func (s *Store) MarkDead(leaseID string, reason string) error
  • func (s *Store) MarkDeadBatch(leaseIDs []string, reason string) (queue.LeaseBatchResult, error)
  • func (s *Store) Nack(leaseID string, delay time.Duration) error
  • func (s *Store) NackBatch(leaseIDs []string, delay time.Duration) (queue.LeaseBatchResult, error)
  • func (s *Store) NotifyCh() <-chan struct{}
  • func (s *Store) RecordAttempt(attempt queue.DeliveryAttempt) error
  • func (s *Store) RequeueDead(req queue.DeadRequeueRequest) (queue.DeadRequeueResponse, error)
  • func (s *Store) RequeueMessages(req queue.MessageRequeueRequest) (queue.MessageRequeueResponse, error)
  • func (s *Store) RequeueMessagesByFilter(req queue.MessageManageFilterRequest) (queue.MessageRequeueResponse, error)
  • func (s *Store) ResumeMessages(req queue.MessageResumeRequest) (queue.MessageResumeResponse, error)
  • func (s *Store) ResumeMessagesByFilter(req queue.MessageManageFilterRequest) (queue.MessageResumeResponse, error)
  • func (s *Store) RuntimeMetrics() queue.StoreRuntimeMetrics
  • func (s *Store) Stats() (queue.Stats, error)
  • func (s *Store) Upsert(rec secrets.Record) error

Functions

func TruncateForTest

func TruncateForTest(t *testing.T, s *Store)

TruncateForTest clears all queue/attempt/trend/secret tables on the given store. It is exported solely for cross-package integration tests and panics on any error so callers don't have to plumb the *testing.T deeper.

Types

type Option

type Option func(*Store)

func WithDLQRetention

func WithDLQRetention(maxAge time.Duration, maxDepth int) Option

func WithDeliveredRetention

func WithDeliveredRetention(maxAge time.Duration) Option

func WithNowFunc

func WithNowFunc(now func() time.Time) Option

func WithPollInterval

func WithPollInterval(d time.Duration) Option

func WithQueueLimits

func WithQueueLimits(maxDepth int, dropPolicy string) Option

func WithRetention

func WithRetention(maxAge, pruneInterval time.Duration) Option

type Store

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

func NewStore

func NewStore(dsn string, opts ...Option) (*Store, error)

func (*Store) Ack

func (s *Store) Ack(leaseID string) error

func (*Store) AckBatch

func (s *Store) AckBatch(leaseIDs []string) (queue.LeaseBatchResult, error)

func (*Store) CancelMessages

func (s *Store) CancelMessages(req queue.MessageCancelRequest) (queue.MessageCancelResponse, error)

func (*Store) CancelMessagesByFilter

func (s *Store) CancelMessagesByFilter(req queue.MessageManageFilterRequest) (queue.MessageCancelResponse, error)

func (*Store) CaptureBacklogTrendSample

func (s *Store) CaptureBacklogTrendSample(at time.Time) error

func (*Store) Close

func (s *Store) Close() error

func (*Store) Delete

func (s *Store) Delete(poolName, id string) (bool, error)

func (*Store) DeleteDead

func (s *Store) DeleteDead(req queue.DeadDeleteRequest) (queue.DeadDeleteResponse, error)

func (*Store) Dequeue

func (s *Store) Dequeue(req queue.DequeueRequest) (queue.DequeueResponse, error)

func (*Store) Enqueue

func (s *Store) Enqueue(env queue.Envelope) error

func (*Store) Extend

func (s *Store) Extend(leaseID string, extendBy time.Duration) error

func (*Store) ListAll

func (s *Store) ListAll() ([]secrets.Record, error)

func (*Store) ListAttempts

func (s *Store) ListAttempts(req queue.AttemptListRequest) (queue.AttemptListResponse, error)

func (*Store) ListBacklogTrend

func (s *Store) ListBacklogTrend(req queue.BacklogTrendListRequest) (queue.BacklogTrendListResponse, error)

func (*Store) ListByPool

func (s *Store) ListByPool(poolName string) ([]secrets.Record, error)

func (*Store) ListDead

func (s *Store) ListDead(req queue.DeadListRequest) (queue.DeadListResponse, error)

func (*Store) ListMessages

func (s *Store) ListMessages(req queue.MessageListRequest) (queue.MessageListResponse, error)

func (*Store) LookupMessages

func (s *Store) LookupMessages(req queue.MessageLookupRequest) (queue.MessageLookupResponse, error)

func (*Store) MarkDead

func (s *Store) MarkDead(leaseID string, reason string) error

func (*Store) MarkDeadBatch

func (s *Store) MarkDeadBatch(leaseIDs []string, reason string) (queue.LeaseBatchResult, error)

func (*Store) Nack

func (s *Store) Nack(leaseID string, delay time.Duration) error

func (*Store) NackBatch

func (s *Store) NackBatch(leaseIDs []string, delay time.Duration) (queue.LeaseBatchResult, error)

func (*Store) NotifyCh

func (s *Store) NotifyCh() <-chan struct{}

NotifyCh returns a channel that is closed when new items become available. After the channel fires, callers must call NotifyCh again to get the next signal.

func (*Store) RecordAttempt

func (s *Store) RecordAttempt(attempt queue.DeliveryAttempt) error

func (*Store) RequeueDead

func (s *Store) RequeueDead(req queue.DeadRequeueRequest) (queue.DeadRequeueResponse, error)

func (*Store) RequeueMessages

func (s *Store) RequeueMessages(req queue.MessageRequeueRequest) (queue.MessageRequeueResponse, error)

func (*Store) RequeueMessagesByFilter

func (s *Store) RequeueMessagesByFilter(req queue.MessageManageFilterRequest) (queue.MessageRequeueResponse, error)

func (*Store) ResumeMessages

func (s *Store) ResumeMessages(req queue.MessageResumeRequest) (queue.MessageResumeResponse, error)

func (*Store) ResumeMessagesByFilter

func (s *Store) ResumeMessagesByFilter(req queue.MessageManageFilterRequest) (queue.MessageResumeResponse, error)

func (*Store) RuntimeMetrics

func (s *Store) RuntimeMetrics() queue.StoreRuntimeMetrics

func (*Store) Stats

func (s *Store) Stats() (queue.Stats, error)

func (*Store) Upsert

func (s *Store) Upsert(rec secrets.Record) error