Repository: ThreeDotsLabs/watermill

Issue:
Allow to select which errors are retryable in the Retry middleware
## Feature request

### Description

Add a new configuration option to the retry middleware that would make it possible to customize which types of errors are retryable.

### Example use case

At the company I work for, we use a custom error package that allows to specify error severities. We have a convention that "retryable errors" are propagated as "Runtime Severity".

It would be great if I could just reuse the existing Retry middleware and have it run a custom function to
check if the returned errors is retryable or not.

A more concrete use case would be to not retry if the input message is in an invalid format, because even if we retry, the result will always be the same once the msg payload is corrupted somehow.

#### How it can look like in code

```go
// Default implementation
func allErrorsAreRetryable(err error) bool {
	return true
}

// Retry provides a middleware that retries the handler if errors are returned.
// The retry behaviour is configurable, with exponential backoff and maximum elapsed time.
type Retry struct {
        // existing options ...
	
        // RetryableError returns true whenever an error can be retried
        // by default all errors are retryable.
	RetryableError func(error) bool

}

// Middleware returns the Retry middleware.
func (r Retry) Middleware(h message.HandlerFunc) message.HandlerFunc {
	return func(msg *message.Message) ([]*message.Message, error) {
		producedMessages, err := h(msg)
		if err == nil {
			return producedMessages, nil
		}

                // >>> Custom implementation to check if the error is retryable or not
		if ok := r.RetryableError(err); !ok {
                        // If the error is not retryable we completely skip the middleware implementation and return
                        // whatever the inner handler function returns
			return producedMessages, err
		}

		expBackoff := backoff.NewExponentialBackOff()
		expBackoff.InitialInterval = r.InitialInterval
                // existing implementation ...
}
```

I would be willing to implement this!

Discussion:
@m110: Released in watermill v1.5.0 

Solution patch:
diff --git a/message/router/middleware/retry.go b/message/router/middleware/retry.go
index b828073..5b37894 100644
--- a/message/router/middleware/retry.go
+++ b/message/router/middleware/retry.go
@@ -10,6 +10,16 @@ import (
 	"github.com/ThreeDotsLabs/watermill/message"
 )
 
+// RetryParams holds the parameters for a retry attempt
+type RetryParams struct {
+	// Err is the error that caused the retry attempt.
+	Err error
+	// RetryNum is the number of the retry attempt, starting from 1.
+	RetryNum int
+	// Delay is the delay for the next retry attempt.
+	Delay time.Duration
+}
+
 // Retry provides a middleware that retries the handler if errors are returned.
 // The retry behaviour is configurable, with exponential backoff and maximum elapsed time.
 type Retry struct {
@@ -32,6 +42,10 @@ type Retry struct {
 	// The number of the current retry is passed as retryNum,
 	OnRetryHook func(retryNum int, delay time.Duration)
 
+	// ShouldRetry is an optional function that will be executed before each retry attempt.
+	// If ShouldRetry returns false, the retry will not be attempted.
+	ShouldRetry func(params RetryParams) bool
+
 	Logger watermill.LoggerAdapter
 }
 
@@ -62,6 +76,11 @@ func (r Retry) Middleware(h message.HandlerFunc) message.HandlerFunc {
 	retryLoop:
 		for {
 			waitTime := expBackoff.NextBackOff()
+
+			if r.ShouldRetry != nil && !r.ShouldRetry(RetryParams{RetryNum: retryNum, Err: err, Delay: waitTime}) {
+				return producedMessages, err
+			}
+
 			select {
 			case <-ctx.Done():
 				return producedMessages, err



Test patch:
diff --git a/message/router/middleware/retry_test.go b/message/router/middleware/retry_test.go
index 8c2fc91..1f10442 100644
--- a/message/router/middleware/retry_test.go
+++ b/message/router/middleware/retry_test.go
@@ -198,3 +198,25 @@ func TestRetry_max_interval(t *testing.T) {
 		assert.True(t, delay <= maxInterval, "wait interval %d (%s) exceeds maxInterval (%s)", i, delay, maxInterval)
 	}
 }
+
+func TestRetry_should_retry(t *testing.T) {
+	retry := middleware.Retry{
+		MaxRetries: 5,
+		ShouldRetry: func(params middleware.RetryParams) bool {
+			return params.Err.Error() != "this should be skipped"
+		},
+	}
+
+	runCount := 0
+
+	h := retry.Middleware(func(msg *message.Message) (messages []*message.Message, e error) {
+		runCount++
+		return nil, errors.New("this should be skipped")
+	})
+
+	handlerMessages, handlerErr := h(message.NewMessage("1", nil))
+
+	assert.Equal(t, 1, runCount)
+	assert.Nil(t, handlerMessages)
+	assert.EqualError(t, handlerErr, "this should be skipped")
+}