Merge pull request #13 from frain-dev/ogban/feat/pause-retries-for-ratelimit-error

Add error handler

Author: ogbanugot

brokers/localstorage/localstorage.go

@@ -85,7 +85,8 @@ func (b *LocalStorage) Process(msg *disq.Message) error {
 
 	if msgErr != nil {
 		//retry
-		atomic.AddUint32(&b.retries, 1)
+		disq.Logger.Println(disq.FormatHandlerError(msg, task.RetryLimit()))
+		_ = disq.ErrorHandler(msg, msgErr, &b.retries)
 		msg.Err = msgErr
 		err := b.Requeue(msg)
 		if err != nil {
@@ -100,7 +101,6 @@ func (b *LocalStorage) Process(msg *disq.Message) error {
 
 func (b *LocalStorage) Requeue(msg *disq.Message) error {
 	//Requeue
-	msg.RetryCount++
 	err := b.Publish(msg)
 	if err != nil {
 		return err

brokers/redis/list.go

@@ -103,7 +103,8 @@ func (b *List) Process(msg *disq.Message) error {
 	msgErr := task.HandleMessage(msg)
 
 	if msgErr != nil {
-		atomic.AddUint32(&b.retries, 1)
+		disq.Logger.Println(disq.FormatHandlerError(msg, task.RetryLimit()))
+		_ = disq.ErrorHandler(msg, msgErr, &b.retries)
 		msg.Err = msgErr
 		err := b.Requeue(msg)
 		if err != nil {
@@ -126,7 +127,6 @@ func (b *List) Requeue(msg *disq.Message) error {
 	if err != nil {
 		return err
 	}
-	msg.RetryCount++
 	err = b.Publish(msg)
 	if err != nil {
 		return err

brokers/redis/stream.go

@@ -125,8 +125,8 @@ func (b *Stream) Process(msg *disq.Message) error {
 
 	if msgErr != nil {
 		//retry
-		msg.Delay = disq.Delay(msg, msgErr)
-		atomic.AddUint32(&b.retries, 1)
+		disq.Logger.Println(disq.FormatHandlerError(msg, task.RetryLimit()))
+		_ = disq.ErrorHandler(msg, msgErr, &b.retries)
 		msg.Err = msgErr
 		err := b.Requeue(msg)
 		if err != nil {
@@ -253,7 +253,7 @@ func (b *Stream) Requeue(msg *disq.Message) error {
 		return err
 	}
 	//Requeue
-	msg.RetryCount++ //to know how many times it has been retried.
+	// msg.RetryCount++ //to know how many times it has been retried.
 	err = b.Publish(msg)
 	if err != nil {
 		return err

errors.go

@@ -1,14 +1,45 @@
 package disq
 
-import "time"
+import (
+	"sync/atomic"
+	"time"
+)
 
-type Delayer interface {
-	Delay() time.Duration
+type DisqError interface {
+	GetDelay() time.Duration
+	GetRateLimit() bool
 }
 
-func Delay(msg *Message, msgErr error) time.Duration {
-	if delayer, ok := msgErr.(Delayer); ok {
-		return delayer.Delay()
+type Error struct {
+	RateLimit bool
+	Delay     time.Duration
+	Err       error
+}
+
+func (e *Error) Error() string {
+	return e.Err.Error()
+}
+
+func (e *Error) GetDelay() time.Duration {
+	return e.Delay
+}
+
+func (e *Error) GetRateLimit() bool {
+	return e.RateLimit
+}
+
+func ErrorHandler(msg *Message, msgErr error, retries *uint32) error {
+	if disqError, ok := msgErr.(DisqError); ok {
+		if disqError.GetRateLimit() {
+			msg.Delay = disqError.GetDelay()
+			return nil
+		}
+		msg.RetryCount++
+		atomic.AddUint32(retries, 1)
+		msg.Delay = disqError.GetDelay()
+		return nil
 	}
-	return msg.Delay
+	msg.RetryCount++
+	atomic.AddUint32(retries, 1)
+	return nil
 }

util.go

@@ -1,6 +1,7 @@
 package disq
 
 import (
+	"fmt"
 	"math/rand"
 	"os"
 	"strconv"
@@ -37,3 +38,8 @@ func StringToBytes(s string) []byte {
 		}{s, len(s)},
 	))
 }
+
+func FormatHandlerError(msg *Message, retrylimit int) error {
+	return fmt.Errorf("task=%q failed (retrycount=%d/%d will retry after delay=%s): reason:%s",
+		msg.TaskName, msg.RetryCount, retrylimit, msg.Delay, msg.Err)
+}