From 9b9d8a306405910a6da1e7406ad0f8f505e64929 Mon Sep 17 00:00:00 2001
From: nquidox <nquidox@gmail.com>
Date: Wed, 8 Apr 2026 12:21:19 +0300
Subject: [PATCH 1/3] removed unused import

---
 publisher.go | 1 -
 1 file changed, 1 deletion(-)

diff --git a/publisher.go b/publisher.go
index ccb415f..c938eb5 100644
--- a/publisher.go
+++ b/publisher.go
@@ -3,7 +3,6 @@ package rabbit
 import (
 	"context"
 	"errors"
-	"log"
 	"time"
 )
 

From fd495892faae5b3880fba6656fbc60a8097c244e Mon Sep 17 00:00:00 2001
From: nquidox <nquidox@gmail.com>
Date: Mon, 27 Apr 2026 00:06:46 +0300
Subject: [PATCH 2/3] queue opts

---
 handler.go | 25 +++++++++++++++++--------
 service.go | 15 ++++++++++++---
 2 files changed, 29 insertions(+), 11 deletions(-)

diff --git a/handler.go b/handler.go
index 68f5815..6a8dd3f 100644
--- a/handler.go
+++ b/handler.go
@@ -13,7 +13,7 @@ import (
 
 type Client struct {
 	mutex           *sync.Mutex
-	queueName       string
+	queueOptions    *QueueOpts
 	logger          *log.Logger
 	connection      *amqp.Connection
 	Channel         *amqp.Channel
@@ -36,7 +36,16 @@ type options struct {
 	logger            *log.Logger
 }
 
-func NewClient(address Address, queueName string, opts ...Option) (*Client, error) {
+type QueueOpts struct {
+	QueueName  string
+	Durable    bool
+	AutoDelete bool
+	Exclusive  bool
+	NoWait     bool
+	Args       amqp.Table
+}
+
+func NewClient(address Address, queueOpts QueueOpts, opts ...Option) (*Client, error) {
 	l := log.New(os.Stdout, "", log.LstdFlags)
 
 	addr, err := address.makeAddr()
@@ -44,16 +53,16 @@ func NewClient(address Address, queueName string, opts ...Option) (*Client, erro
 		return nil, errors.Join(errBadAddr, err)
 	}
 
-	if queueName == "" {
+	if queueOpts.QueueName == "" {
 		l.Fatal(errNoQueue)
 	}
 
 	client := Client{
-		mutex:     &sync.Mutex{},
-		queueName: queueName,
-		done:      make(chan bool),
-		connected: make(chan struct{}),
-		logger:    l,
+		mutex:        &sync.Mutex{},
+		queueOptions: &queueOpts,
+		done:         make(chan bool),
+		connected:    make(chan struct{}),
+		logger:       l,
 	}
 
 	o := options{
diff --git a/service.go b/service.go
index 6e3355d..c3f9ec8 100644
--- a/service.go
+++ b/service.go
@@ -78,7 +78,16 @@ func (c *Client) init(conn *amqp.Connection) error {
 		return err
 	}
 
-	_, err = ch.QueueDeclare(c.queueName, false, false, false, false, nil)
+	//_, err = ch.QueueDeclare(c.queueName, false, false, false, false, nil)
+	_, err = ch.QueueDeclare(
+		c.queueOptions.QueueName,
+		c.queueOptions.Durable,
+		c.queueOptions.AutoDelete,
+		c.queueOptions.Exclusive,
+		c.queueOptions.NoWait,
+		c.queueOptions.Args,
+	)
+
 	if err != nil {
 		return err
 	}
@@ -120,7 +129,7 @@ func (c *Client) unsafePush(data []byte) error {
 	return c.Channel.PublishWithContext(
 		ctx,
 		"",
-		c.queueName,
+		c.queueOptions.QueueName,
 		false,
 		false,
 		amqp.Publishing{
@@ -147,7 +156,7 @@ func (c *Client) consume() (<-chan amqp.Delivery, error) {
 	}
 
 	return c.Channel.Consume(
-		c.queueName,
+		c.queueOptions.QueueName,
 		"",
 		false,
 		false,

From 72f039e4f7819930e5e6ed3faa6aa10ba81392aa Mon Sep 17 00:00:00 2001
From: nquidox <nquidox@gmail.com>
Date: Mon, 4 May 2026 11:50:59 +0300
Subject: [PATCH 3/3] reconnect bugfix

---
 consumer.go | 42 ++++++++++++++++++++++++------------------
 1 file changed, 24 insertions(+), 18 deletions(-)

diff --git a/consumer.go b/consumer.go
index 6f125c7..5c68bf6 100644
--- a/consumer.go
+++ b/consumer.go
@@ -28,6 +28,15 @@ func runConsumer(ctx context.Context, client *Client, msgCh chan []byte) {
 
 	limiter := rate.NewLimiter(rate.Every(client.opts.consumerRateLimit), client.opts.consumerBurstSize)
 
+	reconnectSignal := make(chan struct{}, 1)
+	reconnectTrigger := func() {
+		select {
+		case reconnectSignal <- struct{}{}:
+		default:
+		}
+	}
+
+	// initial consume
 	deliveries, err := client.consume()
 	if err != nil {
 		client.logger.Printf("Could not start consuming: %s\n", err)
@@ -37,18 +46,7 @@ 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():
@@ -58,25 +56,33 @@ func runConsumer(ctx context.Context, client *Client, msgCh chan []byte) {
 			}
 			return
 
-		case amqErr := <-chClosedCh:
-			client.logger.Printf("AMQP Channel closed due to: %s Reconnecting...\n", amqErr)
-			reconnectTimer.Reset(time.Second)
+		case <-reconnectSignal:
+			select {
+			case <-runCtx.Done():
+				return
+			case <-time.After(client.opts.reconnectDelay):
+			}
 
-		case <-reconnectTimer.C:
 			deliveries, err = client.consume()
 			if err != nil {
-				client.logger.Println("Error trying to consume, will try again. Retry in 5 seconds.")
-				reconnectTimer.Reset(time.Second * 5)
+				client.logger.Printf("Consume reconnect failed, retry: %s\n", err)
+				reconnectTrigger()
 				continue
 			}
 
+			// re-create closing chan
 			chClosedCh = make(chan *amqp.Error, 1)
 			client.Channel.NotifyClose(chClosedCh)
+			client.logger.Println("Consume reconnect success")
+
+		case amqErr := <-chClosedCh:
+			client.logger.Printf("AMQP Channel closed due to: %s Reconnecting...\n", amqErr)
+			reconnectTrigger()
 
 		case delivery, ok := <-deliveries:
 			if !ok {
 				client.logger.Println("Deliveries channel closed unexpectedly")
-				reconnectTimer.Reset(time.Second)
+				reconnectTrigger()
 				continue
 			}