Example: Task Queue

This is the canonical "process a queue" example. It registers one task, starts the runner, dispatches a batch of messages, lets them process, and shuts down cleanly. The full runnable source lives at examples/simple.

package main

import (
    "context"
    "fmt"
    "strconv"
    "sync"
    "time"

    "github.com/redis/go-redis/v9"
    "github.com/soroosh-tanzadeh/taskrunner/redisstream"
    "github.com/soroosh-tanzadeh/taskrunner/runner"
)

func main() {
    rdb := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379", DB: 0, PoolSize: 50})
    queue := redisstream.NewRedisStreamMessageQueueWithOptions(
        rdb,
        redisstream.WithPrefix("example"),
        redisstream.WithQueue("tasks"),
        redisstream.WithReClaimDelay(30*time.Second),
        redisstream.WithDeleteOnAck(true),
    )

    tr := runner.NewTaskRunner(runner.TaskRunnerConfig{
        BatchSize:          10,
        ConsumerGroup:      "example",
        ConsumersPrefix:    "default",
        NumWorkers:         8,
        NumFetchers:        4,
        LongQueueHook:      func(s runner.Stats) { fmt.Printf("%+v\n", s) },
        LongQueueThreshold: 30 * time.Second,
    }, rdb, queue)

    tr.RegisterTask(&runner.Task{
        Name:     "exampletask",
        MaxRetry: 5,
        Action: func(ctx context.Context, payload any) error {
            fmt.Printf("Hello from example task %v\n", payload)
            return nil
        },
    })

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    wg := sync.WaitGroup{}
    wg.Add(1)
    go func() { defer wg.Done(); _ = tr.Start(ctx) }()

    for i := 0; i < 10; i++ {
        _ = tr.Dispatch(context.Background(), "exampletask", strconv.Itoa(i))
    }

    time.Sleep(2 * time.Second)
    cancel()
    wg.Wait()
}

Walkthrough

1. Build the queue

queue := redisstream.NewRedisStreamMessageQueueWithOptions(
    rdb,
    redisstream.WithPrefix("example"),        // stream -> "example:tasks"
    redisstream.WithQueue("tasks"),
    redisstream.WithReClaimDelay(30*time.Second),
    redisstream.WithDeleteOnAck(true),
)

deleteOnAck: true keeps the stream from growing unbounded. reClaimDelay: 30s is the visibility timeout -- a message unacked for 30s becomes eligible for reclaim by another consumer.

2. Build the runner

tr := runner.NewTaskRunner(runner.TaskRunnerConfig{...}, rdb, queue)

NewTaskRunner does no I/O. The first Redis call happens when Start runs.

3. Register the task

tr.RegisterTask(&runner.Task{
    Name:     "exampletask",
    MaxRetry: 5,
    Action:   func(ctx context.Context, payload any) error { ... },
})

The registry is in-memory and keyed by Name. Lookups happen on every Dispatch and every worker invocation.

4. Start the runner

ctx, cancel := context.WithCancel(context.Background())
go func() { _ = tr.Start(ctx) }()

Start blocks until ctx is canceled, so run it in a goroutine. It spins up fetchers, the worker pool, election, and the timing aggregator.

5. Dispatch

for i := 0; i < 10; i++ {
    _ = tr.Dispatch(context.Background(), "exampletask", strconv.Itoa(i))
}

Dispatch is synchronous: it XADDs and returns. Payloads are JSON-marshalled, so pass any serializable value.

6. Shut down

time.Sleep(2 * time.Second)
cancel()
wg.Wait()

Canceling the context triggers a graceful drain: fetchers stop, the worker pool drains for up to 30s, the bulk writer flushes, the error channel closes.

Run it

# Requires a Redis on 127.0.0.1:6379
go run examples/simple/main.go

The accompanying test (examples/simple/simple_test.go) uses miniredis, so it runs with no real Redis:

go test ./examples/simple/...

Variations

• Multiple instances: run the binary twice; Redis Streams spreads work across both automatically.
• Real work: replace the fmt.Printf action with your business logic -- HTTP call, DB write, file processing. Remember to keep it idempotent.
• Errors: return an error from Action to trigger a retry; set FailedTaskHandler on the config to handle terminal failures.