goqueue

package module
v1.1.0 Latest Latest
Warning

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

 Go to latest Published: Dec 14, 2025 License: MIT Imports: 13 Imported by: 0

README

GoQueue

Go Version License Status

A simple, flexible, and production-ready queue library for Go

Features • Installation • Quick Start • Documentation • Examples • Contributing

Overview

GoQueue is a lightweight, interface-driven queue library that makes it easy to add background job processing to your Go applications. With support for multiple backends and a clean, intuitive API, you can start simple with in-memory queues and scale to production with Redis.

Why GoQueue?
  • 🎯 Simple API - Minimal boilerplate, maximum productivity
  • 🔌 Pluggable Backends - Swap backends without changing your code
  • 🚀 Production Ready - Built-in retry logic, DLQ, and graceful shutdown
  • ⚡ High Performance - Concurrent workers with configurable pool sizes
  • 📦 Zero Dependencies (for in-memory backend)
  • 🏗️ Clean Architecture - Interface-based design for easy testing

Features

  • ✅ Interface-Based Handlers - Type-safe message handlers with dependency injection
  • ✅ Multiple Backends - In-memory (dev/test), Redis, and RabbitMQ (production)
  • ✅ Concurrent Workers - Process messages in parallel with configurable workers
  • ✅ Automatic Retry - Exponential backoff with configurable retry attempts
  • ✅ Dead Letter Queue - Handle permanently failed messages
  • ✅ Graceful Shutdown - Context-based cancellation
  • ✅ JSON Serialization - Automatic message serialization/deserialization
  • ✅ Distributed Processing - Scale horizontally with Redis backend

Installation

go get github.com/openframebox/goqueue
Requirements
  • Go 1.21 or higher
  • Redis 5.0+ (optional, for Redis backend)
  • RabbitMQ 3.8+ (optional, for RabbitMQ backend)

Quick Start

1. Define Your Message
type EmailTask struct {
    To      string `json:"to"`
    Subject string `json:"subject"`
    Body    string `json:"body"`
}

func (e *EmailTask) QueueName() string {
    return "emails"
}
2. Create a Handler
type EmailHandler struct {
    emailService *EmailService
}

func (h *EmailHandler) QueueName() string {
    return "emails"
}

func (h *EmailHandler) Handle(ctx context.Context, envelope *goqueue.Envelope) error {
    var task EmailTask
    if err := envelope.Unmarshal(&task); err != nil {
        return err
    }

    return h.emailService.Send(task.To, task.Subject, task.Body)
}
3. Set Up the Queue
// Create backend (in-memory for development)
backend := goqueue.NewMemoryBackend()

// Or use Redis for production
// backend := goqueue.NewRedisBackend("localhost:6379")

// Create queue with options
gq := goqueue.New(backend,
    goqueue.WithWorkerCount(5),
    goqueue.WithRetryCount(3),
    goqueue.WithDLQ("failed-emails"),
)

// Register handler
gq.Register(&EmailHandler{emailService: myEmailService})

// Start processing
ctx := context.Background()
gq.Start(ctx)

// Publish messages
gq.Publish(ctx, &EmailTask{
    To:      "user@example.com",
    Subject: "Welcome!",
    Body:    "Thanks for signing up!",
})

// Graceful shutdown
defer gq.Stop()

Backends

In-Memory Backend

Perfect for development, testing, and single-process applications.

backend := goqueue.NewMemoryBackend(
    goqueue.WithBufferSize(200),
)

Features:

  • Zero external dependencies
  • Fast and lightweight
  • Great for testing
  • Single-process only
Redis Backend

Production-ready distributed queue with Redis.

backend := goqueue.NewRedisBackend("localhost:6379",
    goqueue.WithRedisPassword("your-password"),
    goqueue.WithRedisDB(0),
    goqueue.WithPollTimeout(5 * time.Second),
)

// Test connection
if err := backend.Ping(ctx); err != nil {
    log.Fatal(err)
}

Features:

  • Distributed processing across multiple instances
  • Persistent message storage
  • Horizontal scaling
  • Production-ready reliability
RabbitMQ Backend

Enterprise-grade message queue with advanced features.

backend, err := goqueue.NewRabbitMQBackend(
    "amqp://guest:guest@localhost:5672/",
    goqueue.WithRabbitMQDurableQueues(true),
    goqueue.WithRabbitMQPersistence(true),
    goqueue.WithRabbitMQPrefetch(10, 0),
    goqueue.WithRabbitMQPriority(10),
    goqueue.WithRabbitMQDLX("goqueue.dlx", "dead-letter-queue"),
)
if err != nil {
    log.Fatal(err)
}

// Test connection
if err := backend.Ping(context.Background()); err != nil {
    log.Fatal(err)
}

Features:

  • Native Dead Letter Exchange (DLX) support
  • Message persistence and durability
  • Priority queues
  • QoS and prefetch control
  • TLS/SSL support
  • Virtual host isolation
  • Connection pooling
  • Horizontal scaling

Configuration Options:

Option Description Default
WithRabbitMQAuth(user, pass) Set username and password -
WithRabbitMQHost(host, port) Set host and port localhost:5672
WithRabbitMQVHost(vhost) Set virtual host /
WithRabbitMQTLS(cert, key, ca) Enable TLS with certificates Disabled
WithRabbitMQTLSConfig(config) Custom TLS configuration Disabled
WithRabbitMQMaxChannels(n) Channel pool size 100
WithRabbitMQDurableQueues(bool) Persist queues across restarts true
WithRabbitMQPersistence(bool) Persist messages to disk true
WithRabbitMQPrefetch(count, size) QoS settings for load balancing 1, 0
WithRabbitMQPriority(max) Enable priority queues (1-255) Disabled
WithRabbitMQDLX(exchange, queue) Configure Dead Letter Exchange Enabled

TLS Example:

backend, err := goqueue.NewRabbitMQBackend(
    "amqps://user:pass@rabbitmq.example.com:5671/",
    goqueue.WithRabbitMQTLS(
        "/path/to/client-cert.pem",
        "/path/to/client-key.pem",
        "/path/to/ca-cert.pem",
    ),
)

Connection Components Example:

// Build connection from components instead of URL
backend, err := goqueue.NewRabbitMQBackend(
    "",
    goqueue.WithRabbitMQHost("rabbitmq.example.com", 5672),
    goqueue.WithRabbitMQAuth("myuser", "mypassword"),
    goqueue.WithRabbitMQVHost("/production"),
    goqueue.WithRabbitMQDurableQueues(true),
    goqueue.WithRabbitMQPrefetch(20, 0),
)

Configuration

Configure GoQueue behavior with functional options:

Option Description Default
WithWorkerCount(n) Number of concurrent workers per queue 1
WithRetryCount(n) Maximum retry attempts for failed messages 3
WithRetryDelay(d) Initial delay between retries (exponential backoff) 1 second
WithDLQ(name) Enable dead letter queue for failed messages Disabled

Example:

gq := goqueue.New(backend,
    goqueue.WithWorkerCount(10),
    goqueue.WithRetryCount(5),
    goqueue.WithRetryDelay(2 * time.Second),
    goqueue.WithDLQ("failed-jobs"),
)

Examples

See the examples/ directory for complete working examples:

  • Basic - Simple pub/sub with in-memory backend
  • Advanced - Retry, DLQ, concurrent workers
  • Redis - Redis backend usage
  • RabbitMQ - RabbitMQ backend with DLX and priority queues
  • Distributed - Production architecture with separate publisher and worker services
Running Examples
# In-memory example
go run examples/basic/main.go

# Redis example (requires Redis)
REDIS_ADDR=localhost:6379 go run examples/redis/main.go

# With password
REDIS_ADDR=localhost:6379 REDIS_PASSWORD=secret go run examples/redis/main.go

# RabbitMQ example (requires RabbitMQ)
go run examples/rabbitmq/main.go

# Distributed example (run in separate terminals)
cd examples/redis-distributed
go run worker/main.go      # Terminal 1
go run publisher/main.go   # Terminal 2

Production Architecture

GoQueue supports distributed architectures where publishers and workers run as separate services:

┌─────────────┐         ┌────────┐         ┌─────────────┐
│  Publisher  │────────▶│ Redis  │◀────────│   Worker    │
│  (API)      │         │ Queue  │         │  (Process)  │
└─────────────┘         └────────┘         └─────────────┘
                             ▲
                             │
                        ┌────┴────┐
                        │  Worker │
                        │ (Scale) │
                        └─────────┘

See examples/redis-distributed for a complete implementation.

Documentation

Core Interfaces
QueueMessage

All messages must implement this interface:

type QueueMessage interface {
    QueueName() string
}
Handler

All handlers must implement this interface:

type Handler interface {
    QueueName() string
    Handle(ctx context.Context, envelope *Envelope) error
}
Backend

Custom backends must implement:

type Backend interface {
    Publish(ctx context.Context, queue string, envelope *Envelope) error
    Subscribe(ctx context.Context, queue string) (<-chan *Envelope, error)
    Ack(ctx context.Context, messageID string) error
    Nack(ctx context.Context, messageID string) error
    Close() error
}
Envelope Structure

Internal message wrapper with metadata:

type Envelope struct {
    ID          string
    Queue       string
    Data        []byte
    RetryCount  int
    CreatedAt   time.Time
    ProcessedAt *time.Time
    Metadata    map[string]string
}

Error Handling and Retries

When a handler returns an error, GoQueue automatically:

  1. Increments the retry count
  2. Applies exponential backoff: initial_delay * 2^(retry_count-1)
  3. Re-queues the message
  4. After max retries, moves to Dead Letter Queue (if enabled)

Example retry timeline:

  • 1st retry: 1 second
  • 2nd retry: 2 seconds
  • 3rd retry: 4 seconds
  • After max retries: Move to DLQ

Testing

GoQueue is designed to be easily testable:

// Use in-memory backend for tests
func TestMyHandler(t *testing.T) {
    backend := goqueue.NewMemoryBackend()
    gq := goqueue.New(backend)

    handler := &MyHandler{}
    gq.Register(handler)
    gq.Start(context.Background())

    // Publish test message
    gq.Publish(context.Background(), &MyMessage{})

    // Assert expected behavior
}

Environment Variables

Configure examples using environment variables:

# Redis connection
export REDIS_ADDR=localhost:6379
export REDIS_PASSWORD=your-password

# Run example
go run examples/redis/main.go

Monitoring

Redis Queue Monitoring
# Connect to Redis
redis-cli

# Check queue lengths
LLEN goqueue:queue:emails
LLEN goqueue:queue:notifications

# View pending messages
KEYS goqueue:pending:*

# Get message details
GET goqueue:pending:message-id

Performance

  • In-memory backend: 100,000+ messages/sec
  • Redis backend: Depends on Redis and network latency
  • Concurrent workers: Linear scaling up to CPU cores

Roadmap

  • RabbitMQ backend
  • Message priority support (RabbitMQ)
  • AWS SQS backend
  • Kafka backend
  • Delayed/scheduled messages
  • Metrics and monitoring (Prometheus)
  • Message batching
  • Web UI for queue monitoring

Contributing

We welcome contributions! Please see CONTRIBUTING.md for guidelines.

Development
# Clone repository
git clone https://github.com/openframebox/goqueue.git
cd goqueue

# Run tests
go test ./...

# Run with coverage
go test -cover ./...

# Run linter
golangci-lint run

License

GoQueue is released under the MIT License.

Support

Acknowledgments

Built with ❤️ by OpenFrameBox

If you find GoQueue useful, please consider giving it a ⭐️

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrQueueStopped is returned when trying to use a stopped queue
	ErrQueueStopped = errors.New("queue is stopped")

	// ErrWorkersAlreadyRunning is returned when trying to start workers for a queue that already has workers running
	ErrWorkersAlreadyRunning = errors.New("workers already running for this queue")

	// ErrMessageNotFound is returned when a message with the given ID is not found
	ErrMessageNotFound = errors.New("message not found")

	// ErrHandlerAlreadyRegistered is returned when trying to register a handler for a queue that already has one
	ErrHandlerAlreadyRegistered = errors.New("handler already registered for this queue")

	// ErrNoHandlersRegistered is returned when trying to start without any registered handlers
	ErrNoHandlersRegistered = errors.New("no handlers registered")
)

Functions

This section is empty.

Types

type Backend 

type Backend interface {
	// Publish sends a message envelope to the specified queue
	Publish(ctx context.Context, queue string, envelope *Envelope) error

	// Subscribe creates a subscription to the specified queue and returns
	// a channel that will receive message envelopes
	Subscribe(ctx context.Context, queue string) (<-chan *Envelope, error)

	// Ack acknowledges successful processing of a message
	Ack(ctx context.Context, messageID string) error

	// Nack indicates that a message failed to process and should be retried
	Nack(ctx context.Context, messageID string) error

	// Close releases any resources held by the backend
	Close() error
}

Backend defines the interface that all queue backends must implement. This allows easy swapping between different queue implementations (e.g., Redis, RabbitMQ, in-memory, etc.)

type Config 

type Config struct {
	// WorkerCount is the number of concurrent workers processing messages
	WorkerCount int

	// RetryCount is the maximum number of times a failed message will be retried
	RetryCount int

	// RetryDelay is the initial delay between retries (uses exponential backoff)
	RetryDelay time.Duration

	// DLQEnabled enables the dead letter queue for permanently failed messages
	DLQEnabled bool

	// DLQName is the name of the dead letter queue
	DLQName string
}

Config holds the configuration for a GoQueue instance

func DefaultConfig 

func DefaultConfig() Config

DefaultConfig returns a Config with sensible defaults

type Envelope 

type Envelope struct {
	ID          string            `json:"id"`
	Queue       string            `json:"queue"`
	Data        []byte            `json:"data"`
	RetryCount  int               `json:"retry_count"`
	CreatedAt   time.Time         `json:"created_at"`
	ProcessedAt *time.Time        `json:"processed_at,omitempty"`
	Metadata    map[string]string `json:"metadata,omitempty"`
}

Envelope represents the internal wrapper around queue messages with metadata.

func NewEnvelope 

func NewEnvelope(queue string, data []byte) *Envelope

NewEnvelope creates a new envelope with the given queue name and data

func (*Envelope) String 

func (e *Envelope) String() string

String returns a string representation of the envelope

func (*Envelope) Unmarshal 

func (e *Envelope) Unmarshal(v any) error

Unmarshal unmarshals the envelope data into the provided value

type GoQueue 

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

GoQueue is the main queue manager that coordinates workers and the backend

func New 

func New(backend Backend, opts ...Option) *GoQueue

New creates a new GoQueue instance with the given backend and options

func (*GoQueue) Publish 

func (gq *GoQueue) Publish(ctx context.Context, message QueueMessage) error

Publish publishes a message to its designated queue. The message declares which queue it belongs to via QueueName(). The message will be automatically serialized to JSON.

func (*GoQueue) PublishBytes 

func (gq *GoQueue) PublishBytes(ctx context.Context, queue string, data []byte) error

PublishBytes publishes raw bytes to the specified queue.

func (*GoQueue) Register 

func (gq *GoQueue) Register(handler Handler) error

Register registers a handler for its designated queue. Must be called before Start().

func (*GoQueue) Start 

func (gq *GoQueue) Start(ctx context.Context) error

Start starts workers for all registered queues.

func (*GoQueue) Stop 

func (gq *GoQueue) Stop() error

Stop gracefully stops all workers and closes the backend connection

type Handler 

type Handler interface {
	// QueueName returns the name of the queue this handler processes
	QueueName() string

	// Handle processes a single message envelope.
	// Return an error to trigger retry logic, or nil on success.
	Handle(ctx context.Context, envelope *Envelope) error
}

Handler defines the interface that message handlers must implement. Handlers declare which queue they process and how to handle messages.

type HandlerFunc 

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

HandlerFunc is a function adapter that allows using functions as Handlers. This provides backward compatibility and convenience for simple use cases.

func NewHandlerFunc 

func NewHandlerFunc(queueName string, fn func(ctx context.Context, envelope *Envelope) error) HandlerFunc

NewHandlerFunc creates a new HandlerFunc for the given queue

func (HandlerFunc) Handle 

func (f HandlerFunc) Handle(ctx context.Context, envelope *Envelope) error

Handle implements the Handler interface for HandlerFunc

func (HandlerFunc) QueueName 

func (f HandlerFunc) QueueName() string

QueueName implements the Handler interface

type MemoryBackend 

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

MemoryBackend is an in-memory queue backend using Go channels. This is ideal for testing, development, and single-process applications.

func NewMemoryBackend 

func NewMemoryBackend(opts ...MemoryOption) *MemoryBackend

NewMemoryBackend creates a new in-memory backend

func (*MemoryBackend) Ack 

func (b *MemoryBackend) Ack(ctx context.Context, messageID string) error

Ack acknowledges successful processing of a message

func (*MemoryBackend) Close 

func (b *MemoryBackend) Close() error

Close releases resources held by the backend

func (*MemoryBackend) Nack 

func (b *MemoryBackend) Nack(ctx context.Context, messageID string) error

Nack indicates that a message failed to process

func (*MemoryBackend) Publish 

func (b *MemoryBackend) Publish(ctx context.Context, queue string, envelope *Envelope) error

Publish sends a message envelope to the specified queue

func (*MemoryBackend) Subscribe 

func (b *MemoryBackend) Subscribe(ctx context.Context, queue string) (<-chan *Envelope, error)

Subscribe creates a subscription to the specified queue

type MemoryOption 

type MemoryOption func(*MemoryBackend)

MemoryOption is a function that configures the memory backend

func WithBufferSize 

func WithBufferSize(size int) MemoryOption

WithBufferSize sets the channel buffer size for each queue

type Option 

type Option func(*Config)

Option is a function that modifies a Config

func WithDLQ 

func WithDLQ(queueName string) Option

WithDLQ enables the dead letter queue with the specified name

func WithRetryCount 

func WithRetryCount(count int) Option

WithRetryCount sets the maximum number of retries for failed messages

func WithRetryDelay 

func WithRetryDelay(delay time.Duration) Option

WithRetryDelay sets the initial retry delay (exponential backoff is applied)

func WithWorkerCount 

func WithWorkerCount(count int) Option

WithWorkerCount sets the number of concurrent workers

type QueueMessage 

type QueueMessage interface {
	QueueName() string
}

QueueMessage is the interface that all queue messages must implement.

type RabbitMQBackend added in v1.1.0 

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

RabbitMQBackend implements the Backend interface using RabbitMQ

func NewRabbitMQBackend added in v1.1.0 

func NewRabbitMQBackend(url string, opts ...RabbitMQOption) (*RabbitMQBackend, error)

NewRabbitMQBackend creates a new RabbitMQ backend

func NewRabbitMQBackendWithConnection added in v1.1.0 

func NewRabbitMQBackendWithConnection(conn *amqp.Connection, opts ...RabbitMQOption) (*RabbitMQBackend, error)

NewRabbitMQBackendWithConnection creates a new RabbitMQ backend with an existing connection

func (*RabbitMQBackend) Ack added in v1.1.0 

func (b *RabbitMQBackend) Ack(ctx context.Context, messageID string) error

Ack acknowledges successful processing of a message

func (*RabbitMQBackend) Close added in v1.1.0 

func (b *RabbitMQBackend) Close() error

Close releases all resources held by the backend

func (*RabbitMQBackend) Nack added in v1.1.0 

func (b *RabbitMQBackend) Nack(ctx context.Context, messageID string) error

Nack indicates that a message failed to process

func (*RabbitMQBackend) Ping added in v1.1.0 

func (b *RabbitMQBackend) Ping(ctx context.Context) error

Ping checks the connection to RabbitMQ

func (*RabbitMQBackend) Publish added in v1.1.0 

func (b *RabbitMQBackend) Publish(ctx context.Context, queue string, envelope *Envelope) error

Publish sends a message envelope to the specified queue

func (*RabbitMQBackend) Subscribe added in v1.1.0 

func (b *RabbitMQBackend) Subscribe(ctx context.Context, queue string) (<-chan *Envelope, error)

Subscribe creates a subscription to the specified queue and returns a channel

type RabbitMQOption added in v1.1.0 

type RabbitMQOption func(*rabbitmqConfig)

RabbitMQOption is a function that configures RabbitMQBackend

func WithRabbitMQAuth added in v1.1.0 

func WithRabbitMQAuth(username, password string) RabbitMQOption

WithRabbitMQAuth sets the username and password

func WithRabbitMQAutoDelete added in v1.1.0 

func WithRabbitMQAutoDelete(autoDelete bool) RabbitMQOption

WithRabbitMQAutoDelete sets whether queues should auto-delete

func WithRabbitMQDLX added in v1.1.0 

func WithRabbitMQDLX(dlxName, dlqName string) RabbitMQOption

WithRabbitMQDLX configures Dead Letter Exchange

func WithRabbitMQDLXRoutingKey added in v1.1.0 

func WithRabbitMQDLXRoutingKey(key string) RabbitMQOption

WithRabbitMQDLXRoutingKey sets the routing key for DLX

func WithRabbitMQDurableQueues added in v1.1.0 

func WithRabbitMQDurableQueues(durable bool) RabbitMQOption

WithRabbitMQDurableQueues sets whether queues should be durable

func WithRabbitMQExclusive added in v1.1.0 

func WithRabbitMQExclusive(exclusive bool) RabbitMQOption

WithRabbitMQExclusive sets whether queues should be exclusive

func WithRabbitMQHost added in v1.1.0 

func WithRabbitMQHost(host string, port int) RabbitMQOption

WithRabbitMQHost sets the host and port

func WithRabbitMQInsecureSkipVerify added in v1.1.0 

func WithRabbitMQInsecureSkipVerify() RabbitMQOption

WithRabbitMQInsecureSkipVerify skips TLS certificate verification (use only for development)

func WithRabbitMQMaxChannels added in v1.1.0 

func WithRabbitMQMaxChannels(max int) RabbitMQOption

WithRabbitMQMaxChannels sets the maximum number of channels in the pool

func WithRabbitMQPersistence added in v1.1.0 

func WithRabbitMQPersistence(persistent bool) RabbitMQOption

WithRabbitMQPersistence sets whether messages should be persistent

func WithRabbitMQPrefetch added in v1.1.0 

func WithRabbitMQPrefetch(count, size int) RabbitMQOption

WithRabbitMQPrefetch sets QoS prefetch count and size

func WithRabbitMQPriority added in v1.1.0 

func WithRabbitMQPriority(maxPriority uint8) RabbitMQOption

WithRabbitMQPriority enables priority queues with the given max priority

func WithRabbitMQTLS added in v1.1.0 

func WithRabbitMQTLS(certFile, keyFile, caFile string) RabbitMQOption

WithRabbitMQTLS sets TLS configuration from certificate files

func WithRabbitMQTLSConfig added in v1.1.0 

func WithRabbitMQTLSConfig(tlsConfig *tls.Config) RabbitMQOption

WithRabbitMQTLSConfig sets a custom TLS configuration

func WithRabbitMQURL added in v1.1.0 

func WithRabbitMQURL(url string) RabbitMQOption

WithRabbitMQURL sets the full AMQP URL

func WithRabbitMQVHost added in v1.1.0 

func WithRabbitMQVHost(vhost string) RabbitMQOption

WithRabbitMQVHost sets the virtual host

type RedisBackend 

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

RedisBackend is a Redis-based queue backend. This is suitable for production use with distributed systems.

func NewRedisBackend 

func NewRedisBackend(addr string, opts ...RedisOption) *RedisBackend

NewRedisBackend creates a new Redis backend addr is the Redis server address (e.g., "localhost:6379")

func NewRedisBackendWithClient 

func NewRedisBackendWithClient(client *redis.Client, opts ...RedisOption) *RedisBackend

NewRedisBackendWithClient creates a new Redis backend with an existing Redis client

func (*RedisBackend) Ack 

func (b *RedisBackend) Ack(ctx context.Context, messageID string) error

Ack acknowledges successful processing of a message

func (*RedisBackend) Close 

func (b *RedisBackend) Close() error

Close releases resources held by the backend

func (*RedisBackend) Nack 

func (b *RedisBackend) Nack(ctx context.Context, messageID string) error

Nack indicates that a message failed to process

func (*RedisBackend) Ping 

func (b *RedisBackend) Ping(ctx context.Context) error

Ping checks if the Redis connection is alive

func (*RedisBackend) Publish 

func (b *RedisBackend) Publish(ctx context.Context, queue string, envelope *Envelope) error

Publish sends a message envelope to the specified queue

func (*RedisBackend) Subscribe 

func (b *RedisBackend) Subscribe(ctx context.Context, queue string) (<-chan *Envelope, error)

Subscribe creates a subscription to the specified queue

type RedisOption 

type RedisOption func(*RedisBackend)

RedisOption is a function that configures the Redis backend

func WithPollTimeout 

func WithPollTimeout(timeout time.Duration) RedisOption

WithPollTimeout sets the timeout for polling messages from Redis

func WithRedisDB 

func WithRedisDB(db int) RedisOption

WithRedisDB sets the Redis database number (0-15)

func WithRedisPassword 

func WithRedisPassword(password string) RedisOption

WithRedisPassword sets the password for Redis authentication

Source Files

View all Source files

Directories

Path Synopsis
examples
advanced command
basic command
rabbitmq command
redis command
redis-distributed/publisher command
redis-distributed/shared
redis-distributed/worker command

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.