reconnect timer + fixes

consumer.go

@@ -5,21 +5,20 @@ import (
 	amqp "github.com/rabbitmq/amqp091-go"
 	"golang.org/x/time/rate"
 	"log"
+	"time"
 )
 
 type Consumer interface {
-	Start() chan []byte
+	Start(ctx context.Context, chanLen uint) chan []byte
 }
 
 type consumeHandler struct {
-	ctx     context.Context
+	client *Client
-	client  *Client
-	chanLen int
 }
 
-func (c *consumeHandler) Start() chan []byte {
+func (c *consumeHandler) Start(ctx context.Context, chanLen uint) chan []byte {
-	msgCh := make(chan []byte, c.chanLen)
+	msgCh := make(chan []byte, chanLen)
-	go runConsumer(c.ctx, c.client, msgCh)
+	go runConsumer(ctx, c.client, msgCh)
 
 	return msgCh
 }
@@ -39,8 +38,20 @@ func runConsumer(ctx context.Context, client *Client, msgCh chan []byte) {
 	chClosedCh := make(chan *amqp.Error, 1)
 	client.Channel.NotifyClose(chClosedCh)
 
+	reconnectTimer := time.NewTimer(0)
+	defer reconnectTimer.Stop()
+	<-reconnectTimer.C
+
 	for {
+		if !reconnectTimer.Stop() {
+			select {
+			case <-reconnectTimer.C:
+			default:
+			}
+		}
+
 		select {
+
 		case <-runCtx.Done():
 			err = client.Close()
 			if err != nil {
@@ -49,25 +60,49 @@ func runConsumer(ctx context.Context, client *Client, msgCh chan []byte) {
 			return
 
 		case amqErr := <-chClosedCh:
-			log.Printf("AMQP Channel closed due to: %s\n", amqErr)
+			log.Printf("AMQP Channel closed due to: %s Reconnecting...\n", amqErr)
+			reconnectTimer.Reset(time.Second)
 
+		case <-reconnectTimer.C:
 			deliveries, err = client.consume()
 			if err != nil {
-				log.Println("Error trying to consume, will try again")
+				log.Println("Error trying to consume, will try again. Retry in 5 seconds.")
+				reconnectTimer.Reset(time.Second * 5)
 				continue
 			}
 
 			chClosedCh = make(chan *amqp.Error, 1)
 			client.Channel.NotifyClose(chClosedCh)
 
-		case delivery := <-deliveries:
+		case delivery, ok := <-deliveries:
-			msgCh <- delivery.Body
+			if !ok {
-			log.Printf("Received message: %s\n", delivery.Body)
+				log.Println("Deliveries channel closed unexpectedly")
-
+				reconnectTimer.Reset(time.Second)
-			if err = delivery.Ack(false); err != nil {
+				continue
-				log.Printf("Error acknowledging message: %s\n", err)
+			}
+
+			if err = limiter.Wait(runCtx); err != nil {
+				log.Printf("Wait limiter failed: %s\n", err)
+			}
+
+			select {
+			case <-runCtx.Done():
+				if err = delivery.Nack(false, true); err != nil {
+					log.Printf("Error nacking message: %s\n", err)
+				}
+
+				err = client.Close()
+				if err != nil {
+					log.Printf("Close failed: %s\n", err)
+				}
+				return
+
+			case msgCh <- delivery.Body:
+				log.Printf("Received message: %s\n", delivery.Body)
+				if err = delivery.Ack(false); err != nil {
+					log.Printf("Error acknowledging message: %s\n", err)
+				}
 			}
-			limiter.Wait(runCtx)
 		}
 	}
 }

handler.go

@@ -1,7 +1,6 @@
 package rabbit
 
 import (
-	"context"
 	amqp "github.com/rabbitmq/amqp091-go"
 	"log"
 	"os"
@@ -68,6 +67,6 @@ func NewPublisher(client *Client) Publisher {
 	return &pubHandler{client: client}
 }
 
-func NewConsumer(ctx context.Context, client *Client, chanLen int) Consumer {
+func NewConsumer(client *Client) Consumer {
-	return &consumeHandler{ctx: ctx, client: client, chanLen: chanLen}
+	return &consumeHandler{client: client}
 }

publisher.go

@@ -2,18 +2,34 @@ package rabbit
 
 import (
 	"errors"
+	"log"
 	"time"
 )
 
 type Publisher interface {
-	Push(data []byte) error
+	Start(chanLen uint) <-chan []byte
 }
 
 type pubHandler struct {
 	client *Client
 }
 
-func (p *pubHandler) Push(data []byte) error {
+func (p *pubHandler) Start(chanLen uint) <-chan []byte {
+	ch := make(chan []byte, chanLen)
+	go func() {
+		for {
+			select {
+			case msg := <-ch:
+				if err := p.push(msg); err != nil {
+					log.Printf("Error publishing message: %s", err)
+				}
+			}
+		}
+	}()
+	return ch
+}
+
+func (p *pubHandler) push(data []byte) error {
 	p.client.mutex.Lock()
 	if !p.client.isReady {
 		p.client.mutex.Unlock()