pgq

Documentation

Index

• type Config
  • func (c Config) ConnMaxLifetimeWithJitter() time.Duration
  • func (c Config) ConnectionString() string
  • func (c Config) ConnectionURL() string
• type Processor
  • func NewProcessor(ctx context.Context, cfg Config) (*Processor, error)
  • func (p *Processor) ClearDeadJobs(ctx context.Context, jobIDs []string) error
  • func (p *Processor) Close() error
  • func (p *Processor) DeadJobs(ctx context.Context, size, offset int) ([]worker.Job, error)
  • func (p *Processor) Enqueue(ctx context.Context, jobs ...worker.Job) error
  • func (p *Processor) GetSyncJobsByService(ctx context.Context, service string) ([]worker.Job, error)
  • func (p *Processor) Process(ctx context.Context, types []string, fn worker.JobExecutorFunc) error
  • func (p *Processor) Resurrect(ctx context.Context, jobIDs []string) error
  • func (p *Processor) Stats(ctx context.Context) ([]worker.JobTypeStats, error)

Types

type Config

type Config struct {
	Host     string `mapstructure:"host" default:"localhost"`
	Port     int    `mapstructure:"port" default:"5432"`
	Name     string `mapstructure:"name" default:"postgres"`
	Username string `mapstructure:"username" default:"root"`
	Password string `mapstructure:"password" default:""`

	// Connection pool settings
	MaxOpenConns          int           `mapstructure:"max_open_conns" default:"10"`
	MaxIdleConns          int           `mapstructure:"max_idle_conns" default:"4"`
	ConnMaxIdleTime       time.Duration `mapstructure:"conn_max_idle_time" default:"5m"`
	ConnMaxLifetime       time.Duration `mapstructure:"conn_max_lifetime" default:"5m"`
	ConnMaxLifetimeJitter time.Duration `mapstructure:"conn_max_lifetime_jitter" default:"2m"`
}

func (Config) ConnMaxLifetimeWithJitter

func (c Config) ConnMaxLifetimeWithJitter() time.Duration

func (Config) ConnectionString

func (c Config) ConnectionString() string

func (Config) ConnectionURL

func (c Config) ConnectionURL() string

type Processor

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

Processor implements a JobProcessor backed by PostgreSQL.

func NewProcessor

func NewProcessor(ctx context.Context, cfg Config) (*Processor, error)

NewProcessor returns a JobProcessor implementation backed by the PostgreSQL instance identified by the provided config.

func (*Processor) ClearDeadJobs

func (p *Processor) ClearDeadJobs(ctx context.Context, jobIDs []string) error

func (*Processor) Close

func (p *Processor) Close() error

func (*Processor) DeadJobs

func (p *Processor) DeadJobs(ctx context.Context, size, offset int) ([]worker.Job, error)

func (*Processor) Enqueue

func (p *Processor) Enqueue(ctx context.Context, jobs ...worker.Job) error

func (*Processor) GetSyncJobsByService

func (p *Processor) GetSyncJobsByService(ctx context.Context, service string) ([]worker.Job, error)

func (*Processor) Process

func (p *Processor) Process(ctx context.Context, types []string, fn worker.JobExecutorFunc) error

func (*Processor) Resurrect

func (p *Processor) Resurrect(ctx context.Context, jobIDs []string) error

func (*Processor) Stats

func (p *Processor) Stats(ctx context.Context) ([]worker.JobTypeStats, error)