1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

package main

import (
    "context"
    "log"
    "github.com/spf13/viper"
    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"
    _ "go.digitalxero.dev/mq-amqp/v2"
)

func main() {
    // Register message handler globally
    types.RegisterHandler("process", func(ctx context.Context, msg types.Message) error {
        log.Printf("Received: %s", string(msg.Body()))
        return nil
    })

    // Create configuration
    config := viper.New()
    config.Set("url", "amqp://localhost:5672/")
    config.Set("queue", "myqueue")
    config.Set("handler", "process")

    // Create consumer for any supported system
    ctx := context.Background()
    consumer, err := mqutils.NewConsumer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    // Start consuming
    if err := consumer.Run(ctx); err != nil {
        log.Fatal(err)
    }
}

1
2
3
4
5
6
7
8

consumer, err := mqutils.NewConsumerBuilder().
    WithURL("amqp://localhost:5672/").
    WithDestination("myqueue").
    WithHandler(func(ctx context.Context, msg types.Message) error {
        log.Printf("Received: %s", string(msg.Body()))
        return nil
    }).
    Build(ctx)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

package main

import (
    "context"
    "log"
    "github.com/spf13/viper"
    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mq-kafka/v2"
)

func main() {
    // Create configuration
    config := viper.New()
    config.Set("url", "kafka://localhost:9092")

    ctx := context.Background()
    producer, err := mqutils.NewProducer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }
    if err := producer.Start(ctx); err != nil {
        log.Fatal(err)
    }

    // Simple publish
    err = producer.Publish(
        ctx,
        "req-123",               // correlation ID
        "mytopic",               // topic
        "mykey",                 // routing key (Kafka message key)
        "text/plain",            // content type
        []byte("Hello, World!"), // body
    )
    if err != nil {
        log.Fatal(err)
    }

    // Publish with message builder
    message := kafka.NewMessageBuilder().
        WithCorrelationId("req-123").
        WithHeaders(map[string]interface{}{"type": "greeting"}).
        WithBody([]byte("Hello from mqutils!")).
        Build()

    err = producer.PublishMsg(ctx, "mytopic", "mykey", message)
    if err != nil {
        log.Fatal(err)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

package main

import (
    "context"
    "log"
    "os"
    "os/signal"
    "syscall"
    "github.com/spf13/viper"
    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"
    _ "go.digitalxero.dev/mq-amqp/v2"
)

func main() {
    // Register order processing handler
    types.RegisterHandler("order_processor", func(ctx context.Context, msg types.Message) error {
        log.Printf("Processing order: %s", string(msg.Body()))

        // Send reply if needed
        if msg.ReplyTo() != "" {
            response := []byte(`{"status": "processed"}`)
            return msg.Publisher().Publish(ctx,
                msg.CorrelationId(), // correlation ID
                "",                  // exchange (use default)
                msg.ReplyTo(),       // routing key (reply destination)
                "application/json",  // content type
                response)            // body
        }

        return nil
    })

    // Create configuration for AMQP with exchange and routing key
    config := viper.New()
    config.Set("url", "amqp://guest:guest@localhost:5672/")
    config.Set("queue", "orders")
    config.Set("exchange", "events")
    config.Set("routing_key", "order.created")
    config.Set("handler", "order_processor")

    // Graceful shutdown
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    consumer, err := mqutils.NewConsumer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

    go func() {
        <-sigChan
        log.Println("Shutting down...")
        cancel()
    }()

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer error: %v", err)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

// Publisher that routes messages to different queues
func publishToExchange() {
    // Create configuration
    config := viper.New()
    config.Set("url", "amqp://localhost:5672/")

    ctx := context.Background()
    producer, err := mqutils.NewProducer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }
    if err := producer.Start(ctx); err != nil {
        log.Fatal(err)
    }

    // Route to different queues using routing keys
    events := []struct {
        routingKey string
        message    string
    }{
        {"user.created", `{"event": "user_created", "id": 123}`},
        {"user.updated", `{"event": "user_updated", "id": 123}`},
        {"order.placed", `{"event": "order_placed", "order_id": 456}`},
    }

    for _, event := range events {
        // Use direct publish method for routing
        err := producer.Publish(
            ctx,
            "",                    // correlation ID
            "events",              // exchange
            event.routingKey,      // routing key
            "application/json",    // content type
            []byte(event.message), // body
        )
        if err != nil {
            log.Printf("Failed to publish %s: %v", event.routingKey, err)
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

package main

import (
    "context"
    "log"
    "github.com/spf13/viper"
    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"
    _ "go.digitalxero.dev/mq-kafka/v2"
)

func main() {
    // Register batch handler for high throughput
    types.RegisterBatchHandler("user_events", func(ctx context.Context, msgs []types.Message) error {
        log.Printf("Processing batch of %d messages", len(msgs))

        for _, msg := range msgs {
            // Process each message
            log.Printf("User event: %s", string(msg.Body()))
        }

        return nil
    })

    // Create configuration for Kafka with consumer group and batch processing
    config := viper.New()
    config.Set("url", "kafka://localhost:9092")
    config.Set("topic", "user-events")
    config.Set("consumer_group", "user-service")
    config.Set("initial_offset", "earliest")
    config.Set("retry_topic", "user-events-retry")
    config.Set("retry_max_retries", 3)
    config.Set("dead_letter_topic", "user-events-dlt")
    config.Set("handler", "user_events")
    config.Set("enable_batch_processing", true)
    config.Set("batch_size", 50)
    config.Set("batch_timeout", "1s")

    ctx := context.Background()
    consumer, err := mqutils.NewConsumer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    if err := consumer.Run(ctx); err != nil {
        log.Fatal(err)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

func kafkaProducerExample() {
    // Create configuration
    config := viper.New()
    config.Set("url", "kafka://localhost:9092")

    ctx := context.Background()
    producer, err := mqutils.NewProducer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }
    if err := producer.Start(ctx); err != nil {
        log.Fatal(err)
    }

    // Produce messages
    for i := 0; i < 100; i++ {
        err := producer.Publish(
            ctx,
            fmt.Sprintf("msg-%d", i), // correlation ID
            "user-events",            // topic
            fmt.Sprintf("user-%d", i), // routing key (Kafka message key)
            "application/json",       // content type
            []byte(fmt.Sprintf(`{"user_id": %d, "action": "login"}`, i)), // body
        )
        if err != nil {
            log.Printf("Failed to publish message %d: %v", i, err)
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

package main

import (
    "context"
    "log"
    "github.com/spf13/viper"
    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"
    _ "go.digitalxero.dev/mq-nats/v2"
)

func main() {
    // Register event processor handler
    types.RegisterHandler("event_processor", func(ctx context.Context, msg types.Message) error {
        subject := msg.RoutingKey() // NATS subject
        log.Printf("Received on %s: %s", subject, string(msg.Body()))
        return nil
    })

    // Create configuration for NATS Core with subject pattern
    config := viper.New()
    config.Set("url", "nats://localhost:4222")
    config.Set("subject", "events.>")
    config.Set("queue_group", "event-processors") // optional load balancing
    config.Set("handler", "event_processor")

    ctx := context.Background()
    consumer, err := mqutils.NewConsumer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    if err := consumer.Run(ctx); err != nil {
        log.Fatal(err)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

func jetStreamExample() {
    // Register order processor handler
    types.RegisterHandler("order_processor", func(ctx context.Context, msg types.Message) error {
        log.Printf("Processing order: %s", string(msg.Body()))

        // JetStream provides message replay and persistence;
        // returning an error redelivers (up to max_deliver)
        return nil
    })

    // Create configuration for JetStream with durable consumer
    config := viper.New()
    config.Set("url", "jetstream://localhost:4222")
    config.Set("subject", "orders.created")
    config.Set("use_jetstream", true)
    config.Set("stream_name", "orders")
    config.Set("consumer_name", "order-processor")
    config.Set("durable", true)
    config.Set("max_deliver", 3)
    config.Set("handler", "order_processor")

    ctx := context.Background()
    consumer, err := mqutils.NewConsumer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    if err := consumer.Run(ctx); err != nil {
        log.Fatal(err)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

package main

import (
    "context"
    "log"
    "time"
    "github.com/spf13/viper"
    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"
    _ "go.digitalxero.dev/mq-aws/v2"
)

func main() {
    // Register SQS processor handler
    types.RegisterHandler("sqs_processor", func(ctx context.Context, msg types.Message) error {
        log.Printf("Processing SQS message: %s", string(msg.Body()))

        // Long processing time handled by visibility timeout
        return processLongRunningTask(msg.Body())
    })

    // Create configuration for SQS with visibility timeout
    config := viper.New()
    config.Set("url", "sqs://us-east-1/my-queue")
    config.Set("visibility_timeout", 300)
    config.Set("max_messages", 10)
    // Optional: applies a RedrivePolicy so failed messages land on a DLQ
    config.Set("dead_letter_queue_url", "https://sqs.us-east-1.amazonaws.com/123456789012/my-dlq")
    config.Set("max_retries", 3)
    config.Set("handler", "sqs_processor")

    ctx := context.Background()
    consumer, err := mqutils.NewConsumer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    if err := consumer.Run(ctx); err != nil {
        log.Fatal(err)
    }
}

func processLongRunningTask(data []byte) error {
    // Simulate long-running task
    time.Sleep(60 * time.Second)
    return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

func sqsFifoExample() {
    // Create producer configuration for a FIFO queue
    config := viper.New()
    config.Set("url", "sqs://us-east-1/my-queue.fifo")

    ctx := context.Background()
    producer, err := mqutils.NewProducer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }
    if err := producer.Start(ctx); err != nil {
        log.Fatal(err)
    }

    // Publish messages; the correlation ID doubles as the deduplication ID
    for i := 0; i < 5; i++ {
        err := producer.Publish(
            ctx,
            fmt.Sprintf("msg-%d", i), // correlation ID (deduplication ID)
            "",                       // exchange (not used in SQS)
            "my-queue.fifo",          // queue name
            "application/json",       // content type
            []byte(fmt.Sprintf(`{"sequence": %d}`, i)), // body
        )
        if err != nil {
            log.Printf("Failed to publish FIFO message: %v", err)
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

package main

import (
    "context"
    "log"
    "github.com/spf13/viper"
    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"
    _ "go.digitalxero.dev/mq-gcp/v2"
)

func main() {
    // Register Pub/Sub processor handler
    types.RegisterHandler("pubsub_processor", func(ctx context.Context, msg types.Message) error {
        log.Printf("Pub/Sub message: %s", string(msg.Body()))

        // Extract Pub/Sub attributes from headers
        if attrs := msg.Headers(); attrs != nil {
            if orderKey, ok := attrs["ordering_key"]; ok {
                log.Printf("Ordering key: %v", orderKey)
            }
        }

        return nil
    })

    // Create configuration for GCP Pub/Sub with project and subscription
    config := viper.New()
    config.Set("url", "pubsub://my-project/my-topic?subscription=my-subscription")
    config.Set("enable_message_ordering", true)
    config.Set("max_concurrent_handlers", 10)
    // Optional: DeadLetterPolicy (requires IAM grants to the Pub/Sub
    // service account)
    config.Set("dead_letter_topic", "my-topic-dlt")
    config.Set("max_delivery_attempts", 5)
    config.Set("handler", "pubsub_processor")

    ctx := context.Background()
    consumer, err := mqutils.NewConsumer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    if err := consumer.Run(ctx); err != nil {
        log.Fatal(err)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

package main

import (
    "context"
    "log"
    "github.com/spf13/viper"
    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"
    _ "go.digitalxero.dev/mq-redis/v2"
)

func main() {
    // Register Redis processor handler
    types.RegisterHandler("redis_processor", func(ctx context.Context, msg types.Message) error {
        channel := msg.RoutingKey()
        log.Printf("Redis channel %s: %s", channel, string(msg.Body()))
        return nil
    })

    // Create configuration for Redis Pub/Sub with a channel pattern
    config := viper.New()
    config.Set("url", "redis://localhost:6379")
    config.Set("channel_name", "events.*")
    config.Set("use_patterns", true)
    config.Set("handler", "redis_processor")

    ctx := context.Background()
    consumer, err := mqutils.NewConsumer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    if err := consumer.Run(ctx); err != nil {
        log.Fatal(err)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

func redisStreamsExample() {
    // Register stream processor handler
    types.RegisterHandler("stream_processor", func(ctx context.Context, msg types.Message) error {
        log.Printf("Stream message ID %s: %s", msg.MessageId(), string(msg.Body()))

        // Redis Streams provide message ID and ordering
        return nil
    })

    // Create configuration for Redis Streams with consumer group.
    // Pending messages abandoned by dead consumers are recovered via
    // XAutoClaim (Redis >= 6.2) using claim_idle_time_seconds.
    config := viper.New()
    config.Set("url", "redisstream://localhost:6379/mystream")
    config.Set("consumer_group", "processors")
    config.Set("consumer_name", "worker-1")
    config.Set("claim_idle_time_seconds", 30)
    config.Set("handler", "stream_processor")

    ctx := context.Background()
    consumer, err := mqutils.NewConsumer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    if err := consumer.Run(ctx); err != nil {
        log.Fatal(err)
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109

package main

import (
    "context"
    "fmt"
    "log"
    "time"
    "github.com/spf13/viper"
    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"
    _ "go.digitalxero.dev/mq-amqp/v2"
)

// Request-response pattern implementation
func requestResponseExample() {
    // Setup responder
    go setupResponder()

    // Setup requester
    time.Sleep(time.Second) // Wait for responder to start
    sendRequest()
}

func setupResponder() {
    // Register request handler
    types.RegisterHandler("request_handler", func(ctx context.Context, msg types.Message) error {
        log.Printf("Processing request: %s", string(msg.Body()))

        // Process request and send response
        if msg.ReplyTo() != "" {
            response := fmt.Sprintf("Processed: %s", string(msg.Body()))
            return msg.Publisher().Publish(ctx,
                msg.CorrelationId(), // correlation ID
                "",                  // exchange (use default)
                msg.ReplyTo(),       // routing key (reply destination)
                "text/plain",        // content type
                []byte(response))    // body
        }

        return nil
    })

    // Create configuration for responder
    config := viper.New()
    config.Set("url", "amqp://localhost:5672/")
    config.Set("queue", "requests")
    config.Set("handler", "request_handler")

    ctx := context.Background()
    consumer, err := mqutils.NewConsumer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Responder error: %v", err)
    }
}

func sendRequest() {
    // Register response handler
    types.RegisterHandler("response_handler", func(ctx context.Context, msg types.Message) error {
        log.Printf("Received response: %s", string(msg.Body()))
        return nil
    })

    // Consumer for the reply queue
    config := viper.New()
    config.Set("url", "amqp://localhost:5672/")
    config.Set("queue", "responses")
    config.Set("handler", "response_handler")

    ctx := context.Background()
    consumer, err := mqutils.NewConsumer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    // Producer for the request
    producerConfig := viper.New()
    producerConfig.Set("url", "amqp://localhost:5672/")
    producer, err := mqutils.NewProducer(ctx, producerConfig)
    if err != nil {
        log.Fatal(err)
    }
    if err := producer.Start(ctx); err != nil {
        log.Fatal(err)
    }

    // Send request
    err = producer.Publish(
        ctx,
        "req-123",                            // correlation ID
        "",                                   // exchange (use default)
        "requests",                           // routing key (destination)
        "text/plain",                         // content type
        []byte("Hello, please process this"), // body
    )
    if err != nil {
        log.Fatal(err)
    }

    runCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
    defer cancel()

    if err := consumer.Run(runCtx); err != nil {
        log.Printf("Requester error: %v", err)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

package main

import (
    "context"
    "log"
    "time"
    "github.com/spf13/viper"
    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"
    _ "go.digitalxero.dev/mq-kafka/v2"
)

func healthMonitoringExample() {
    // Register health processor handler
    types.RegisterHandler("health_processor", func(ctx context.Context, msg types.Message) error {
        return nil
    })

    // Create configuration for health monitoring
    config := viper.New()
    config.Set("url", "kafka://localhost:9092")
    config.Set("topic", "health-test")
    config.Set("consumer_group", "health-monitor")
    config.Set("handler", "health_processor")

    ctx := context.Background()
    consumer, err := mqutils.NewConsumer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    // Start health monitoring
    go monitorHealth(consumer)

    if err := consumer.Run(ctx); err != nil {
        log.Fatal(err)
    }
}

func monitorHealth(consumer types.Consumer) {
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()

    for range ticker.C {
        health, err := consumer.HealthCheck(context.Background())
        if err != nil {
            log.Printf("Health check failed: %v", err)
            continue
        }

        log.Printf("Health Status: %s", health.Status())
        log.Printf("Health Message: %s", health.Message())

        if details := health.Details(); details != nil {
            log.Printf("Health Details: %+v", details)
        }

        if health.Status() != types.HealthStatusHealthy {
            log.Printf("WARNING: Consumer is not healthy!")
            // Implement alerting logic here
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

func batchProcessingExample() {
    // Register batch processor handler
    types.RegisterBatchHandler("batch_processor", func(ctx context.Context, msgs []types.Message) error {
        log.Printf("Processing batch of %d messages", len(msgs))

        failures := 0

        for _, msg := range msgs {
            if err := processIndividualMessage(msg); err != nil {
                log.Printf("Failed to process message %s: %v", msg.MessageId(), err)
                _ = msg.Nack() // reject just this message
                failures++
                continue
            }
            _ = msg.Ack() // settle successful messages individually
        }

        if failures > 0 {
            log.Printf("Batch processing completed with %d failures", failures)
        }

        // Everything is already settled; nil is safe either way
        return nil
    })

    // Create configuration for batch processing
    config := viper.New()
    config.Set("url", "kafka://localhost:9092")
    config.Set("topic", "batch-events")
    config.Set("consumer_group", "batch-processor")
    config.Set("handler", "batch_processor")
    config.Set("enable_batch_processing", true)
    config.Set("batch_size", 100)
    config.Set("batch_timeout", "5s")

    ctx := context.Background()
    consumer, err := mqutils.NewConsumer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    if err := consumer.Run(ctx); err != nil {
        log.Fatal(err)
    }
}

func processIndividualMessage(msg types.Message) error {
    // Simulate processing
    return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

package main

import (
    "context"
    "testing"

    "github.com/stretchr/testify/require"
    "go.digitalxero.dev/mqutils/v2/types"
)

func TestMessageProcessing(t *testing.T) {
    // Test the handler logic directly — it is just a function
    handler := func(ctx context.Context, msg types.Message) error {
        // your processing logic
        return nil
    }

    types.RegisterHandler("test_handler", handler)

    registered := types.GetHandler("test_handler")
    require.NotNil(t, registered)

    // Drive the handler with a fake types.Message implementation
    // (see the queue_testing module for ready-made helpers).
}