fix: retry logic with dlx feature on rabbitmq (#17)

Author: bxcodec

README.md

@@ -1,12 +1,13 @@
 # goqueue
 
-GoQueue - one library to rule them all. A golang wrapper that handles all the complexity of every Queue platforms. Extensible and easy to learn
+GoQueue - One library to rule them all. A Golang wrapper that handles all the complexity of various Queue platforms. Extensible and easy to learn.
 
 ## Index
 
 - [Support](#support)
 - [Getting Started](#getting-started)
 - [Example](#example)
+- [Advance Setups](#advance-setups)
 - [Contribution](#contribution)
 
 ## Support
@@ -58,12 +59,15 @@ func initExchange(ch *amqp.Channel, exchangeName string) error {
 }
 
 func main() {
+
+	// Initialize the RMQ connection
 	rmqDSN := "amqp://rabbitmq:rabbitmq@localhost:5672/"
 	rmqConn, err := amqp.Dial(rmqDSN)
 	if err != nil {
 		panic(err)
 	}
 
+	// Initialize the Publisher
 	rmqPub := publisher.NewPublisher(
 		publisherOpts.PublisherPlatformRabbitMQ,
 		publisherOpts.WithRabbitMQPublisherConfig(&publisherOpts.RabbitMQPublisherConfig{
@@ -151,6 +155,13 @@ func handler() interfaces.InboundMessageHandlerFunc {
 
 ```
 
+## Advance Setups
+
+### RabbitMQ -- Retry Concept
+
+![Goqueue Retry Architecture RabbitMQ](misc/images/rabbitmq-retry.png)
+Src: [Excalidraw Link](https://link.excalidraw.com/readonly/9sphJpzXzQIAVov3z8G7)
+
 ## Contribution
 
 ---

examples/rabbitmq/basic/docker-compose.yaml

@@ -0,0 +1,12 @@
+version: "3.7"
+services:
+  rabbitmq-test:
+    image: rabbitmq:3.13.3-management-alpine
+    container_name: "goqueue-rabbitmq-example-basic"
+    hostname: rabbitmq
+    ports:
+      - "15671:15672"
+      - "5672:5672"
+    volumes:
+      - ../../../tests/localconf/rabbitmq/rabbitmq.definition.json:/opt/definitions.json:ro
+      - ../../../tests/localconf/rabbitmq/rabbitmq.config:/etc/rabbitmq/rabbitmq.config:ro

examples/rabbitmq/main.go examples/rabbitmq/basic/main.goexamples/rabbitmq/main.go renamed to examples/rabbitmq/basic/main.go

@@ -0,0 +1,12 @@
+version: "3.7"
+services:
+  rabbitmq-test:
+    image: rabbitmq:3.13.3-management-alpine
+    container_name: "goqueue-rabbitmq-example-with-retries"
+    hostname: rabbitmq
+    ports:
+      - "15671:15672"
+      - "5672:5672"
+    volumes:
+      - ../../../tests/localconf/rabbitmq/rabbitmq.definition.json:/opt/definitions.json:ro
+      - ../../../tests/localconf/rabbitmq/rabbitmq.config:/etc/rabbitmq/rabbitmq.config:ro

examples/rabbitmq/withretries/docker-compose.yaml

@@ -0,0 +1,121 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+
+	"github.com/bxcodec/goqueue"
+	"github.com/bxcodec/goqueue/consumer"
+	"github.com/bxcodec/goqueue/interfaces"
+	"github.com/bxcodec/goqueue/middleware"
+	"github.com/bxcodec/goqueue/options"
+	consumerOpts "github.com/bxcodec/goqueue/options/consumer"
+	publisherOpts "github.com/bxcodec/goqueue/options/publisher"
+	"github.com/bxcodec/goqueue/publisher"
+)
+
+func initExchange(ch *amqp.Channel, exchangeName string) error {
+	return ch.ExchangeDeclare(
+		exchangeName,
+		"topic",
+		true,
+		false,
+		false,
+		false,
+		nil,
+	)
+}
+
+func main() {
+	rmqDSN := "amqp://rabbitmq:rabbitmq@localhost:5672/"
+	rmqConn, err := amqp.Dial(rmqDSN)
+	if err != nil {
+		panic(err)
+	}
+
+	rmqPub := publisher.NewPublisher(
+		publisherOpts.PublisherPlatformRabbitMQ,
+		publisherOpts.WithRabbitMQPublisherConfig(&publisherOpts.RabbitMQPublisherConfig{
+			Conn:                     rmqConn,
+			PublisherChannelPoolSize: 5,
+		}),
+		publisherOpts.WithPublisherID("publisher_id"),
+		publisherOpts.WithMiddlewares(
+			middleware.HelloWorldMiddlewareExecuteBeforePublisher(),
+			middleware.HelloWorldMiddlewareExecuteAfterPublisher(),
+		),
+	)
+
+	requeueChannel, err := rmqConn.Channel()
+	if err != nil {
+		panic(err)
+	}
+
+	defer requeueChannel.Close()
+	initExchange(requeueChannel, "goqueue")
+
+	consumerChannel, err := rmqConn.Channel()
+	if err != nil {
+		panic(err)
+	}
+	defer consumerChannel.Close()
+	rmqConsumer := consumer.NewConsumer(
+		consumerOpts.ConsumerPlatformRabbitMQ,
+		consumerOpts.WithRabbitMQConsumerConfig(consumerOpts.RabbitMQConfigWithDefaultTopicFanOutPattern(
+			consumerChannel,
+			requeueChannel,
+			"goqueue",                      // exchange name
+			[]string{"goqueue.payments.#"}, // routing keys pattern
+		)),
+		consumerOpts.WithConsumerID("consumer_id"),
+		consumerOpts.WithMiddlewares(
+			middleware.HelloWorldMiddlewareExecuteAfterInboundMessageHandler(),
+			middleware.HelloWorldMiddlewareExecuteBeforeInboundMessageHandler(),
+		),
+		consumerOpts.WithMaxRetryFailedMessage(5),
+		consumerOpts.WithBatchMessageSize(1),
+		consumerOpts.WithQueueName("consumer_queue"),
+	)
+
+	queueSvc := goqueue.NewQueueService(
+		options.WithConsumer(rmqConsumer),
+		options.WithPublisher(rmqPub),
+		options.WithMessageHandler(handler()),
+	)
+	go func() {
+		for i := 0; i < 10; i++ {
+			data := map[string]interface{}{
+				"message": fmt.Sprintf("Hello World %d", i),
+			}
+			jbyt, _ := json.Marshal(data)
+			err := queueSvc.Publish(context.Background(), interfaces.Message{
+				Data:   data,
+				Action: "goqueue.payments.create",
+				Topic:  "goqueue",
+			})
+			if err != nil {
+				panic(err)
+			}
+			fmt.Println("Message Sent: ", string(jbyt))
+		}
+	}()
+
+	// change to context.Background() if you want to run it forever
+	// ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	// defer cancel()
+	err = queueSvc.Start(context.Background())
+	if err != nil {
+		panic(err)
+	}
+}
+
+func handler() interfaces.InboundMessageHandlerFunc {
+	return func(ctx context.Context, m interfaces.InboundMessage) (err error) {
+		fmt.Printf("Message: %+v\n", m)
+		// something happend, we need to requeue the message
+		return m.RetryWithDelayFn(ctx, interfaces.ExponentialBackoffDelayFn)
+	}
+}

examples/rabbitmq/withretries/main.go

@@ -9,4 +9,6 @@ const (
 	ContentType        = "goqueue-content-type"
 	QueueServiceAgent  = "goqueue-queue-service-agent"
 	MessageID          = "goqueue-message-id"
+	OriginalTopicName  = "goqueue-original-topic-name"
+	OriginalActionName = "goqueue-original-action-name"
 )

headers/key/const.go

@@ -1,16 +1,23 @@
 package interfaces
 
-type DelayFn func(retries int64) (delay int64)
+type DelayFn func(currenRetries int64) (delay int64)
 
 var (
 	// ExponentialBackoffDelayFn is a delay function that implements exponential backoff.
-	// It takes the number of retries as input and returns the delay in milliseconds.
-	ExponentialBackoffDelayFn DelayFn = func(retries int64) (delay int64) {
-		return 2 << retries
+	// It takes the number of retries as input and returns the delay in seconds.
+	ExponentialBackoffDelayFn DelayFn = func(currenRetries int64) (delay int64) {
+		return 2 << (currenRetries - 1)
 	}
+
+	// LinearDelayFn is a delay function that implements linear delay.
+	// It takes the number of retries as input and returns the delay in seconds.
+	LinearDelayFn DelayFn = func(currenRetries int64) (delay int64) {
+		return currenRetries
+	}
+
 	// NoDelayFn is a DelayFn implementation that returns 0 delay for retries.
-	NoDelayFn DelayFn = func(retries int64) (delay int64) {
+	NoDelayFn DelayFn = func(currenRetries int64) (delay int64) {
 		return 0
 	}
-	DefaultDelayFn DelayFn = NoDelayFn
+	DefaultDelayFn DelayFn = LinearDelayFn
 )

interfaces/delayfn.go

@@ -28,5 +28,5 @@ type InboundMessage struct {
 	// eg RabbitMQ: https://www.rabbitmq.com/docs/dlx
 	MoveToDeadLetterQueue func(ctx context.Context) (err error) `json:"-"`
 	// Requeue is used to put the message back to the tail of the queue after a delay.
-	PutToBackOfQueueWithDelay func(ctx context.Context, delayFn DelayFn) (err error) `json:"-"`
+	RetryWithDelayFn func(ctx context.Context, delayFn DelayFn) (err error) `json:"-"`
 }

interfaces/inboundmessagehandler.go

@@ -273,7 +273,7 @@ func handlerRequeue(t *testing.T) interfaces.InboundMessageHandlerFunc {
 			return m.RetryCount
 		}
 
-		err = m.PutToBackOfQueueWithDelay(ctx, delayFn)
+		err = m.RetryWithDelayFn(ctx, delayFn)
 		assert.NoError(t, err)
 		return
 	}