queue

package module
v0.1.1 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 9, 2026 License: MIT Imports: 12 Imported by: 1

README

Queue — Lemmego Tasker Plugin

queue is a standalone, self-contained module that wraps tasker as a lemmego-compatible provider. It implements Provider, CommandProvider, RouteProvider, PublishableProvider, and ShutdownProvider interfaces.

Usage

In bootstrap/providers.go
import "github.com/lemmego/queue"

func LoadProviders() []app.Provider {
    return []app.Provider{
        &queue.Provider{},  // auto-configures from app config
    }
}

That's it. The provider auto-reads database credentials from your existing configs/database.go (same sql.connections.<default> that gormconnector uses). No hardcoded DSN needed.

With explicit configuration
&queue.Provider{
    Config: &queue.Config{
        Driver:             "redis",
        RedisAddr:          ":6379",
        RoutePrefix:        "/background-jobs",  // default: "/tasker"
        DefaultQueue:       "default",
        DefaultMaxAttempts: 3,
        Workers:            map[string]int{"default": 3, "email": 2},
        EnableAutoscale:    false,
    },
}

Configuration Priority

1. Explicit Config fields (highest)
2. App config (published tasker.go + env vars)
3. Sensible defaults (lowest)
.env overrides
TASKER_ROUTE_PREFIX=/jobs
TASKER_TABLE_PREFIX=myapp_
TASKER_QUEUE=default
TASKER_MAX_ATTEMPTS=3
TASKER_HEARTBEAT_SEC=5
TASKER_REQUEUE_SEC=60
TASKER_PRUNE_HOURS=168
TASKER_AUTOSCALE=false
When Config is nil (auto-mode)
  • Reads sql.connections.<default> from app config (same source as gormconnector)
  • Picks up driver, database, host, port, user, password
  • Works with SQLite, MySQL, PostgreSQL out of the box
  • Uses tasker_ table prefix, /tasker route prefix
When Driver is "redis"

Skips SQL config entirely — use RedisAddr, RedisPass, RedisDB fields.

What Gets Created

Always (HTTP + Console)
  • Driver — database connection (SQL or Redis)
  • Migration — tables auto-created (CREATE TABLE IF NOT EXISTS)
  • Manager — job dispatcher, registered as tasker.Global()
HTTP only (skipped in console mode)
  • Supervisor — worker pool manager (needed by tasker:work)
  • Web server — dashboard UI + API at RoutePrefix

Commands

Registered automatically via CommandProvider:

lemmego run tasker:work --queue=default --workers=3

Routes

Mounted automatically via RouteProvider:

Prefix Description
GET /tasker/ Dashboard UI
GET /tasker/api/* JSON API
POST /tasker/api/* Actions (retry, cancel, pause, resume, prune)

Customize with Config.RoutePrefix:

Config: &Config{RoutePrefix: "/jobs"}  // → GET /jobs/api/stats

Publishing Config

The provider publishes internal/configs/tasker.go via AddPublishables():

lemmego run publish --tags=config

This creates a config file with env-var-driven defaults for all tasker settings.

Dependencies

  • github.com/lemmego/tasker — core job system
  • github.com/lemmego/api — lemmego framework interfaces
  • github.com/spf13/cobra — CLI commands

No GPA dependency. The driver manages its own connection pool directly.

License

MIT

Documentation

Index

Constants

View Source
const ConfigStub = `` /* 653-byte string literal not displayed */

Variables

This section is empty.

Functions

func AddGlobalMiddleware

func AddGlobalMiddleware(mw tasker.JobMiddleware)

func Cancel

func Cancel(ctx context.Context, id tasker.JobID) (*tasker.JobRow, error)

func Chain

func Chain(ctx context.Context, jobs []tasker.Job, opts ...tasker.DispatchOpt) ([]*tasker.JobRow, error)

func Dispatch

func Dispatch(ctx context.Context, job tasker.Job, opts ...tasker.DispatchOpt) (*tasker.JobRow, error)

func DispatchBatch

func DispatchBatch(ctx context.Context, jobs []tasker.Job, opts ...tasker.DispatchOpt) ([]*tasker.JobRow, error)

func GetJob

func GetJob(ctx context.Context, id tasker.JobID) (*tasker.JobRow, error)

func RegisterJob

func RegisterJob(name string, factory tasker.JobFactory)

func Retry

func Retry(ctx context.Context, id tasker.JobID) (*tasker.JobRow, error)

Types

type Config

type Config struct {
	Driver       string
	DSN          string
	DriverName   string
	TablePrefix  string
	RoutePrefix  string
	MaxOpenConns int
	MaxIdleConns int

	RedisAddr   string
	RedisPass   string
	RedisDB     int
	RedisPrefix string

	DefaultQueue       string
	DefaultMaxAttempts int

	Workers         map[string]int
	EnableAutoscale bool

	HeartbeatInterval int
	RequeueTimeout    int
	PruneAfterHours   int
}

func DefaultConfig

func DefaultConfig() *Config

func (*Config) ApplyOverrides

func (cfg *Config) ApplyOverrides(c config.M)

ApplyOverrides reads from a config.M (the "tasker" section) and overrides matching fields on the Config. Keys are flat (no "tasker." prefix).

func (*Config) AutoFillDatabase

func (cfg *Config) AutoFillDatabase(c config.M)

AutoFillDatabase reads the app's default SQL connection config and fills DSN/DriverName. Only applies when Driver is "sql" and DSN is empty.

type Provider added in v0.1.1

type Provider struct {
	Config *Config
}

func (*Provider) AddCommands added in v0.1.1

func (p *Provider) AddCommands() []app.Command

func (*Provider) AddPublishables added in v0.1.1

func (p *Provider) AddPublishables() []*app.Publishable

func (*Provider) AddRoutes added in v0.1.1

func (p *Provider) AddRoutes() app.RouteCallback

func (*Provider) Provide added in v0.1.1

func (p *Provider) Provide(a app.App) error

func (*Provider) Shutdown added in v0.1.1

func (p *Provider) Shutdown(ctx context.Context) error

Jump to

Keyboard shortcuts

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