reconnect on queue close

taikoxyz · Mar 2, 2024 · 2b17d99 · 2b17d99
1 parent ae26d8a
commit 2b17d99
Showing 1 changed file with 12 additions and 2 deletions.
diff --git a/packages/relayer/pkg/queue/rabbitmq/queue.go b/packages/relayer/pkg/queue/rabbitmq/queue.go
@@ -163,8 +163,6 @@ func (r *RabbitMQ) Ack(ctx context.Context, msg queue.Message) error {
 
 	err := rmqMsg.Ack(false)
 
-	slog.Info("attempted acknowledge rabbitmq message")
-
 	if err != nil {
 		slog.Error("error acknowledging rabbitmq message", "err", err.Error())
 		return err
@@ -214,6 +212,12 @@ func (r *RabbitMQ) Notify(ctx context.Context, wg *sync.WaitGroup) error {
 				slog.Error("rabbitmq notify close connection")
 			}
 
+			r.Close(ctx)
+
+			if err := r.connect(); err != nil {
+				return err
+			}
+
 			return queue.ErrClosed
 		case err := <-r.chErrCh:
 			if err != nil {
@@ -222,6 +226,12 @@ func (r *RabbitMQ) Notify(ctx context.Context, wg *sync.WaitGroup) error {
 				slog.Error("rabbitmq notify close channel")
 			}
 
+			r.Close(ctx)
+
+			if err := r.connect(); err != nil {
+				return err
+			}
+
 			return queue.ErrClosed
 		case returnMsg := <-r.notifyReturnCh:
 			slog.Error("rabbitmq notify return", "id", returnMsg.MessageId, "err", returnMsg.ReplyText)