hopper

package module

Go to main page

Versions in this module

v0

v0.12.0

Nov 4, 2026

Changes in this version

+ const DefaultStuckGrace

+ var ErrNotLeader = errors.New("hopper: not the leader")

+ var ErrOutputTooLarge = errors.New("hopper: output larger than Config.MaxOutputBytes")

type Client

+ func (c *Client[TTx]) LeaderResign(ctx context.Context, leaderID int64) (int64, error)

+ func (c *Client[TTx]) StopAndCancel(ctx context.Context) error

+ func (c *Client[TTx]) Stopped() <-chan struct{}

type Config

+ MaxOutputBytes int

+ OnStuck func(ctx context.Context, job *JobRow, cause error)

+ StuckGrace time.Duration

type EventKind

+ const EventJobStuck

+ const EventLeaderResigned

+ type OutputTooLargeError struct

+ Limit int

+ Size int

+ func (e *OutputTooLargeError) Error() string

+ func (e *OutputTooLargeError) Is(target error) bool

v0.11.0

Nov 2, 2026

Changes in this version

type StepOpts

+ AfterAny [][]Step

+ func AfterAny(steps ...Step) *StepOpts

v0.10.2

Oct 31, 2026

v0.10.1

Oct 29, 2026

v0.10.0

Oct 29, 2026

Changes in this version

+ const DefaultLogListSize

+ const DefaultLogReadBytes

+ const MaxLogListSize

+ const MaxLogReadBytes

+ var ErrNotSupported = driver.ErrNotSupported

type Client

+ func (c *Client[TTx]) JobLogFollow(ctx context.Context, id JobID, opts *LogFollowOpts) iter.Seq2[LogUpdate, error]

+ func (c *Client[TTx]) JobLogRead(ctx context.Context, id JobID, opts *LogReadOpts) (LogReadResult, error)

+ func (c *Client[TTx]) JobLogRuns(ctx context.Context, id JobID, after string, limit int) (LogRunPage, error)

+ func (c *Client[TTx]) LogRetentionStats() LogRetentionStats

+ func (c *Client[TTx]) MaintainJobLogs(ctx context.Context) (result LogRetentionResult, err error)

+ type Dependency = driver.Dependency

type JobLogsConfig

+ Retention *LogRetentionConfig

+ type LogChunk struct

+ Log string

+ NextOffset int64

+ Offset int64

+ Size int64

+ type LogCursor struct

+ Fence int64

+ Offset int64

+ type LogFollowOpts struct

+ PollInterval time.Duration

+ type LogListParams struct

+ After string

+ Fence int64

+ JobID JobID

+ Limit int

+ type LogNotifications struct

+ func (n *LogNotifications) Notify(id JobID)

+ func (n *LogNotifications) Watch(ctx context.Context, id JobID) (<-chan struct{}, error)

+ type LogReadOpts struct

+ Cursor LogCursor

+ Fence int64

+ Limit int

+ type LogReadParams struct

+ Fence int64

+ JobID JobID

+ Limit int

+ Offset int64

+ type LogReadResult struct

+ Cursor LogCursor

+ Run LogRun

+ type LogRetentionConfig struct

+ BatchSize int

+ CompletedRetention time.Duration

+ FailedRetention time.Duration

+ GracePeriod time.Duration

+ Interval time.Duration

+ MaxRunBytes int64

+ MaxRuns int

+ Timeout time.Duration

+ type LogRetentionResult struct

+ At time.Time

+ Deleted int

+ DeletedBytes int64

+ More bool

+ Scanned int

+ ScannedBytes int64

+ type LogRetentionStats struct

+ Deleted int64

+ DeletedBytes int64

+ Failures int64

+ InventoryAt time.Time

+ InventoryBytes int64

+ InventoryRuns int64

+ Last LogRetentionResult

+ LastError string

+ Passes int64

+ type LogRun struct

+ Attempt int

+ Fence int64

+ JobID JobID

+ Size int64

+ UpdatedAt time.Time

+ Version string

+ type LogRunPage struct

+ Next string

+ Runs []LogRun

+ type LogStoreLister interface

+ List func(ctx context.Context, params LogListParams) (LogRunPage, error)

+ type LogStorePruner interface

+ Delete func(ctx context.Context, run LogRun) (bool, error)

+ type LogStoreReader interface

+ Read func(ctx context.Context, params LogReadParams) (LogChunk, error)

+ type LogStoreWatcher interface

+ Watch func(ctx context.Context, id JobID) (<-chan struct{}, error)

+ type LogUpdate struct

+ Missing bool

+ Reset bool

type StepOpts

+ Dependencies bool

+ Name string

v0.9.0

Oct 28, 2026

Changes in this version

+ const DefaultJobLeaseTTL

+ const RescueClientLost

+ const RescueLeaseExpired

+ const RescueStuck

+ var ErrJobNotClaimable = driver.ErrJobNotClaimable

type Client

+ func (c *Client[TTx]) Leases() *Leases[TTx]

type Config

+ JobLeaseGrace time.Duration

+ type JobLease struct

+ ExpiresAt time.Time

+ Job *JobRow

+ Token LeaseToken

+ type LeaseOpts struct

+ TTL time.Duration

+ type LeaseResult struct

+ Err error

+ Output any

+ type LeaseStatus struct

+ CancelRequested bool

+ ExpiresAt time.Time

+ type LeaseToken struct

+ func ParseLeaseToken(s string) (LeaseToken, error)

+ func (t *LeaseToken) UnmarshalText(b []byte) error

+ func (t LeaseToken) Fence() int64

+ func (t LeaseToken) IsZero() bool

+ func (t LeaseToken) JobID() JobID

+ func (t LeaseToken) MarshalText() ([]byte, error)

+ func (t LeaseToken) String() string

+ type Leases struct

+ func (l *Leases[TTx]) Claim(ctx context.Context, id JobID, fence int64, opts *LeaseOpts) (*JobLease, error)

+ func (l *Leases[TTx]) ClaimTx(ctx context.Context, tx TTx, id JobID, fence int64, opts *LeaseOpts) (*JobLease, error)

+ func (l *Leases[TTx]) Finish(ctx context.Context, token LeaseToken, res LeaseResult) error

+ func (l *Leases[TTx]) FinishTx(ctx context.Context, tx TTx, token LeaseToken, res LeaseResult) error

+ func (l *Leases[TTx]) RecordCursor(ctx context.Context, token LeaseToken, cursor any) (*Progress, error)

+ func (l *Leases[TTx]) RecordCursorTx(ctx context.Context, tx TTx, token LeaseToken, cursor any) (*Progress, error)

+ func (l *Leases[TTx]) RecordStep(ctx context.Context, token LeaseToken, step string, cursor any) (*Progress, error)

+ func (l *Leases[TTx]) RecordStepTx(ctx context.Context, tx TTx, token LeaseToken, step string, cursor any) (*Progress, error)

+ func (l *Leases[TTx]) Renew(ctx context.Context, token LeaseToken) (*LeaseStatus, error)

+ type RescueReason = driver.RescueReason

v0.8.0

Oct 26, 2026

Changes in this version

+ const DefaultJobLogAttempts

+ const DefaultJobLogBytes

+ const MaxJobLogAttempts

+ const MaxJobLogBytes

+ const MaxJobLogTotalBytes

+ const MaxTagBytes

+ const MaxTags

+ var ErrInvalidFilter = driver.ErrInvalidFilter

+ var ErrJobLocked = driver.ErrJobLocked

+ func Logger(ctx context.Context) *slog.Logger

+ func RecordCursorTx[TTx any, T JobArgs](ctx context.Context, tx TTx, job *Job[T], cursor any) error

+ func RecordCursor[T JobArgs](ctx context.Context, job *Job[T], cursor any) error

+ func RecordStepTx[TTx any, T JobArgs](ctx context.Context, tx TTx, job *Job[T], step string, cursor any) error

+ func RecordStep[T JobArgs](ctx context.Context, job *Job[T], step string, cursor any) error

+ func RunSteps[T JobArgs](ctx context.Context, job *Job[T], steps ...JobStep[T]) error

+ func StepCursor[C any](job *JobRow) (cursor C, ok bool, err error)

+ type AttemptLog = driver.AttemptLog

type Client

+ func (c *Client[TTx]) JobCancelTx(ctx context.Context, tx TTx, id JobID) (*JobRow, error)

+ func (c *Client[TTx]) JobDelete(ctx context.Context, id JobID) (*JobRow, error)

+ func (c *Client[TTx]) JobDeleteTx(ctx context.Context, tx TTx, id JobID) (*JobRow, error)

+ func (c *Client[TTx]) JobLogs(ctx context.Context, id JobID) ([]AttemptLog, error)

+ func (c *Client[TTx]) JobRetryOrderingKeyTx(ctx context.Context, tx TTx, queue, orderingKey string) (int, error)

+ func (c *Client[TTx]) JobRetryTx(ctx context.Context, tx TTx, id JobID, opts ...RetryOption) (*JobRow, error)

+ func (c *Client[TTx]) PeriodicJobs() *PeriodicJobs

type Config

+ JobLogs JobLogsConfig

+ OnError func(ctx context.Context, job *JobRow, err error) ErrorOutcome

+ RetryPolicy RetryPolicy

+ type ErrorOutcome struct

+ Discard bool

+ RetryAt time.Time

+ Snooze time.Duration

+ type ExponentialRetry struct

+ Base time.Duration

+ Max time.Duration

+ func (r ExponentialRetry) NextRetry(job *JobRow) time.Time

type InsertOpts

+ Tags []string

+ type JobArgsWithKindAliases interface

+ KindAliases func() []string

type JobFilter

+ IDs []JobID

+ Metadata json.RawMessage

+ Priorities []Priority

+ TagsAll []string

+ TagsAny []string

+ type JobLogsConfig struct

+ Level slog.Leveler

+ MaxAttempts int

+ MaxBytes int

+ ReplaceAttr func(groups []string, a slog.Attr) slog.Attr

+ Store LogStore

+ type JobStep struct

+ func NamedStep[T JobArgs](name string, run func(ctx context.Context, job *Job[T]) error) JobStep[T]

+ type LogStore interface

+ Logs func(ctx context.Context, id JobID) ([]AttemptLog, error)

+ Writer func(ctx context.Context, job *JobRow) (io.WriteCloser, error)

+ type PeriodicJobs struct

+ func (p *PeriodicJobs) Add(job PeriodicJob) error

+ func (p *PeriodicJobs) AddMany(jobs []PeriodicJob) error

+ func (p *PeriodicJobs) All() []PeriodicJob

+ func (p *PeriodicJobs) Clear()

+ func (p *PeriodicJobs) Remove(name string) error

+ type Progress = driver.Progress

type QueueConfig

+ RetryPolicy RetryPolicy

+ type RetryOption func(*driver.JobRetryParams)

+ func FromStart() RetryOption

+ type RetryPolicy interface

+ NextRetry func(job *JobRow) time.Time

type UniqueOpts

+ Finished []JobState

v0.7.0

Oct 14, 2026

Changes in this version

+ const JobStateCanceled

+ var ErrJobCanceled = errors.New("hopper: job canceled")

+ var ErrJobFenced = errors.New("hopper: job is no longer running in this attempt")

+ func CompleteTx[TTx any, T JobArgs](ctx context.Context, tx TTx, job *Job[T]) error

type Client

+ func (c *Client[TTx]) JobRetryOrderingKey(ctx context.Context, queue, orderingKey string) (int, error)

type EventKind

+ const EventJobCanceled

type JobFilter

+ NewestFirst bool

+ OrderingKey string

v0.6.0

Oct 13, 2026

Changes in this version

+ var ErrSchemaOutdated = errors.New("hopper: database schema is out of date")

type SnoozeError

+ Reason string

v0.5.0

Oct 3, 2026

Changes in this version

type Config

+ NeverLead bool

v0.4.0

Oct 2, 2026

Changes in this version

+ const DefaultLeaseTTL

type Config

+ LeaseTTL time.Duration

v0.3.1

Oct 2, 2026

v0.3.0

Oct 1, 2026

v0.2.0

Oct 1, 2026

Changes in this version

+ func ContextWithClient[TTx any](ctx context.Context, client *Client[TTx]) context.Context

type Client

+ func ClientFromContext[TTx any](ctx context.Context) (*Client[TTx], bool)

v0.1.2

Oct 1, 2026

v0.1.1

Sep 29, 2026

v0.1.0

Sep 29, 2026

Changes in this version

+ const ChannelControl

+ const DefaultCompletedRetention

+ const DefaultFailedRetention

+ const DefaultJobTimeout

+ const DefaultMaxAttempts

+ const DefaultPollInterval

+ const DefaultStopTimeout

+ const DefaultStreamRetention

+ const DependencyCancel

+ const DependencyIgnore

+ const JobStateAvailable

+ const JobStateCancelled

+ const JobStateCompleted

+ const JobStateDiscarded

+ const JobStatePending

+ const JobStateRetryable

+ const JobStateRunning

+ const JobStateScheduled

+ const QueueDefault

+ const StreamStartEarliest

+ const StreamStartLatest

+ var ErrClientStarted = errors.New("hopper: client already started")

+ var ErrClientStopped = errors.New("hopper: client stopped")

+ var ErrJobCancelled = errors.New("hopper: job cancelled")

+ var ErrJobRunning = driver.ErrJobRunning

+ var ErrJobTimeout = errors.New("hopper: job timed out")

+ var ErrNotFound = driver.ErrNotFound

+ var ErrNotInJob = errors.New("hopper: context does not belong to a running job")

+ var ErrUniqueConflict = driver.ErrUniqueConflict

+ func AddWorkFunc[T JobArgs](workers *Workers, fn func(ctx context.Context, job *Job[T]) error)

+ func AddWorker[T JobArgs](workers *Workers, worker Worker[T])

+ func Await[T any, TTx any](ctx context.Context, c *Client[TTx], id JobID) (T, error)

+ func Cancel(err error) error

+ func Consume[T any](workers *Workers, consumer Consumer, ...)

+ func PerSecond(n float64) float64

+ func SetOutput(ctx context.Context, v any) error

+ func Snooze(d time.Duration) error

+ func Subscribe[T any](workers *Workers, sub Subscription, ...)

+ type AppendOpts struct

+ Headers map[string]string

+ Key string

+ type AttemptError = driver.AttemptError

+ type Batch struct

+ func (b *Batch[TTx]) Add(args JobArgs, opts *InsertOpts)

+ func (b *Batch[TTx]) Insert(ctx context.Context) (*BatchResult, error)

+ func (b *Batch[TTx]) InsertTx(ctx context.Context, tx TTx) (*BatchResult, error)

+ type BatchOpts struct

+ CallbackOpts *InsertOpts

+ Metadata []byte

+ OnComplete JobArgs

+ OnFailure JobArgs

+ OnSuccess JobArgs

+ type BatchResult struct

+ ID JobID

+ Jobs []*InsertResult

+ type BatchRow = driver.BatchRow

+ type CancelError struct

+ Err error

+ func (e *CancelError) Error() string

+ func (e *CancelError) Unwrap() error

+ type Client struct

+ func NewClient[TTx any](d driver.Driver[TTx], cfg *Config) (*Client[TTx], error)

+ func (c *Client[TTx]) BatchGet(ctx context.Context, id JobID) (*BatchRow, error)

+ func (c *Client[TTx]) Clients(ctx context.Context) ([]*ClientRow, error)

+ func (c *Client[TTx]) Events(ctx context.Context, kinds ...EventKind) iter.Seq[Event]

+ func (c *Client[TTx]) Insert(ctx context.Context, args JobArgs, opts *InsertOpts) (*InsertResult, error)

+ func (c *Client[TTx]) InsertMany(ctx context.Context, params []InsertParams) ([]*InsertResult, error)

+ func (c *Client[TTx]) InsertManyTx(ctx context.Context, tx TTx, params []InsertParams) ([]*InsertResult, error)

+ func (c *Client[TTx]) InsertTx(ctx context.Context, tx TTx, args JobArgs, opts *InsertOpts) (*InsertResult, error)

+ func (c *Client[TTx]) InsertWorkflow(ctx context.Context, wf *Workflow) (*WorkflowResult, error)

+ func (c *Client[TTx]) InsertWorkflowTx(ctx context.Context, tx TTx, wf *Workflow) (*WorkflowResult, error)

+ func (c *Client[TTx]) JobCancel(ctx context.Context, id JobID) (*JobRow, error)

+ func (c *Client[TTx]) JobGet(ctx context.Context, id JobID) (*JobRow, error)

+ func (c *Client[TTx]) JobRetry(ctx context.Context, id JobID) (*JobRow, error)

+ func (c *Client[TTx]) Jobs(ctx context.Context, filter JobFilter) iter.Seq2[*JobRow, error]

+ func (c *Client[TTx]) JobsTx(ctx context.Context, tx TTx, filter JobFilter) iter.Seq2[*JobRow, error]

+ func (c *Client[TTx]) NewBatch(opts BatchOpts) *Batch[TTx]

+ func (c *Client[TTx]) Publish(ctx context.Context, msg Topic, opts *PublishOpts) (*PublishResult, error)

+ func (c *Client[TTx]) PublishTx(ctx context.Context, tx TTx, msg Topic, opts *PublishOpts) (*PublishResult, error)

+ func (c *Client[TTx]) Queues() *QueueControl[TTx]

+ func (c *Client[TTx]) ReplayDiscarded(ctx context.Context, subscription string) (int, error)

+ func (c *Client[TTx]) Run(ctx context.Context) error

+ func (c *Client[TTx]) Start(ctx context.Context) error

+ func (c *Client[TTx]) Stats(ctx context.Context) (*Stats, error)

+ func (c *Client[TTx]) Stop(ctx context.Context) error

+ func (c *Client[TTx]) Streams() *Streams[TTx]

+ func (c *Client[TTx]) Subscriptions(ctx context.Context) ([]*SubscriptionRow, error)

+ func (c *Client[TTx]) WorkflowGet(ctx context.Context, id JobID) (*WorkflowRow, error)

+ type ClientRow = driver.ClientRow

+ type Codec interface

+ Marshal func(v any) ([]byte, error)

+ Unmarshal func(data []byte, v any) error

+ type Config struct

+ Codec Codec

+ CompletedRetention time.Duration

+ FailedRetention time.Duration

+ Hostname string

+ JobTimeout time.Duration

+ Logger *slog.Logger

+ MaxAttempts int

+ Middleware []Middleware

+ Periodic []PeriodicJob

+ PollInterval time.Duration

+ Queues map[string]QueueConfig

+ RescueStuckAfter time.Duration

+ StopTimeout time.Duration

+ StreamRetention time.Duration

+ StrictKinds bool

+ Workers *Workers

+ type Consumer struct

+ MaxAttempts int

+ Metadata json.RawMessage

+ Name string

+ Pattern string

+ Queue string

+ Start StreamStart

+ func (c Consumer) Kind() string

+ type ConsumerRow = driver.StreamConsumerRow

+ type CronSchedule struct

+ func ParseCron(spec string) (*CronSchedule, error)

+ func (s *CronSchedule) In(loc *time.Location) *CronSchedule

+ func (s *CronSchedule) Next(t time.Time) time.Time

+ func (s *CronSchedule) String() string

+ type DependencyFailure = driver.DependencyFailure

+ type Event struct

+ At time.Time

+ ClientID int64

+ Error string

+ Job *JobRow

+ Kind EventKind

+ type EventKind string

+ const EventJobCancelled

+ const EventJobCompleted

+ const EventJobDiscarded

+ const EventJobFailed

+ const EventJobRescued

+ const EventJobSnoozed

+ const EventLeaderElected

+ const EventLeaderLost

+ const EventLeaseLost

+ type InsertOpts struct

+ Await bool

+ MaxAttempts int

+ Metadata json.RawMessage

+ OrderingKey string

+ PartitionKey string

+ Priority Priority

+ Queue string

+ ScheduledAt time.Time

+ TTL time.Duration

+ Unique *UniqueOpts

+ type InsertParams struct

+ Args JobArgs

+ Opts *InsertOpts

+ type InsertResult struct

+ Duplicate bool

+ Job *JobRow

+ type IntervalSchedule struct

+ Interval time.Duration

+ func (s IntervalSchedule) Next(t time.Time) time.Time

+ type JSONCodec struct

+ func (JSONCodec) Marshal(v any) ([]byte, error)

+ func (JSONCodec) Unmarshal(data []byte, v any) error

+ type Job struct

+ Args T

+ type JobArgs interface

+ Kind func() string

+ type JobArgsWithInsertOpts interface

+ InsertOpts func() InsertOpts

+ type JobFailedError struct

+ Job *JobRow

+ func (e *JobFailedError) Error() string

+ type JobFilter struct

+ After JobID

+ Kinds []string

+ Queue string

+ States []JobState

+ type JobID = driver.JobID

+ func ParseJobID(s string) (JobID, error)

+ type JobRow = driver.JobRow

+ type JobState = driver.JobState

+ type Message struct

+ Headers map[string]string

+ ID string

+ OrderingKey string

+ Payload T

+ Position StreamPosition

+ Stream string

+ Topic string

+ type Middleware interface

+ Insert func(ctx context.Context, params []InsertParams, ...) ([]*InsertResult, error)

+ Work func(ctx context.Context, job *JobRow, next func(ctx context.Context) error) error

+ func InsertMiddleware(...) Middleware

+ func WorkMiddleware(...) Middleware

+ type MiddlewareDefaults struct

+ func (MiddlewareDefaults) Insert(ctx context.Context, params []InsertParams, ...) ([]*InsertResult, error)

+ func (MiddlewareDefaults) Work(ctx context.Context, _ *JobRow, next func(ctx context.Context) error) error

+ type PanicError struct

+ Stack string

+ Value any

+ func (e *PanicError) Error() string

+ type PeriodicJob struct

+ Args JobArgs

+ Name string

+ Opts *PeriodicOpts

+ Schedule Schedule

+ func Cron(spec string, args JobArgs, opts *PeriodicOpts) PeriodicJob

+ func Every(interval time.Duration, args JobArgs, opts *PeriodicOpts) PeriodicJob

+ type PeriodicOpts struct

+ CatchUp int

+ InsertOpts *InsertOpts

+ Location *time.Location

+ Name string

+ RunOnStart bool

+ type Priority int

+ const PriorityHigh

+ const PriorityLow

+ const PriorityLowest

+ const PriorityNormal

+ type PublishOpts struct

+ Await bool

+ DedupKey string

+ Delay time.Duration

+ Headers map[string]string

+ OrderingKey string

+ Priority Priority

+ TTL time.Duration

+ type PublishResult struct

+ Deliveries []*InsertResult

+ MessageID string

+ type QueueConfig struct

+ DeleteCompleted bool

+ GlobalLimit int

+ MaxWorkers int

+ PartitionLimit int

+ PriorityAging time.Duration

+ RateBurst int

+ RateLimit float64

+ type QueueControl struct

+ func (q *QueueControl[TTx]) Add(ctx context.Context, name string, cfg QueueConfig) error

+ func (q *QueueControl[TTx]) List(ctx context.Context) ([]*QueueRow, error)

+ func (q *QueueControl[TTx]) Pause(ctx context.Context, name string) error

+ func (q *QueueControl[TTx]) Remove(ctx context.Context, name string) error

+ func (q *QueueControl[TTx]) Resume(ctx context.Context, name string) error

+ func (q *QueueControl[TTx]) SetLimits(ctx context.Context, name string, limits QueueLimits) error

+ type QueueLimits = driver.QueueLimits

+ type QueueRow = driver.QueueRow

+ type QueueStats = driver.QueueStats

+ type Raw = json.RawMessage

+ type Schedule interface

+ Next func(t time.Time) time.Time

+ type SeekOpts struct

+ Earliest bool

+ Latest bool

+ Position StreamPosition

+ Time time.Time

+ type SnoozeError struct

+ Duration time.Duration

+ func (e *SnoozeError) Error() string

+ type Stats = driver.Stats

+ type Step struct

+ type StepOpts struct

+ After []Step

+ Insert *InsertOpts

+ OnDependencyFailure DependencyFailure

+ func After(steps ...Step) *StepOpts

+ type StreamEvent = driver.StreamEvent

+ type StreamPosition = driver.StreamPosition

+ func ParseStreamPosition(s string) (StreamPosition, error)

+ type StreamStart = driver.StreamStart

+ type Streams struct

+ func (s *Streams[TTx]) Append(ctx context.Context, event Topic, opts *AppendOpts) (*StreamEvent, error)

+ func (s *Streams[TTx]) AppendTx(ctx context.Context, tx TTx, event Topic, opts *AppendOpts) (*StreamEvent, error)

+ func (s *Streams[TTx]) Consumers(ctx context.Context) ([]*ConsumerRow, error)

+ func (s *Streams[TTx]) Read(ctx context.Context, pattern string, after StreamPosition, limit int) iter.Seq2[*StreamEvent, error]

+ func (s *Streams[TTx]) Seek(ctx context.Context, consumer string, opts SeekOpts) error

+ type Subscription struct

+ MaxAttempts int

+ Metadata json.RawMessage

+ Name string

+ Pattern string

+ Queue string

+ func (s Subscription) Kind() string

+ type SubscriptionRow = driver.SubscriptionRow

+ type Topic interface

+ Topic func() string

+ type UniqueConflict int

+ const UniqueReplace

+ const UniqueSkip

+ type UniqueOpts struct

+ ByArgs bool

+ ByPeriod time.Duration

+ ByQueue bool

+ OnConflict UniqueConflict

+ type UnknownKindError struct

+ Kind string

+ func (e *UnknownKindError) Error() string

+ type Worker interface

+ NextRetry func(job *Job[T]) time.Time

+ Timeout func(job *Job[T]) time.Duration

+ Work func(ctx context.Context, job *Job[T]) error

+ type WorkerDefaults struct

+ func (WorkerDefaults[T]) NextRetry(*Job[T]) time.Time

+ func (WorkerDefaults[T]) Timeout(*Job[T]) time.Duration

+ type Workers struct

+ func NewWorkers() *Workers

+ func (w *Workers) Work(ctx context.Context, job *JobRow, codec Codec) error

+ type Workflow struct

+ func NewWorkflow(name string, opts *WorkflowOpts) *Workflow

+ func (w *Workflow) Add(args JobArgs, opts *StepOpts) Step

+ type WorkflowOpts struct

+ CallbackOpts *InsertOpts

+ Metadata json.RawMessage

+ OnComplete JobArgs

+ OnFailure JobArgs

+ OnSuccess JobArgs

+ type WorkflowResult struct

+ ID JobID

+ Jobs []*InsertResult

+ type WorkflowRow = driver.WorkflowRow

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL