rabbitmq

Go RabbitMQ

Home  /

A modular, production-ready Go library for RabbitMQ with pluggable architecture. Built on the contract-implementation pattern, where core interfaces live in the root package and concrete implementations are provided by specialized sub-packages. This design enables maximum flexibility, testability, and extensibility for enterprise messaging solutions.

Table of Contents

• Requirements
• Installation
• Pluggable Sub-Packages
• Key Features
• Simple Queue Configuration
• Quick Start
• Production Usage
• Advanced Examples
• Documentation
• Contributing
• Security
• License

🔝 back to top

Requirements

• Go 1.24+ (recommended)
• RabbitMQ 4.0+ (recommended)

🔝 back to top

Installation

go get github.com/cloudresty/go-rabbitmq

🔝 back to top

Pluggable Sub-Packages

Each sub-package implements core interfaces defined in the root package, enabling you to mix and match features as needed:

🔝 back to top

Key Features

Contract-Implementation Architecture

• Core Interfaces: All contracts defined in the root package
• Pluggable Implementations: Concrete implementations in specialized sub-packages
• Mix & Match: Combine any features - encryption + compression + pooling + streams
• Testing: Easy mocking of interfaces for comprehensive unit testing

🔝 back to top

Production-Ready Features

• Topology Auto-Healing: Automatic topology validation and recreation enabled by default
• Connection Pooling: Distribute load across multiple connections with health monitoring
• Message Encryption: AES-256-GCM encryption with secure key management
• Compression: Gzip/Zlib compression with configurable thresholds
• Saga Pattern: Distributed transaction orchestration with automatic compensation
• Streams Support: High-throughput RabbitMQ Streams for event sourcing
• Performance Monitoring: Latency tracking, rate monitoring, comprehensive metrics

🔝 back to top

Developer Experience

• ULID Message IDs: 6x faster than UUIDs, database-optimized, lexicographically sortable
• Auto-Reconnection: Intelligent retry with configurable exponential backoff
• Graceful Shutdown: Signal handling with proper resource cleanup and timeouts
• Comprehensive Documentation: Each sub-package has detailed README with examples

🔝 back to top

Simple Queue Configuration

This library provides a straightforward approach to queue configuration with user control over topology:

Quorum Queues by Default

• High Availability: Built-in replication across cluster nodes
• Data Safety: No message loss during node failures
• Poison Message Protection: Automatic delivery limits prevent infinite redelivery loops
• Better Performance: Optimized for throughput in clustered environments

🔝 back to top

Dead Letter Configuration

• Manual Configuration: Full control over dead letter exchange and routing configuration
• Flexible Setup: Configure dead letter handling exactly as needed for your topology
• Error Handling: Failed messages routed according to your dead letter configuration

🔝 back to top

Topology Auto-Healing (Enabled by Default)

• Automatic Validation: Every publish/consume operation validates topology exists
• Auto-Recreation: Missing exchanges, queues, and bindings are automatically recreated
• Background Monitoring: Periodic validation every 30 seconds (default, customizable)
• Zero Configuration: Works out of the box - no setup required

🔝 back to top

Easy Customization

// Default: Auto-healing quorum queue (production-ready!)
client, _ := rabbitmq.NewClient(rabbitmq.FromEnv())
admin := client.Admin()
queue, _ := admin.DeclareQueue(ctx, "orders")  // Automatically tracked & protected

// Custom quorum settings
queue, _ := admin.DeclareQueue(ctx, "payments",
    rabbitmq.WithQuorumGroupSize(5),       // Custom cluster size
    rabbitmq.WithDeliveryLimit(3),         // Max retry attempts
)

// With dead letter configuration  
queue, _ := admin.DeclareQueue(ctx, "processing",
    rabbitmq.WithDeadLetter("errors.dlx", "failed"), // Manual DLX setup
)

// Advanced users can opt-out if needed
client, _ := rabbitmq.NewClient(
    rabbitmq.FromEnv(),
    rabbitmq.WithoutTopologyValidation(),            // Disable auto-healing
    rabbitmq.WithoutTopologyAutoRecreation(),        // Keep validation, disable auto-recreation
    rabbitmq.WithoutTopologyBackgroundValidation(),  // Disable background monitoring only
)

// Or customize the background validation interval
client, _ := rabbitmq.NewClient(
    rabbitmq.FromEnv(),
    rabbitmq.WithTopologyValidationInterval(10*time.Second), // Custom interval
)

// Legacy compatibility (opt-in)
queue, _ := admin.DeclareQueue(ctx, "legacy",
    rabbitmq.WithClassicQueue(),           // Classic queue type
)

Benefits: Get enterprise-grade reliability and availability with simple, user-controlled configuration.

🔝 back to top

Quick Start

package main

import (
    "context"
    "log"

    "github.com/cloudresty/go-rabbitmq"
)

func main() {
    // Create a basic client (topology auto-healing enabled by default!)
    client, err := rabbitmq.NewClient(
        rabbitmq.WithHosts("localhost:5672"),
        rabbitmq.WithCredentials("guest", "guest"),
        rabbitmq.WithConnectionName("my-service"),
    )
    if err != nil {
        log.Fatal("Failed to create client:", err)
    }
    defer client.Close()

    // Declare a queue - automatically tracked and protected from deletion!
    admin := client.Admin()
    queue, err := admin.DeclareQueue(context.Background(), "user-events")
    if err != nil {
        log.Fatal("Failed to declare queue:", err)
    }
    log.Printf("Created auto-healing production queue: %s", queue.Name)

    // Create a publisher
    publisher, err := client.NewPublisher(
        rabbitmq.WithDefaultExchange("events"),
    )
    if err != nil {
        log.Fatal("Failed to create publisher:", err)
    }
    defer publisher.Close()

    // Publish a message with auto-generated ULID
    message := rabbitmq.NewMessage([]byte(`{"event": "user_signup", "user_id": 123}`))
    err = publisher.Publish(context.Background(), "events", "user.created", message)
    if err != nil {
        log.Fatal("Failed to publish:", err)
    }

    // Create a consumer
    consumer, err := client.NewConsumer()
    if err != nil {
        log.Fatal("Failed to create consumer:", err)
    }
    defer consumer.Close()

    // Consume messages
    err = consumer.Consume(context.Background(), "user-events", func(ctx context.Context, delivery *rabbitmq.Delivery) error {
        log.Printf("Received: %s (ID: %s)", delivery.Body, delivery.MessageId)
        return nil
    })
    if err != nil {
        log.Fatal("Failed to consume:", err)
    }
}

🔝 back to top

Production Usage

Environment-Based Configuration

# Set environment variables for deployment
export RABBITMQ_HOST=rabbitmq.production.com
export RABBITMQ_USERNAME=myservice
export RABBITMQ_PASSWORD=securepassword
export RABBITMQ_VHOST=/production
export RABBITMQ_CONNECTION_NAME=order-service

🔝 back to top

High-Availability Setup

// Production-ready client with HA
client, err := rabbitmq.NewClient(
    rabbitmq.WithHosts("rabbit1:5672", "rabbit2:5672", "rabbit3:5672"),
    rabbitmq.WithCredentials("user", "pass"),
    rabbitmq.WithVHost("/production"),
    rabbitmq.WithConnectionName("order-service"),
    rabbitmq.WithReconnectPolicy(&rabbitmq.ExponentialBackoff{
        InitialDelay: 1 * time.Second,
        MaxDelay:     30 * time.Second,
        Multiplier:   2.0,
        MaxAttempts:  10,
    }),
)

// Connection pooling for high throughput
pool, err := pool.New(20,
    pool.WithClientOptions(/* client options */),
    pool.WithHealthCheck(30 * time.Second),
    pool.WithAutoRepair(true),
)

// Performance monitoring
monitor := performance.NewMonitor()

// Graceful shutdown management
shutdownManager := shutdown.NewManager()
shutdownManager.RegisterComponents(publisher, consumer, pool)
shutdownManager.SetupSignalHandler()

🔝 back to top

Advanced Examples

Production Defaults Demo

See examples/production-defaults/ for a comprehensive demonstration of the production-ready defaults including quorum queues, dead letter configuration, and customization options.

Topology Auto-Healing Demo

See examples/topology-features/ for a comprehensive demonstration of topology validation, auto-recreation, environment configuration, and opt-out options.

Complete Feature Integration

package main

import (
    "context"
    "log"

    "github.com/cloudresty/go-rabbitmq"
    "github.com/cloudresty/go-rabbitmq/compression"
    "github.com/cloudresty/go-rabbitmq/encryption"
    "github.com/cloudresty/go-rabbitmq/pool"
    "github.com/cloudresty/go-rabbitmq/performance"
    "github.com/cloudresty/go-rabbitmq/shutdown"
)

func main() {
    // Create connection pool for high-throughput
    connectionPool, err := pool.New(10,
        pool.WithClientOptions(
            rabbitmq.WithHosts("localhost:5672"),
            rabbitmq.WithCredentials("guest", "guest"),
        ),
        pool.WithHealthCheck(30*time.Second),
        pool.WithAutoRepair(true),
    )
    if err != nil {
        log.Fatal("Failed to create pool:", err)
    }
    defer connectionPool.Close()

    // Get client from pool
    client, err := connectionPool.Get()
    if err != nil {
        log.Fatal("No healthy connections:", err)
    }

    // Create pluggable features
    compressor := compression.NewGzip()
    encryptor, _ := encryption.NewAESGCM([]byte("your-32-byte-encryption-key-here"))
    monitor := performance.NewMonitor()

    // Create publisher with all features
    publisher, err := client.NewPublisher(
        rabbitmq.WithCompression(compressor),
        rabbitmq.WithEncryption(encryptor),
        rabbitmq.WithCompressionThreshold(100),
        rabbitmq.WithConfirmation(5*time.Second),
    )
    if err != nil {
        log.Fatal("Failed to create publisher:", err)
    }
    defer publisher.Close()

    // Publish with monitoring
    message := rabbitmq.NewMessage([]byte(`{"large": "payload with lots of data..."}`))

    start := time.Now()
    err = publisher.Publish(context.Background(), "events", "data.processed", message)
    duration := time.Since(start)

    // Record performance metrics
    monitor.RecordPublish(err == nil, duration)

    // Setup graceful shutdown
    shutdownManager := shutdown.NewManager()
    shutdownManager.RegisterComponents(publisher, consumer, connectionPool)

    // Handle shutdown signals
    shutdownManager.SetupSignalHandler()
    shutdownManager.Wait() // Blocks until SIGINT/SIGTERM
}

🔝 back to top

Saga Pattern for Distributed Transactions

package main

import (
    "context"
    "log"

    "github.com/cloudresty/go-rabbitmq"
    "github.com/cloudresty/go-rabbitmq/saga"
)

func main() {
    client, _ := rabbitmq.NewClient(rabbitmq.WithHosts("localhost:5672"))
    store := saga.NewInMemoryStore()

    // Define step and compensation handlers
    stepHandlers := map[string]saga.StepHandler{
        "create_order":    createOrderHandler,
        "reserve_inventory": reserveInventoryHandler,
        "charge_payment":  chargePaymentHandler,
    }

    compensationHandlers := map[string]saga.CompensationHandler{
        "cancel_order":      cancelOrderHandler,
        "release_inventory": releaseInventoryHandler,
        "refund_payment":    refundPaymentHandler,
    }

    // Create saga manager with orchestration engine
    manager, err := saga.NewManager(client, store, saga.Config{
        SagaExchange:         "sagas",
        StepQueue:           "saga.steps",
        CompensateQueue:     "saga.compensate",
        StepHandlers:        stepHandlers,
        CompensationHandlers: compensationHandlers,
    })
    if err != nil {
        log.Fatal(err)
    }
    defer manager.Close()

    // Start orchestration engine
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    go func() {
        if err := manager.Run(ctx); err != nil {
            log.Printf("Orchestration engine error: %v", err)
        }
    }()

    // Define distributed transaction steps
    steps := []saga.Step{
        {Name: "create_order", Action: "create_order", Compensation: "cancel_order"},
        {Name: "reserve_inventory", Action: "reserve_inventory", Compensation: "release_inventory"},
        {Name: "charge_payment", Action: "charge_payment", Compensation: "refund_payment"},
    }

    // Start saga (engine will orchestrate automatically)
    s, err := manager.Start(ctx, "order_processing", steps, map[string]any{
        "customer_id": "cust-123",
        "order_total": 99.99,
    })
    if err != nil {
        log.Fatal(err)
    }

    log.Printf("Started saga: %s", s.ID)
}

// Implement your step and compensation handlers...
func createOrderHandler(ctx context.Context, s *saga.Saga, step *saga.Step) error {
    // Business logic for order creation
    return nil
}

🔝 back to top

High-Performance Streams

package main

import (
    "context"
    "log"
    "time"

    "github.com/cloudresty/go-rabbitmq"
    "github.com/cloudresty/go-rabbitmq/streams"
)

func main() {
    client, _ := rabbitmq.NewClient(rabbitmq.WithHosts("localhost:5672"))
    streamsHandler := streams.NewHandler(client)

    // Create high-throughput stream
    config := rabbitmq.StreamConfig{
        MaxAge:            24 * time.Hour,
        MaxLengthMessages: 10_000_000,
        MaxLengthBytes:    10 * 1024 * 1024 * 1024, // 10GB
        InitialClusterSize: 3, // High availability
    }

    err := streamsHandler.CreateStream(context.Background(), "events.stream", config)
    if err != nil {
        log.Printf("Stream creation: %v", err)
    }

    // High-speed publishing
    for i := 0; i < 100000; i++ {
        message := rabbitmq.NewMessage([]byte(fmt.Sprintf("Event %d: %s", i, time.Now())))
        err := streamsHandler.PublishToStream(context.Background(), "events.stream", message)
        if err != nil {
            log.Printf("Failed to publish: %v", err)
            break
        }
    }

    // Consume from stream
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    err = streamsHandler.ConsumeFromStream(ctx, "events.stream", func(ctx context.Context, delivery *rabbitmq.Delivery) error {
        log.Printf("Stream message: %s", delivery.Body)
        return nil
    })
}

🔝 back to top

Documentation

🔝 back to top

Contributing

We welcome contributions! Please see our Contributing Guidelines for details.

1. Fork the repository
2. Create a feature branch
3. Add tests for your changes
4. Ensure all tests pass
5. Submit a pull request

🔝 back to top

Security

If you discover a security vulnerability, please report it via email to security@cloudresty.com.

🔝 back to top

License

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

🔝 back to top

An open source project brought to you by the Cloudresty team.

Website  |  LinkedIn  |  BlueSky  |  GitHub  |  Docker Hub