Retry feature.

amqp/connection.go

@@ -126,9 +126,9 @@ func (c *Connection) handleConnectionClose() {
 			if err == nil {
 				log.WithFields(log.Fields{
 					"type":     "goevents",
-					"subType": "connection",
+					"sub_type": "connection",
 					"attempt":  i,
-				}).Info("Connection reestablished")
+				}).Info("Connection reestablished.")
 
 				for _, c := range c.reestablishs {
 					c <- true
@@ -138,7 +138,7 @@ func (c *Connection) handleConnectionClose() {
 			} else {
 				log.WithFields(log.Fields{
 					"type":     "goevents",
-					"subType": "connection",
+					"sub_type": "connection",
 					"error":    err,
 					"attempt":  i,
 				}).Error("Error reestablishing connection. Retrying...")

amqp/consumer.go

@@ -12,10 +12,25 @@ import (
 	amqplib "github.com/streadway/amqp"
 )
 
+const (
+	MaxInt32   = 1<<31 - 1
+	MaxRetries = MaxInt32
+)
+
+var (
+	logger = log.WithFields(log.Fields{
+		"type":     "goevents",
+		"sub_type": "consumer",
+	})
+)
+
 type handler struct {
 	action           string
-	handler messaging.EventHandler
+	fn               messaging.EventHandler
 	re               *regexp.Regexp
+	maxRetries       int32
+	retryDelay       time.Duration
+	delayProgression bool
 }
 
 type Consumer struct {
@@ -28,6 +43,7 @@ type Consumer struct {
 
 	channel    *amqplib.Channel
 	queue      *amqplib.Queue
+	retryQueue *amqplib.Queue
 
 	exchangeName string
 	queueName    string
@@ -106,12 +122,12 @@ func (c *Consumer) setupTopology() error {
 		nil,         // arguments
 	)
 
-	c.queue = &q
-
 	if err != nil {
 		return err
 	}
 
+	c.queue = &q
+
 	return nil
 }
 
@@ -122,53 +138,144 @@ func (c *Consumer) handleReestablishedConnnection() {
 		err := c.setupTopology()
 
 		if err != nil {
-			log.WithFields(log.Fields{
+			logger.WithFields(log.Fields{
-				"type":    "goevents",
-				"subType": "consumer",
 				"error": err,
-			}).Error("Error setting up topology after reconnection")
+			}).Error("Error setting up topology after reconnection.")
 		}
 	}
 }
 
 func (c *Consumer) dispatch(msg amqplib.Delivery) {
-	if fn, ok := c.getHandler(msg.RoutingKey); ok {
+	if h, ok := c.getHandler(msg.RoutingKey); ok {
+		delay, ok := getXRetryDelayHeader(msg)
+
+		if !ok {
+			delay = h.retryDelay
+		}
+
+		retryCount, _ := getXRetryCountHeader(msg)
+
 		defer func() {
 			if err := recover(); err != nil {
+				if h.maxRetries > 0 {
+					c.requeueMessage(msg, h, retryCount, delay)
+				} else {
+					logger.WithFields(log.Fields{
+						"error":      err,
+						"message_id": msg.MessageId,
+					}).Error("Failed to process event.")
+
 					if !c.autoAck {
-					msg.Nack(false, true)
+						msg.Ack(false)
+					}
 				}
 			}
 		}()
 
-		ok := fn(msg.Body)
+		death, ok := getXRetryDeathHeader(msg)
+
+		if ok {
+			since := time.Since(death)
+
+			if since < delay {
+				time.Sleep(delay - since)
+			}
+		}
+
+		err := h.fn(msg.Body)
+
+		if err != nil {
+			if h.maxRetries > 0 {
+				if retryCount >= h.maxRetries {
+					logger.WithFields(log.Fields{
+						"max_retries": h.maxRetries,
+						"message_id":  msg.MessageId,
+					}).Error("Maximum retries reached. Giving up.")
 
 					if !c.autoAck {
-			if ok {
 						msg.Ack(false)
-			} else {
-				msg.Nack(false, true)
 					}
+				} else {
+					logger.WithFields(log.Fields{
+						"error":      err,
+						"message_id": msg.MessageId,
+					}).Error("Failed to process event. Retrying...")
+
+					c.requeueMessage(msg, h, retryCount, delay)
+				}
+			} else {
+				logger.WithFields(log.Fields{
+					"error":      err,
+					"message_id": msg.MessageId,
+				}).Error("Failed to process event.")
+			}
+		} else if !c.autoAck {
+			msg.Ack(false)
 		}
 	} else {
 		// got a message from wrong exchange?
 		// ignore and don't requeue.
+		if !c.autoAck {
 			msg.Nack(false, false)
 		}
 	}
+}
 
-func (c *Consumer) getHandler(action string) (messaging.EventHandler, bool) {
+func (c *Consumer) requeueMessage(msg amqplib.Delivery, h *handler, retryCount int32, delay time.Duration) {
+	delayNs := delay.Nanoseconds()
+
+	if h.delayProgression {
+		delayNs *= 2
+	}
+
+	retryMsg := amqplib.Publishing{
+		Headers: amqplib.Table{
+			"x-retry-death": time.Now().UTC(),
+			"x-retry-count": retryCount + 1,
+			"x-retry-max":   h.maxRetries,
+			"x-retry-delay": delayNs,
+		},
+		DeliveryMode: amqplib.Persistent,
+		Timestamp:    time.Now(),
+		Body:         msg.Body,
+		MessageId:    msg.MessageId,
+	}
+
+	err := c.channel.Publish(msg.Exchange, msg.RoutingKey, false, false, retryMsg)
+
+	if err != nil {
+		logger.WithFields(log.Fields{
+			"error": err,
+		}).Error("Failed to retry.")
+
+		if !c.autoAck {
+			msg.Nack(false, true)
+		}
+	} else if !c.autoAck {
+		msg.Ack(false)
+	}
+}
+
+func (c *Consumer) getHandler(action string) (*handler, bool) {
 	for _, h := range c.handlers {
 		if h.re.MatchString(action) {
-			return h.handler, true
+			return &h, true
 		}
 	}
 
 	return nil, false
 }
 
-// Subscribe allow to subscribe an action handler.
+// Subscribe allows to subscribe an action handler.
+// By default it won't retry any failed event.
 func (c *Consumer) Subscribe(action string, handlerFn messaging.EventHandler) error {
+	return c.SubscribeWithOptions(action, handlerFn, time.Duration(0), false, 0)
+}
+
+// SubscribeWithOptions allows to subscribe an action handler with retry options.
+func (c *Consumer) SubscribeWithOptions(action string, handlerFn messaging.EventHandler,
+	retryDelay time.Duration, delayProgression bool, maxRetries int32) error {
+
 	// TODO: Replace # pattern too.
 	pattern := strings.Replace(action, "*", "(.*)", 0)
 	re, err := regexp.Compile(pattern)
@@ -190,9 +297,12 @@ func (c *Consumer) Subscribe(action string, handlerFn messaging.EventHandler) er
 	}
 
 	c.handlers = append(c.handlers, handler{
-		action,
+		action:           action,
-		handlerFn,
+		fn:               handlerFn,
-		re,
+		re:               re,
+		maxRetries:       maxRetries,
+		retryDelay:       retryDelay,
+		delayProgression: delayProgression,
 	})
 
 	return nil
@@ -230,9 +340,7 @@ func (c *Consumer) Unsubscribe(action string) error {
 // Listen start to listen for new messages.
 func (c *Consumer) Consume() {
 	for !c.closed {
-		log.WithFields(log.Fields{
+		logger.WithFields(log.Fields{
-			"type":    "goevents",
-			"subType": "consumer",
 			"queue": c.queueName,
 		}).Debug("Setting up consumer channel...")
 
@@ -247,9 +355,7 @@ func (c *Consumer) Consume() {
 		)
 
 		if err != nil {
-			log.WithFields(log.Fields{
+			logger.WithFields(log.Fields{
-				"type":    "goevents",
-				"subType": "consumer",
 				"queue": c.queueName,
 				"error": err,
 			}).Error("Error setting up consumer...")
@@ -259,21 +365,45 @@ func (c *Consumer) Consume() {
 			continue
 		}
 
-		log.WithFields(log.Fields{
+		logger.WithFields(log.Fields{
-			"type":    "goevents",
-			"subType": "consumer",
 			"queue": c.queueName,
 		}).Info("Consuming messages...")
 
 		for m := range msgs {
-			c.dispatch(m)
+			go c.dispatch(m)
 		}
 
-		log.WithFields(log.Fields{
+		logger.WithFields(log.Fields{
-			"type":    "goevents",
-			"subType": "consumer",
 			"queue":  c.queueName,
 			"closed": c.closed,
-		}).Info("Consumption finished")
+		}).Info("Consumption finished.")
 	}
 }
+
+func getXRetryDeathHeader(msg amqplib.Delivery) (time.Time, bool) {
+	if d, ok := msg.Headers["x-retry-death"]; ok {
+		return d.(time.Time), true
+	}
+
+	return time.Time{}, false
+}
+
+func getXRetryCountHeader(msg amqplib.Delivery) (int32, bool) {
+	if c, ok := msg.Headers["x-retry-count"]; ok {
+		return c.(int32), true
+	}
+
+	return 0, false
+}
+
+func getXRetryDelayHeader(msg amqplib.Delivery) (time.Duration, bool) {
+	if d, ok := msg.Headers["x-retry-delay"]; ok {
+		f, ok := d.(int64)
+
+		if ok {
+			return time.Duration(f), true
+		}
+	}
+
+	return time.Duration(0), false
+}

amqp/integration_test.go

@@ -1,6 +1,7 @@
 package amqp
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
@@ -27,14 +28,14 @@ func TestPublishConsume(t *testing.T) {
 	consumer := c.(*Consumer)
 	consumer.channel.QueuePurge(consumer.queueName, false)
 
-	c.Subscribe("my_action_1", func(body []byte) bool {
+	c.Subscribe("my_action_1", func(body []byte) error {
 		func1 <- true
-		return true
+		return nil
 	})
 
-	c.Subscribe("my_action_2", func(body []byte) bool {
+	c.Subscribe("my_action_2", func(body []byte) error {
 		func2 <- true
-		return true
+		return nil
 	})
 
 	go c.Consume()
@@ -74,14 +75,14 @@ func TestPublishConsumeWildcardAction(t *testing.T) {
 	consumer := c.(*Consumer)
 	consumer.channel.QueuePurge(consumer.queueName, false)
 
-	c.Subscribe("webinar.*", func(body []byte) bool {
+	c.Subscribe("webinar.*", func(body []byte) error {
 		func1 <- true
-		return true
+		return nil
 	})
 
-	c.Subscribe("foobar.*", func(body []byte) bool {
+	c.Subscribe("foobar.*", func(body []byte) error {
 		func2 <- true
-		return true
+		return nil
 	})
 
 	go c.Consume()
@@ -121,14 +122,14 @@ func TestPublishConsumeWildcardActionOrderMatters1(t *testing.T) {
 	consumer := c.(*Consumer)
 	consumer.channel.QueuePurge(consumer.queueName, false)
 
-	c.Subscribe("webinar.*", func(body []byte) bool {
+	c.Subscribe("webinar.*", func(body []byte) error {
 		func1 <- true
-		return true
+		return nil
 	})
 
-	c.Subscribe("webinar.state_changed", func(body []byte) bool {
+	c.Subscribe("webinar.state_changed", func(body []byte) error {
 		func2 <- true
-		return true
+		return nil
 	})
 
 	go c.Consume()
@@ -168,14 +169,14 @@ func TestPublishConsumeWildcardActionOrderMatters2(t *testing.T) {
 	consumer := c.(*Consumer)
 	consumer.channel.QueuePurge(consumer.queueName, false)
 
-	c.Subscribe("webinar.state_changed", func(body []byte) bool {
+	c.Subscribe("webinar.state_changed", func(body []byte) error {
 		func1 <- true
-		return true
+		return nil
 	})
 
-	c.Subscribe("webinar.*", func(body []byte) bool {
+	c.Subscribe("webinar.*", func(body []byte) error {
 		func2 <- true
-		return true
+		return nil
 	})
 
 	go c.Consume()
@@ -215,15 +216,15 @@ func TestPublishConsumeRequeueIfFail(t *testing.T) {
 	consumer := c.(*Consumer)
 	consumer.channel.QueuePurge(consumer.queueName, false)
 
-	c.Subscribe("my_action", func(body []byte) bool {
+	c.SubscribeWithOptions("my_action", func(body []byte) error {
 		if calledOnce {
 			called <- true
-			return true
+			return nil
 		} else {
 			calledOnce = true
-			return false
+			return fmt.Errorf("Error.")
 		}
-	})
+	}, 1*time.Second, false, 5)
 
 	go c.Consume()
 
@@ -260,15 +261,15 @@ func TestPublishConsumeRequeueIfPanic(t *testing.T) {
 	consumer := c.(*Consumer)
 	consumer.channel.QueuePurge(consumer.queueName, false)
 
-	c.Subscribe("my_action", func(body []byte) bool {
+	c.SubscribeWithOptions("my_action", func(body []byte) error {
 		if calledOnce {
 			called <- true
-			return true
+			return nil
 		} else {
 			calledOnce = true
 			panic("this is a panic!")
 		}
-	})
+	}, 1*time.Second, false, 5)
 
 	go c.Consume()
 

amqp/producer.go

@@ -87,7 +87,7 @@ func (p *Producer) Close() {
 func (p *Producer) setupTopology() error {
 	log.WithFields(log.Fields{
 		"type":     "goevents",
-		"subType": "producer",
+		"sub_type": "producer",
 	}).Debug("Setting up topology...")
 
 	p.m.Lock()
@@ -125,8 +125,8 @@ func (p *Producer) setupTopology() error {
 
 	log.WithFields(log.Fields{
 		"type":     "goevents",
-		"subType": "producer",
+		"sub_type": "producer",
-	}).Debug("Topology ready")
+	}).Debug("Topology ready.")
 
 	return nil
 }
@@ -143,7 +143,7 @@ func (p *Producer) handleReestablishedConnnection() {
 			log.WithFields(log.Fields{
 				"type":  "amqp",
 				"error": err,
-			}).Error("Error setting up topology after reconnection")
+			}).Error("Error setting up topology after reconnection.")
 		}
 	}
 }
@@ -151,7 +151,10 @@ func (p *Producer) handleReestablishedConnnection() {
 func (p *Producer) drainInternalQueue() {
 	for m := range p.internalQueue {
 		for i := 0; !p.closed; i++ {
+			messageId, _ := NewUUIDv4()
+
 			msg := amqplib.Publishing{
+				MessageId:    messageId,
 				DeliveryMode: amqplib.Persistent,
 				Timestamp:    time.Now(),
 				Body:         m.data,
@@ -163,9 +166,9 @@ func (p *Producer) drainInternalQueue() {
 
 				log.WithFields(log.Fields{
 					"type":     "goevents",
-					"subType": "producer",
+					"sub_type": "producer",
 					"attempt":  i,
-				}).Debug("Publishing message to the exchange")
+				}).Debug("Publishing message to the exchange.")
 
 				return p.channel.Publish(p.exchangeName, m.action, false, false, msg)
 			}()
@@ -173,7 +176,7 @@ func (p *Producer) drainInternalQueue() {
 			if err != nil {
 				log.WithFields(log.Fields{
 					"type":     "goevents",
-					"subType": "producer",
+					"sub_type": "producer",
 					"error":    err,
 					"attempt":  i,
 				}).Error("Error publishing message to the exchange. Retrying...")
@@ -188,7 +191,7 @@ func (p *Producer) drainInternalQueue() {
 			case <-p.nackChannel:
 				log.WithFields(log.Fields{
 					"type":     "goevents",
-					"subType": "producer",
+					"sub_type": "producer",
 					"attempt":  i,
 				}).Error("Error publishing message to the exchange. Retrying...")
 

amqp/uuid.go

@@ -0,0 +1,22 @@
+package amqp
+
+import (
+	"crypto/rand"
+	"fmt"
+	"io"
+)
+
+func NewUUIDv4() (string, error) {
+	var uuid [16]byte
+
+	_, err := io.ReadFull(rand.Reader, uuid[:])
+
+	if err != nil {
+		return "", err
+	}
+
+	uuid[6] = (uuid[6] & 0x0f) | 0x40 // Version 4
+	uuid[8] = (uuid[8] & 0x3f) | 0x80 // Variant is 10
+
+	return fmt.Sprintf("%x-%x-%x-%x-%x", uuid[0:4], uuid[4:6], uuid[6:8], uuid[8:10], uuid[10:]), nil
+}

examples/consumer/consumer.go

@@ -1,12 +1,13 @@
 package main
 
 import (
+	"fmt"
 	"os"
 	"os/signal"
 	"sync"
 	"syscall"
+	"time"
 
-	"fmt"
 	"github.com/eventials/goevents/amqp"
 )
 
@@ -23,30 +24,40 @@ func main() {
 		panic(err)
 	}
 
-	consumerA.Subscribe("object.eventA", func(body []byte) bool {
+	consumerA.Subscribe("object.eventA", func(body []byte) error {
 		fmt.Println("object.eventA:", string(body))
-		return true
+		return nil
 	})
 
-	consumerA.Subscribe("object.eventB", func(body []byte) bool {
+	consumerA.Subscribe("object.eventB", func(body []byte) error {
 		fmt.Println("object.eventB:", string(body))
-		return true
+		return nil
 	})
 
+	consumerA.SubscribeWithOptions("object.eventToRetryDelay", func(body []byte) error {
+		fmt.Println("object.eventToRetryDelay:", string(body))
+		return fmt.Errorf("Try again.")
+	}, 1*time.Second, true, 5)
+
+	consumerA.SubscribeWithOptions("object.eventToRetry", func(body []byte) error {
+		fmt.Println("object.eventToRetry:", string(body))
+		return fmt.Errorf("Try again.")
+	}, 1*time.Second, false, 10)
+
 	consumerB, err := conn.Consumer(false, "events-exchange", "events-queue-b")
 
 	if err != nil {
 		panic(err)
 	}
 
-	consumerB.Subscribe("object.eventC", func(body []byte) bool {
+	consumerB.Subscribe("object.eventC", func(body []byte) error {
 		fmt.Println("object.eventC:", string(body))
-		return true
+		return nil
 	})
 
-	consumerB.Subscribe("object.eventD", func(body []byte) bool {
+	consumerB.Subscribe("object.eventD", func(body []byte) error {
 		fmt.Println("object.eventD:", string(body))
-		return true
+		return nil
 	})
 
 	var wg sync.WaitGroup

messaging/consumer.go

@@ -1,9 +1,15 @@
 package messaging
 
-type EventHandler func(body []byte) bool
+import (
+	"time"
+)
+
+type EventHandler func(body []byte) error
 
 type Consumer interface {
 	Subscribe(action string, handler EventHandler) error
+	SubscribeWithOptions(action string, handlerFn EventHandler,
+		retryDelay time.Duration, delayPow2 bool, maxRetries int32) error
 	Unsubscribe(action string) error
 	Consume()
 	Close()