Reconnect on disconnection

client.go

@@ -3,6 +3,7 @@ package amqp
 import (
 	"fmt"
 	"context"
+	"time"
 	"sync"
 
 	"github.com/streadway/amqp"
@@ -12,6 +13,11 @@ import (
 	"github.com/luraproject/lura/proxy"
 )
 
+const (
+	retryInterval = 3 * time.Second
+	maxRetries = 15
+)
+
 func NewBackendFactory(ctx context.Context, logger logging.Logger, bf proxy.BackendFactory) proxy.BackendFactory {
 	f := backendFactory{
 		logger:    logger,
@@ -48,8 +54,28 @@ func (f backendFactory) New(remote *config.Backend) proxy.Proxy {
 	return f.bf(remote)
 }
 
-func (f backendFactory) newChannel(path string) (*amqp.Channel, closer, error) {
-	conn, err := amqp.Dial(path)
+func (f backendFactory) newChannel(ctx context.Context, path string) (*amqp.Channel, closer, error) {
+	var (
+		retries int
+		conn *amqp.Connection
+		err error
+	)
+	for {
+		conn, err = amqp.Dial(path)
+		if err == nil {
+			break
+		}
+		retries += 1
+		f.logger.Error(fmt.Sprintf("AMQP: connection attempt #%d: %s", retries, err.Error()))
+		if retries > maxRetries {
+			break
+		}
+		select {
+		case <-time.After(retryInterval):
+		case <-ctx.Done():
+			break
+		}
+	}
 	if err != nil {
 		return nil, nopCloser, err
 	}

consumer.go

@@ -47,7 +47,7 @@ func (f backendFactory) initConsumer(ctx context.Context, remote *config.Backend
 		return consumerBackend(remote, msgs), nil
 	}
 
-	ch, close, err := f.newChannel(dns)
+	ch, close, err := f.newChannel(ctx, dns)
 	if err != nil {
 		f.logger.Error(fmt.Sprintf("AMQP: getting the channel for %s/%s: %s", dns, cfg.Name, err.Error()))
 		return proxy.NoopProxy, err

producer.go

@@ -54,7 +54,7 @@ func (f backendFactory) initProducer(ctx context.Context, remote *config.Backend
 		return proxy.NoopProxy, err
 	}
 
-	ch, close, err := f.newChannel(dns)
+	ch, close, err := f.newChannel(ctx, dns)
 	if err != nil {
 		f.logger.Error(fmt.Sprintf("AMQP: getting the channel for %s/%s: %s", dns, cfg.Name, err.Error()))
 		return proxy.NoopProxy, err

rpc.go

@@ -45,22 +45,29 @@ type rpcCfg struct {
 	RoutingKey    string `json:"routing_key"`
 }
 
-func (f backendFactory) initRpc(ctx context.Context, remote *config.Backend) (proxy.Proxy, error) {
-	if len(remote.Host) < 1 {
-		return proxy.NoopProxy, errNoBackendHostDefined
-	}
-	dns := remote.Host[0]
+type rpcChannel struct {
+	*amqp.Channel
+	mu *sync.Mutex
+	listeners *sync.Map
+}
 
-	cfg, err := getRpcConfig(remote)
-	if err != nil {
-		f.logger.Debug(fmt.Sprintf("AMQP: %s: %s", dns, err.Error()))
-		return proxy.NoopProxy, err
-	}
+func (c *rpcChannel) Store(ch *amqp.Channel) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.Channel = ch
+}
 
-	ch, close, err := f.newChannel(dns)
+func (c *rpcChannel) Load() *amqp.Channel {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.Channel
+}
+
+func (f backendFactory) initRpcChannel(ctx context.Context, cfg *rpcCfg, dns string, rch *rpcChannel) error {
+	ch, close, err := f.newChannel(ctx, dns)
 	if err != nil {
 		f.logger.Error(fmt.Sprintf("AMQP: getting the channel for %s/%s: %s", dns, cfg.Name, err.Error()))
-		return proxy.NoopProxy, err
+		return err
 	}
 
 	err = ch.ExchangeDeclare(
@@ -75,17 +82,19 @@ func (f backendFactory) initRpc(ctx context.Context, remote *config.Backend) (pr
 	if err != nil {
 		f.logger.Error(fmt.Sprintf("AMQP: declaring the exchange for %s/%s: %s", dns, cfg.Name, err.Error()))
 		close()
-		return proxy.NoopProxy, err
+		return err
 	}
 
 	replies, err := ch.Consume(directReplyTo, "", true, true, false, false, nil)
 	if err != nil {
 		f.logger.Error(fmt.Sprintf("AMQP: consuming direct reply queue %s/%s: %s", dns, cfg.Name, err.Error()))
 		close()
-		return proxy.NoopProxy, err
+		return err
 	}
 
-	listeners := new(sync.Map)
+	rch.Store(ch)
+
+	closeCh := ch.NotifyClose(make(chan *amqp.Error))
 
 	go func() {
 		loop:
@@ -94,19 +103,50 @@ func (f backendFactory) initRpc(ctx context.Context, remote *config.Backend) (pr
 			case <-ctx.Done():
 				close()
 				break loop
+			case _, ok := <-closeCh:
+				if !ok {
+					break
+				}
+				if err := f.initRpcChannel(ctx, cfg, dns, rch); err == nil {
+					break
+				}
 			case reply, ok := <-replies:
 				if !ok {
 					// connection closed
 					break loop
 				}
-				if v, ok := listeners.Load(reply.CorrelationId); ok {
+				if v, ok := rch.listeners.Load(reply.CorrelationId); ok {
 					listener := v.(chan amqp.Delivery)
 					listener <- reply
 				}
 			}
 		}
 	}()
 
+	return nil
+}
+
+func (f backendFactory) initRpc(ctx context.Context, remote *config.Backend) (proxy.Proxy, error) {
+	if len(remote.Host) < 1 {
+		return proxy.NoopProxy, errNoBackendHostDefined
+	}
+	dns := remote.Host[0]
+
+	cfg, err := getRpcConfig(remote)
+	if err != nil {
+		f.logger.Debug(fmt.Sprintf("AMQP: %s: %s", dns, err.Error()))
+		return proxy.NoopProxy, err
+	}
+
+	ch := &rpcChannel{
+		mu: new(sync.Mutex),
+		listeners: new(sync.Map),
+	}
+
+	if err := f.initRpcChannel(ctx, cfg, dns, ch); err != nil {
+		return proxy.NoopProxy, err
+	}
+
 	return func(ctx context.Context, r *proxy.Request) (*proxy.Response, error) {
 		body, err := ioutil.ReadAll(r.Body)
 		if err != nil {
@@ -116,8 +156,8 @@ func (f backendFactory) initRpc(ctx context.Context, remote *config.Backend) (pr
 		correlationId := uuid.NewV4().String()
 
 		listener := make(chan amqp.Delivery, 1)
-		listeners.Store(correlationId, listener)
-		defer listeners.Delete(correlationId)
+		ch.listeners.Store(correlationId, listener)
+		defer ch.listeners.Delete(correlationId)
 
 		contentType := ""
 		headers := amqp.Table{}
@@ -151,7 +191,7 @@ func (f backendFactory) initRpc(ctx context.Context, remote *config.Backend) (pr
 			routingKey = v
 		}
 
-		err = ch.Publish(
+		err = ch.Load().Publish(
 			cfg.Exchange,
 			routingKey,
 			cfg.Mandatory,