fix: rabbitmq will reconnect if the channel is closed (#99)

* fix: rabbitmq will reconnect if the channel is closed

* chore: ensure push fail after 5 attempt

Author: 42atomys

pkg/storage/rabbitmq/rabbitmq.go

@@ -1,8 +1,11 @@
 package rabbitmq
 
 import (
+	"errors"
 	"fmt"
+	"time"
 
+	"github.com/rs/zerolog/log"
 	"github.com/streadway/amqp"
 
 	"atomys.codes/webhooked/internal/valuable"
@@ -32,10 +35,12 @@ type config struct {
 	Exchange           string            `mapstructure:"exchange" json:"exchange"`
 }
 
+const maxAttempt = 5
+
 // ContentType is the function for get content type used to push data in the
 // storage. When no content type is defined, the default one is used instead
 // Default: text/plain
-func (c config) ContentType() string {
+func (c *config) ContentType() string {
 	if c.DefinedContentType != "" {
 		return c.DefinedContentType
 	}
@@ -67,6 +72,15 @@ func NewStorage(configRaw map[string]interface{}) (*storage, error) {
 		return nil, err
 	}
 
+	go func() {
+		for {
+			reason := <-newClient.client.NotifyClose(make(chan *amqp.Error))
+			log.Warn().Msgf("connection to rabbitmq closed, reason: %v", reason)
+
+			newClient.reconnect()
+		}
+	}()
+
 	if newClient.routingKey, err = newClient.channel.QueueDeclare(
 		newClient.config.QueueName,
 		newClient.config.Durable,
@@ -83,26 +97,60 @@ func NewStorage(configRaw map[string]interface{}) (*storage, error) {
 
 // Name is the function for identified if the storage config is define in the webhooks
 // Run is made from external caller
-func (c storage) Name() string {
+func (c *storage) Name() string {
 	return "rabbitmq"
 }
 
 // Push is the function for push data in the storage
 // A run is made from external caller
 // @param value that will be pushed
 // @return an error if the push failed
-func (c storage) Push(value interface{}) error {
-	if err := c.channel.Publish(
-		c.config.Exchange,
-		c.routingKey.Name,
-		c.config.Mandatory,
-		c.config.Immediate,
-		amqp.Publishing{
-			ContentType: c.config.ContentType(),
-			Body:        []byte(fmt.Sprintf("%v", value)),
-		}); err != nil {
-		return err
+func (c *storage) Push(value interface{}) error {
+	for attempt := 0; attempt < maxAttempt; attempt++ {
+		err := c.channel.Publish(
+			c.config.Exchange,
+			c.routingKey.Name,
+			c.config.Mandatory,
+			c.config.Immediate,
+			amqp.Publishing{
+				ContentType: c.config.ContentType(),
+				Body:        []byte(fmt.Sprintf("%v", value)),
+			})
+
+		if err != nil {
+			if errors.Is(err, amqp.ErrClosed) {
+				log.Warn().Err(err).Msg("connection to rabbitmq closed. reconnecting...")
+				c.reconnect()
+				continue
+			} else {
+				return err
+			}
+		}
+		return nil
 	}
 
-	return nil
+	return errors.New("max attempt to publish reached")
+}
+
+// reconnect is the function to reconnect to the amqp server if the connection
+// is lost. It will try to reconnect every seconds until it succeed to connect
+func (c *storage) reconnect() {
+	for {
+		// wait 1s for reconnect
+		time.Sleep(time.Second)
+
+		conn, err := amqp.Dial(c.config.DatabaseURL.First())
+		if err == nil {
+			c.client = conn
+			c.channel, err = c.client.Channel()
+			if err != nil {
+				log.Error().Err(err).Msg("channel cannot be connected")
+				continue
+			}
+			log.Debug().Msg("reconnect success")
+			break
+		}
+
+		log.Error().Err(err).Msg("reconnect failed")
+	}
 }

pkg/storage/rabbitmq/rabbitmq_test.go

@@ -68,7 +68,33 @@ func TestRunRabbitMQPush(t *testing.T) {
 }
 
 func TestContentType(t *testing.T) {
-	assert.Equal(t, "text/plain", config{}.ContentType())
-	assert.Equal(t, "text/plain", config{DefinedContentType: ""}.ContentType())
-	assert.Equal(t, "application/json", config{DefinedContentType: "application/json"}.ContentType())
+	assert.Equal(t, "text/plain", (&config{}).ContentType())
+	assert.Equal(t, "text/plain", (&config{DefinedContentType: ""}).ContentType())
+	assert.Equal(t, "application/json", (&config{DefinedContentType: "application/json"}).ContentType())
+}
+
+func TestReconnect(t *testing.T) {
+	if testing.Short() {
+		t.Skip("rabbitmq testing is skiped in short version of test")
+		return
+	}
+
+	newClient, err := NewStorage(map[string]interface{}{
+		"databaseUrl":      "amqp://user:password@127.0.0.1:5672",
+		"queueName":        "hello",
+		"contentType":      "text/plain",
+		"durable":          false,
+		"deleteWhenUnused": false,
+		"exclusive":        false,
+		"noWait":           false,
+		"mandatory":        false,
+		"immediate":        false,
+	})
+	assert.NoError(t, err)
+
+	assert.NoError(t, newClient.Push("Hello"))
+	assert.NoError(t, newClient.client.Close())
+	assert.NoError(t, newClient.Push("Hello"))
+	assert.NoError(t, newClient.channel.Close())
+	assert.NoError(t, newClient.Push("Hello"))
 }