Help us shape the future of Go! Take the 2026 Go Developer Survey

gopherqueue

package module
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Aug 6, 2026 License: MIT Imports: 8 Imported by: 0

README

🚀 GopherQueue v0.1.0

Go Reference Release

GopherQueue v0.1.0 is an open-source, high-performance distributed task processing library and background job engine for Go. Engineered for both embedded Go applications and standalone microservice daemons, GopherQueue features at-least-once atomic Redis delivery (BLMOVE), first-class rate-limiting synergy with distlimit, an HTTP-style middleware pipeline, and zero-dependency Server-Sent Events (SSE) telemetry paired with a Vue.js web dashboard.

Table of Contents

⚡ Unique Value Propositions (UVP)

  • 🔄 Dual-Mode Deployment: Import directly as an idiomatic Go package (embedded mode) or run as a standalone worker/dashboard daemon (standalone mode).
  • 🛡️ First-Class Rate Limiting Synergy (distlimit): Native integration adapter with github.com/balramadan/distlimit. Rate-limited jobs are gracefully rescheduled to the Delayed Queue (RATE_LIMITED / DEFERRED) without consuming retry budgets.
  • ⚡ At-Least-Once Atomic Delivery: Powered by atomic Redis BLMOVE operations and crash reclaimers to prevent job loss during worker crashes.
  • 🔗 HTTP-Style Middleware Pipeline: Expressive Chi-like middleware chaining (Use(...)) supporting structured logging (slog), panic recovery, and rate limiting.
  • 📊 Real-time SSE Telemetry: Built-in Server-Sent Events streamer broadcasting live queue counters (pending, processing, delayed, dlq) to a Vue.js web dashboard.

📊 Competitive Matrix

Feature / Capability Asynq River Machinery GopherQueue
Primary Broker Redis PostgreSQL Redis / AMQP / SQS Redis (Pluggable Storage)
Deployment Mode Embedded Only Embedded Only Embedded Only Dual-Mode (Embedded & Standalone Microservice)
Middleware Pipeline Basic Basic None Full HTTP-Style Pipeline (Use(...))
Distributed Rate Limiting Basic Key/Limit Basic None Native distlimit Integration Adapter
Real-time Telemetry Separate Web UI Postgres Views None Embedded SSE Streamer + Built-in Vue Dashboard

📦 Installation

go get github.com/balramadan/gopherqueue

⚡ Quickstart

1. Embedded Producer Client & Worker Server
package main

import (
	"context"
	"fmt"
	"log"

	"github.com/balramadan/gopherqueue"
	gopherqredis "github.com/balramadan/gopherqueue/storage/redis"
	"github.com/redis/go-redis/v9"
)

func main() {
	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
	redisStore := gopherqredis.NewStore(rdb)

	// 1. Initialize Producer Client
	client, _ := gopherqueue.NewClient(
		gopherqueue.WithStorage(redisStore),
		gopherqueue.WithQueue("default"),
	)

	// 2. Initialize Worker Server
	server := gopherqueue.NewServer(
		gopherqueue.WithStorage(redisStore),
		gopherqueue.WithQueue("default"),
		gopherqueue.WithConcurrency(10),
	)

	// 3. Register Job Handler
	server.Register("email:send", func(ctx context.Context, job *gopherqueue.Job) error {
		to, _ := job.Payload["to"].(string)
		fmt.Printf("Sending email to %s (Job ID: %s)\n", to, job.ID)
		return nil
	})

	// 4. Enqueue Job
	job, _ := gopherqueue.NewJob("job-101", "default", "email:send", map[string]any{
		"to": "user@example.com",
	})
	_ = client.Enqueue(context.Background(), job)

	// 5. Run Server Engine
	if err := server.Run(context.Background()); err != nil {
		log.Fatalf("Server error: %v", err)
	}
}
2. Integration with distlimit Rate Limiter

Protect third-party APIs (OpenAI, Payment Gateways, SMS Services) from rate-limit exhaustion:

package main

import (
	"context"
	"time"

	"github.com/balramadan/distlimit"
	"github.com/balramadan/distlimit/algorithm/tokenbucket"
	distlimitredis "github.com/balramadan/distlimit/driver/redis"
	"github.com/balramadan/gopherqueue"
	gopherqdist "github.com/balramadan/gopherqueue/middleware/distlimit"
	gopherqredis "github.com/balramadan/gopherqueue/storage/redis"
	"github.com/redis/go-redis/v9"
)

func main() {
	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
	redisStore := gopherqredis.NewStore(rdb)

	// Initialize distlimit Rate Limiter (Allow 100 requests per minute)
	limiter, _ := distlimit.New(
		distlimitredis.New(rdb),
		distlimit.WithLimit(100),
		distlimit.WithWindow(1*time.Minute),
		distlimit.WithAlgorithm(tokenbucket.New()),
	)

	server := gopherqueue.NewServer(
		gopherqueue.WithStorage(redisStore),
		gopherqueue.WithConcurrency(10),
	)

	// Attach distlimit Execution Guard Middleware
	server.Use(gopherqdist.New(limiter, func(job *gopherqueue.Job) string {
		return "rate_limit:" + job.Type
	}))

	server.Register("api:call", func(ctx context.Context, job *gopherqueue.Job) error {
		// Handler logic
		return nil
	})

	_ = server.Run(context.Background())
}

🛠️ Middleware Ecosystem

GopherQueue provides built-in middlewares under middleware/:

  • middleware/distlimit: Execution guard adapter for github.com/balramadan/distlimit.
  • middleware/logger: Structured logging using Go standard log/slog.
  • middleware/recover: Panic recovery guard preventing worker goroutine crashes.
import (
	gopherqlogger "github.com/balramadan/gopherqueue/middleware/logger"
	gopherqrecover "github.com/balramadan/gopherqueue/middleware/recover"
)

server.Use(gopherqlogger.New(slog.Default()))
server.Use(gopherqrecover.New())

🖥️ Zero-Dependency Web Dashboard GUI

GopherQueue features a modern, dark-mode Vue 3 Web Dashboard GUI embedded directly into the Go binary using Go //go:embed (web.DistFS()).

Option 1: Standalone Single-Binary Execution (Zero Setup)

No Node.js, npm, or pnpm required! Developers can launch the Web Dashboard GUI with a single command:

# 1. Run directly using Go CLI from local workspace
go run ./cmd/gopherq-dashboard

# 2. Or run directly from GitHub
go run github.com/balramadan/gopherqueue/cmd/gopherq-dashboard@latest

# 3. Or install the binary globally to your PATH
go install github.com/balramadan/gopherqueue/cmd/gopherq-dashboard@latest
gopherq-dashboard

Open http://localhost:8080 in your browser to inspect real-time metrics (Pending, Processing, Delayed, DLQ), active SSE connection status, and job payloads.

Option 2: Embedded Route in your Go Application

Mount the embedded Vue Dashboard static assets directly inside any Go HTTP router (Gin, Chi, Echo, or standard net/http):

package main

import (
	"net/http"

	gopherqredis "github.com/balramadan/gopherqueue/storage/redis"
	"github.com/balramadan/gopherqueue/telemetry"
	"github.com/balramadan/gopherqueue/web"
	"github.com/redis/go-redis/v9"
)

func main() {
	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
	store := gopherqredis.NewStore(rdb)

	sseBroadcaster := telemetry.NewSSEBroadcaster(store, "default")
	http.HandleFunc("/api/v1/stream", sseBroadcaster.StreamMetrics)

	// Serve the Embedded Vue Web Dashboard GUI
	distFS, _ := web.DistFS()
	http.Handle("/dashboard/", http.StripPrefix("/dashboard/", http.FileServer(http.FS(distFS))))

	http.ListenAndServe(":8080", nil)
}

🤝 Contributing

We welcome contributions from the open-source community! Please review our CONTRIBUTING.md guide for guidelines on submitting issues, feature requests, and pull requests.

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

Documentation

Overview

Package gopherqueue is an open-source, high-performance distributed task processing library and background job engine for Go.

GopherQueue supports Dual-Mode deployment: 1. Embedded Mode: Imported directly as a Go package (`go get github.com/balramadan/gopherqueue`) inside monolith or microservice Go applications. 2. Standalone Mode: Executed as a standalone worker daemon (`cmd/gopherq-server`) and web dashboard (`cmd/gopherq-dashboard`).

Key Features:

  • At-Least-Once Atomic Delivery: Powered by Redis BLMOVE operations to prevent job loss during worker crashes.
  • First-Class Rate Limiting Synergy: Native integration with distlimit (`github.com/balramadan/distlimit`). Rate-limited jobs are gracefully deferred to the delayed queue without consuming retry budgets.
  • HTTP-Style Middleware Pipeline: Expressive Chi-like middleware chaining (`Use(...)`) supporting structured logging (`slog`), panic recovery, and rate-limiting.
  • Zero-Dependency Telemetry: Server-Sent Events (SSE) broadcaster delivering real-time queue metrics to a Vue.js web dashboard.

Basic Usage: 

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/balramadan/gopherqueue"
	gopherqredis "github.com/balramadan/gopherqueue/storage/redis"
	"github.com/redis/go-redis/v9"
)

func main() {
	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
	store := gopherqredis.NewStore(rdb)

	server := gopherqueue.NewServer(
		gopherqueue.WithStorage(store),
		gopherqueue.WithQueue("default"),
		gopherqueue.WithConcurrency(5),
	)

	server.Register("email:send", func(ctx context.Context, job *gopherqueue.Job) error {
		fmt.Printf("Processing job %s\n", job.ID)
		return nil
	})

	if err := server.Run(context.Background()); err != nil {
		log.Fatalf("Server error: %v", err)
	}
}

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	// ErrJobNotFound indicates that the requested job ID does not exist in storage.
	ErrJobNotFound = errors.New("gopherqueue: job not found")
	// ErrQueueEmpty indicates that no pending job is available for fetching.
	ErrQueueEmpty = errors.New("gopherqueue: queue is empty")
	// ErrInvalidJobData indicates missing required fields when creating or processing a job.
	ErrInvalidJobData = errors.New("gopherqueue: invalid job data or metadata")
	// ErrServerClosed indicates operations were attempted on a closed server.
	ErrServerClosed = errors.New("gopherqueue: server is closed")
)

Functions

This section is empty.

Types

type Chain 

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

Chain manages an ordered slice of MiddlewareFunc and compiles them into a single Handler.

func NewChain 

func NewChain(middlewares ...MiddlewareFunc) Chain

NewChain creates a new Chain instance with the provided middlewares.

func (Chain) Append 

func (c Chain) Append(middlewares ...MiddlewareFunc) Chain

Append creates a new Chain by appending additional middlewares to the existing chain.

func (Chain) Then 

func (c Chain) Then(final Handler) Handler

Then links the middleware chain to a final Job Handler.

type Client 

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

Client provides a producer API for submitting jobs to GopherQueue.

func NewClient 

func NewClient(opts ...Option) (*Client, error)

NewClient initializes a new Client instance with functional options.

Example

ExampleNewClient demonstrates how to enqueue immediate and delayed jobs. 

store := &mockRootStorage{}

client, err := gopherqueue.NewClient(
	gopherqueue.WithStorage(store),
	gopherqueue.WithQueue("default"),
)
if err != nil {
	fmt.Printf("Failed to create client: %v\n", err)
	return
}

job, _ := gopherqueue.NewJob("job-101", "default", "email:send", map[string]any{
	"to": "user@example.com",
})

_ = client.Enqueue(context.Background(), job)
_ = client.EnqueueAt(context.Background(), job, time.Now().Add(5*time.Minute))

fmt.Println("Job enqueued successfully")

Output:
Job enqueued successfully

func (*Client) Enqueue 

func (c *Client) Enqueue(ctx context.Context, job *Job) error

Enqueue submits a job to the queue for immediate processing.

func (*Client) EnqueueAt 

func (c *Client) EnqueueAt(ctx context.Context, job *Job, executeAt time.Time) error

EnqueueAt schedules a job to be executed at a specific future timestamp.

type Config 

type Config struct {
	QueueName      string
	Concurrency    int
	PollInterval   time.Duration
	DefaultTimeout time.Duration
	Storage        Storage
}

Config holds configuration parameters for Server and Client instances.

type ErrRateLimited 

type ErrRateLimited struct {
	Key        string
	RetryAfter time.Duration
}

ErrRateLimited is a custom error returned by rate limiting middleware to signal that execution was rate limited and the job should be deferred to the delayed queue.

func IsRateLimited 

func IsRateLimited(err error) (*ErrRateLimited, bool)

IsRateLimited inspects an error chain and returns *ErrRateLimited if present.

func (*ErrRateLimited) Error 

func (e *ErrRateLimited) Error() string

Error returns a formatted error message.

type Handler 

type Handler interface {
	ProcessJob(ctx context.Context, job *Job) error
}

Handler defines the interface for executing a Job.

type HandlerFunc 

type HandlerFunc func(ctx context.Context, job *Job) error

HandlerFunc is an adapter type allowing ordinary functions to act as a Handler.

func (HandlerFunc) ProcessJob 

func (f HandlerFunc) ProcessJob(ctx context.Context, job *Job) error

ProcessJob calls the underlying HandlerFunc.

type HandlerRegistry 

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

HandlerRegistry maintains a thread-safe mapping of job types to JobHandlers.

func NewHandlerRegistry 

func NewHandlerRegistry() *HandlerRegistry

NewHandlerRegistry initializes a new HandlerRegistry instance.

func (*HandlerRegistry) Execute 

func (r *HandlerRegistry) Execute(ctx context.Context, job *Job) error

Execute looks up and invokes the registered handler for the job type.

func (*HandlerRegistry) Register 

func (r *HandlerRegistry) Register(jobType string, handler JobHandler)

Register registers a handler function for a given job type.

type Job 

type Job struct {
	ID             string         `json:"id"`
	Queue          string         `json:"queue"`
	Type           string         `json:"type"`
	Payload        map[string]any `json:"payload"`
	Status         JobStatus      `json:"status"`
	MaxRetries     int            `json:"max_retries"`
	CurrentRetry   int            `json:"current_retry"`
	TimeoutSeconds int            `json:"timeout_seconds"`
	ExecuteAt      int64          `json:"execute_at"`
	CreatedAt      int64          `json:"created_at"`
	UpdatedAt      int64          `json:"updated_at"`
	LastError      string         `json:"last_error,omitempty"`
}

Job represents a unit of work submitted to GopherQueue.

func NewJob 

func NewJob(id, queue, jobType string, payload map[string]any) (*Job, error)

NewJob creates a new Job instance with default configuration.

func UnmarshalJob 

func UnmarshalJob(data string) (*Job, error)

UnmarshalJob deserializes a JSON string into a Job instance.

func (*Job) CalculateNextRetryDelay 

func (j *Job) CalculateNextRetryDelay() time.Duration

CalculateNextRetryDelay calculates an exponential backoff delay based on CurrentRetry.

func (*Job) Marshal 

func (j *Job) Marshal() (string, error)

Marshal serializes the Job into a JSON string.

type JobExecutor 

type JobExecutor interface {
	Execute(ctx context.Context, job *Job) error
}

JobExecutor defines an interface for executing a job.

type JobHandler 

type JobHandler func(ctx context.Context, job *Job) error

JobHandler represents a handler function signature for processing a job.

type JobStatus 

type JobStatus string

JobStatus represents the state of a job in the queue lifecycle. 

const (
	// StatusEnqueued indicates the job has been submitted to the queue.
	StatusEnqueued JobStatus = "ENQUEUED"
	// StatusPending indicates the job is ready to be fetched by a worker.
	StatusPending JobStatus = "PENDING"
	// StatusDelayed indicates the job is scheduled to run at a future timestamp.
	StatusDelayed JobStatus = "DELAYED"
	// StatusProcessing indicates a worker is actively processing the job.
	StatusProcessing JobStatus = "PROCESSING"
	// StatusCompleted indicates the job finished successfully and was ACKed.
	StatusCompleted JobStatus = "COMPLETED"
	// StatusFailed indicates job execution failed.
	StatusFailed JobStatus = "FAILED"
	// StatusRetrying indicates the job failed and is scheduled for a retry.
	StatusRetrying JobStatus = "RETRYING"
	// StatusDeadLetter indicates the job exceeded its maximum retry budget and was moved to DLQ.
	StatusDeadLetter JobStatus = "DEAD_LETTER"
	// StatusRateLimited indicates job execution was rate-limited and deferred without incrementing retry count.
	StatusRateLimited JobStatus = "RATE_LIMITED"
)

type MiddlewareFunc 

type MiddlewareFunc func(next Handler) Handler

MiddlewareFunc defines a middleware function wrapper around a Handler.

type Option 

type Option func(*Config)

Option represents a functional option for configuring Server or Client.

func WithConcurrency 

func WithConcurrency(n int) Option

WithConcurrency sets the number of concurrent worker goroutines.

func WithPollInterval 

func WithPollInterval(d time.Duration) Option

WithPollInterval sets the interval between worker queue polling calls.

func WithQueue 

func WithQueue(name string) Option

WithQueue sets the queue name to listen to or push jobs into.

func WithStorage 

func WithStorage(s Storage) Option

WithStorage sets the storage engine implementation (e.g. redis.Store).

type Server 

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

Server represents the main GopherQueue worker engine orchestrator.

func NewServer 

func NewServer(opts ...Option) *Server

NewServer initializes a new Server instance with functional options.

Example

ExampleNewServer demonstrates how to initialize and run a GopherQueue worker server. 

store := &mockRootStorage{}

server := gopherqueue.NewServer(
	gopherqueue.WithStorage(store),
	gopherqueue.WithQueue("default"),
	gopherqueue.WithConcurrency(2),
)

server.Register("email:send", func(ctx context.Context, job *gopherqueue.Job) error {
	fmt.Printf("Executing job %s of type %s\n", job.ID, job.Type)
	return nil
})

// Server is ready to run with server.Run(ctx)
fmt.Println("Server initialized successfully")

Output:
Server initialized successfully

func (*Server) Register 

func (s *Server) Register(jobType string, fn func(ctx context.Context, job *Job) error)

Register registers a job handler function for a specific job type.

func (*Server) Run 

func (s *Server) Run(ctx context.Context) error

Run starts the worker pool and blocks until the context is cancelled or Shutdown is called.

func (*Server) Shutdown 

func (s *Server) Shutdown(ctx context.Context) error

Shutdown gracefully stops all active workers.

func (*Server) Use 

func (s *Server) Use(middlewares ...MiddlewareFunc)

Use appends one or more middlewares to the server's global middleware pipeline.

type Storage 

type Storage interface {
	Enqueue(ctx context.Context, job *Job) error
	Fetch(ctx context.Context, queue string, timeout time.Duration) (*Job, error)
	Ack(ctx context.Context, job *Job) error
	Nack(ctx context.Context, job *Job, execErr error) error
	Reschedule(ctx context.Context, job *Job, delay time.Duration) error
	GetPendingCount(ctx context.Context, queue string) (int64, error)
	GetProcessingCount(ctx context.Context, queue string) (int64, error)
	GetDelayedCount(ctx context.Context, queue string) (int64, error)
	GetDLQCount(ctx context.Context, queue string) (int64, error)
}

Storage defines the pluggable storage driver interface required by GopherQueue.

type WorkerPool 

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

WorkerPool manages worker goroutines that fetch and process jobs concurrently.

func NewWorkerPool 

func NewWorkerPool(queueName string, numWorkers int, pollInterval time.Duration, store Storage, exec JobExecutor) *WorkerPool

NewWorkerPool initializes a new WorkerPool instance.

func (*WorkerPool) Start 

func (wp *WorkerPool) Start(ctx context.Context)

Start launches worker goroutines.

func (*WorkerPool) Stop 

func (wp *WorkerPool) Stop()

Stop signals all workers to stop and waits for them to finish.

Source Files

View all Source files

Directories

Path Synopsis
_examples
basic command
middleware command
rate_limited command
cmd
gopherq-dashboard command
gopherq-server command
middleware
distlimit
logger
recover
redis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.