Documentation
¶
Index ¶
- func FlushFailedJobs()
- func RecordFailedJob(job Job, err error)
- func RetryFailedJob(id int64) error
- type BatchBuilder
- type Bus
- type ChainBuilder
- type ChainableJob
- type DatabaseDriver
- type Driver
- type FailedJob
- type FailedJobStore
- type Fake
- func (f *Fake) AssertNotPushed() bool
- func (f *Fake) AssertNotPushedJob(jobType any) bool
- func (f *Fake) AssertPushed() bool
- func (f *Fake) AssertPushedCount(count int) bool
- func (f *Fake) AssertPushedJob(jobType any) bool
- func (f *Fake) Clear()
- func (f *Fake) GetJobCount() int
- func (f *Fake) GetJobs() []Job
- func (f *Fake) GetLastJob() Job
- func (f *Fake) HasJob(jobType any) bool
- func (f *Fake) Push(job Job) error
- func (f *Fake) Size() int
- func (f *Fake) Sync() error
- func (f *Fake) SyncJobs(jobType any) error
- type HasBackoff
- type HasMaxAttempts
- type HasMaxRetries
- type HasPriority
- type HasTimeout
- type InMemoryUniqueStore
- type Job
- type JobBatch
- type JobMiddleware
- func EnsureUnique(store UniqueJobStore) JobMiddleware
- func LoggingMiddleware() JobMiddleware
- func RateLimitByKeyMiddleware(keyFunc func(Job) string, maxJobs int, per time.Duration) JobMiddleware
- func RateLimitMiddleware(maxJobs int, per time.Duration) JobMiddleware
- func RetryMiddleware(maxRetries int) JobMiddleware
- func TimeoutMiddleware(timeout time.Duration) JobMiddleware
- func WithoutRetry() JobMiddleware
- type JobWithMiddleware
- type Manager
- type MemoryDriver
- type MiddlewareFunc
- type RabbitMQDriver
- type RedisDriver
- type SQSDriver
- type ShouldBeUnique
- type ShouldBeUniqueUntilProcessing
- type SyncDriver
- type UniqueJobStore
- type Worker
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func RecordFailedJob ¶
RecordFailedJob stores a failed job.
func RetryFailedJob ¶
RetryFailedJob retries a failed job by ID.
Types ¶
type BatchBuilder ¶
type BatchBuilder struct {
// contains filtered or unexported fields
}
BatchBuilder configures a batch of jobs.
func (*BatchBuilder) Catch ¶
func (bb *BatchBuilder) Catch(fn func(*JobBatch, error)) *BatchBuilder
Catch sets the callback for when a job fails.
func (*BatchBuilder) Dispatch ¶
func (bb *BatchBuilder) Dispatch() (*JobBatch, error)
Dispatch dispatches all jobs in the batch.
func (*BatchBuilder) Then ¶
func (bb *BatchBuilder) Then(fn func(*JobBatch)) *BatchBuilder
Then sets the callback for when all jobs complete.
type Bus ¶
type Bus struct {
// contains filtered or unexported fields
}
Bus dispatches jobs and manages batches.
func (*Bus) Batch ¶
func (b *Bus) Batch(jobs []Job) *BatchBuilder
Batch creates a batch of jobs and dispatches them.
func (*Bus) Chain ¶
func (b *Bus) Chain(jobs []Job) *ChainBuilder
Chain dispatches jobs sequentially.
type ChainBuilder ¶
type ChainBuilder struct {
// contains filtered or unexported fields
}
ChainBuilder configures sequential job execution.
func (*ChainBuilder) Catch ¶
func (cb *ChainBuilder) Catch(fn func(error)) *ChainBuilder
Catch sets the callback for when a job fails.
func (*ChainBuilder) Dispatch ¶
func (cb *ChainBuilder) Dispatch() error
Dispatch dispatches jobs sequentially.
func (*ChainBuilder) Then ¶
func (cb *ChainBuilder) Then(fn func()) *ChainBuilder
Then sets the callback for when all jobs complete.
type ChainableJob ¶
type ChainableJob interface {
// Chain returns the middleware chain for this job type.
Chain() []JobMiddleware
}
ChainableJob allows a job to define middleware via a static method.
type DatabaseDriver ¶
type DatabaseDriver struct {
// contains filtered or unexported fields
}
DatabaseDriver is a queue driver backed by a SQL database.
func NewDatabaseDriver ¶
func NewDatabaseDriver(db *orm.DB, defaultQueueName string) *DatabaseDriver
NewDatabaseDriver creates a new Database queue driver.
func (*DatabaseDriver) Pop ¶
func (d *DatabaseDriver) Pop() (Job, error)
Pop retrieves and reserves a job from the jobs table using a transaction to prevent race conditions between concurrent workers.
func (*DatabaseDriver) Push ¶
func (d *DatabaseDriver) Push(job Job) error
Push pushes a job into the jobs table.
type FailedJobStore ¶
type FailedJobStore struct {
// contains filtered or unexported fields
}
FailedJobStore manages failed jobs.
type Fake ¶
type Fake struct {
// contains filtered or unexported fields
}
Fake is a fake queue dispatcher for testing.
func (*Fake) AssertNotPushed ¶
AssertNotPushed asserts that no jobs were pushed.
func (*Fake) AssertNotPushedJob ¶
AssertNotPushedJob asserts that a specific job type was NOT pushed.
func (*Fake) AssertPushed ¶
AssertPushed asserts that at least one job was pushed.
func (*Fake) AssertPushedCount ¶
AssertPushedCount asserts the exact number of jobs pushed.
func (*Fake) AssertPushedJob ¶
AssertPushedJob asserts that a specific job type was pushed.
func (*Fake) GetJobCount ¶
GetJobCount returns the number of jobs queued.
func (*Fake) GetLastJob ¶
GetLastJob returns the last captured job, or nil if none.
type HasBackoff ¶
HasBackoff allows a job to specify its own backoff strategy.
type HasMaxAttempts ¶
type HasMaxAttempts interface {
MaxAttempts() int
}
HasMaxAttempts allows a job to specify a hard limit on total attempts.
type HasMaxRetries ¶
type HasMaxRetries interface {
MaxRetries() int
}
HasMaxRetries allows a job to specify its own max retry count.
type HasPriority ¶
type HasPriority interface {
Priority() int
}
HasPriority allows a job to specify its priority (0 = highest, 10 = lowest). Jobs with lower priority numbers are processed first.
type HasTimeout ¶
HasTimeout allows a job to specify its own timeout.
type InMemoryUniqueStore ¶
type InMemoryUniqueStore struct {
// contains filtered or unexported fields
}
InMemoryUniqueStore is an in-memory implementation of UniqueJobStore.
func NewInMemoryUniqueStore ¶
func NewInMemoryUniqueStore() *InMemoryUniqueStore
func (*InMemoryUniqueStore) Acquire ¶
func (s *InMemoryUniqueStore) Acquire(key string) bool
func (*InMemoryUniqueStore) Release ¶
func (s *InMemoryUniqueStore) Release(key string)
type Job ¶
type Job interface {
// Handle executes the job logic.
Handle() error
// Failed handles job failure after all retries are exhausted.
Failed(err error)
}
Job defines the interface for an executable queue job.
type JobBatch ¶
type JobBatch struct {
ID int64
Jobs []Job
Pending int
Completed int
Failed int
Then func(batch *JobBatch) // callback when all jobs complete
Catch func(batch *JobBatch, err error) // callback on failure
}
JobBatch represents a batch of jobs.
type JobMiddleware ¶
JobMiddleware wraps job execution with custom logic.
func EnsureUnique ¶
func EnsureUnique(store UniqueJobStore) JobMiddleware
EnsureUnique ensures only one instance of a job with the same UniqueID runs at a time.
func LoggingMiddleware ¶
func LoggingMiddleware() JobMiddleware
LoggingMiddleware logs job execution.
func RateLimitByKeyMiddleware ¶
func RateLimitByKeyMiddleware(keyFunc func(Job) string, maxJobs int, per time.Duration) JobMiddleware
RateLimitByKeyMiddleware limits jobs by a key extracted from the job. Useful for per-user or per-tenant rate limiting.
func RateLimitMiddleware ¶
func RateLimitMiddleware(maxJobs int, per time.Duration) JobMiddleware
RateLimitMiddleware limits job execution to N jobs per time period. Uses a token bucket algorithm.
func RetryMiddleware ¶
func RetryMiddleware(maxRetries int) JobMiddleware
RetryMiddleware adds retry logic with exponential backoff.
func TimeoutMiddleware ¶
func TimeoutMiddleware(timeout time.Duration) JobMiddleware
TimeoutMiddleware adds a timeout to job execution.
type JobWithMiddleware ¶
type JobWithMiddleware interface {
Middleware() []JobMiddleware
}
JobWithMiddleware allows a job to declare its own middleware.
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
Manager resolves queue drivers.
func NewManager ¶
NewManager initializes a new Queue Manager.
func (*Manager) Connection ¶
Connection gets a driver by name.
type MemoryDriver ¶
type MemoryDriver struct {
// contains filtered or unexported fields
}
MemoryDriver is a goroutine-safe queue driver backed by a buffered channel.
func NewMemoryDriver ¶
func NewMemoryDriver(capacity int) *MemoryDriver
NewMemoryDriver initializes a new MemoryDriver with a specified buffer capacity.
func (*MemoryDriver) Len ¶
func (d *MemoryDriver) Len() int
Len returns the current number of jobs waiting in the queue.
func (*MemoryDriver) Pop ¶
func (d *MemoryDriver) Pop() (Job, error)
Pop blocks until a job is available and returns it.
func (*MemoryDriver) Push ¶
func (d *MemoryDriver) Push(job Job) error
Push adds a new job to the channel. Returns an error if the queue is full after 5 seconds.
func (*MemoryDriver) TryPop ¶
func (d *MemoryDriver) TryPop() Job
TryPop attempts to read a job from the channel without blocking. Returns nil if the queue is empty.
type MiddlewareFunc ¶
MiddlewareFunc is a function adapter for JobMiddleware.
type RabbitMQDriver ¶
type RabbitMQDriver struct {
ConnectionString string
QueueName string
Exchange string
// contains filtered or unexported fields
}
RabbitMQDriver implements the Driver interface for RabbitMQ
func NewRabbitMQDriver ¶
func NewRabbitMQDriver(connectionString, queueName, exchange string) *RabbitMQDriver
NewRabbitMQDriver creates a new RabbitMQ driver
func (*RabbitMQDriver) Pop ¶
func (d *RabbitMQDriver) Pop() (Job, error)
Pop pops a job from RabbitMQ
func (*RabbitMQDriver) Push ¶
func (d *RabbitMQDriver) Push(job Job) error
Push pushes a job to RabbitMQ
type RedisDriver ¶
type RedisDriver struct {
// contains filtered or unexported fields
}
RedisDriver is a queue driver backed by Redis.
func NewRedisDriver ¶
func NewRedisDriver(client *redis.Client, defaultQueueName string) *RedisDriver
NewRedisDriver creates a new Redis queue driver.
func (*RedisDriver) Pop ¶
func (d *RedisDriver) Pop() (Job, error)
Pop pops a job off the Redis queue list.
func (*RedisDriver) Push ¶
func (d *RedisDriver) Push(job Job) error
Push pushes a job onto the Redis queue list.
type SQSDriver ¶
type SQSDriver struct {
QueueURL string
Region string
AccessKey string
SecretKey string
Endpoint string // Optional custom endpoint (e.g., for LocalStack)
// contains filtered or unexported fields
}
SQSDriver implements the Driver interface for Amazon SQS
func NewSQSDriver ¶
NewSQSDriver creates a new SQS driver
type ShouldBeUnique ¶
type ShouldBeUnique interface {
UniqueID() string
}
ShouldBeUnique interface for jobs that must be unique in the queue. If a job with the same UniqueID is already pending, it will not be dispatched.
type ShouldBeUniqueUntilProcessing ¶
type ShouldBeUniqueUntilProcessing interface {
ShouldBeUnique
UniqueVia() string // optional: custom lock key prefix
}
ShouldBeUniqueUntilProcessing is like ShouldBeUnique but releases the lock once the job begins processing (not after completion).
type SyncDriver ¶
type SyncDriver struct {
// contains filtered or unexported fields
}
SyncDriver is an in-memory channel-based queue driver for local development.
func NewSyncDriver ¶
func NewSyncDriver(bufferSize int) *SyncDriver
NewSyncDriver creates a new SyncDriver with the given buffer size.
func (*SyncDriver) Channel ¶
func (d *SyncDriver) Channel() <-chan Job
Channel returns the underlying Go channel.
func (*SyncDriver) Pop ¶
func (d *SyncDriver) Pop() (Job, error)
Pop retrieves the next job from the channel (blocking). In a real channel based worker, the worker just ranges over the channel.
type UniqueJobStore ¶
UniqueJobStore is an interface for storing job uniqueness locks.
Source Files