chore(queue): add auto-ack flag

Signed-off-by: Bo-Yi Wu <[email protected]>

Author: appleboy

options.go

@@ -22,6 +22,7 @@ type options struct {
 	exchangeName string
 	//  Exchange Types: Direct, Fanout, Topic and Headers
 	exchangeType string
+	autoAck      bool
 }
 
 // WithAddr setup the URI
@@ -60,6 +61,13 @@ func WithTag(tag string) Option {
 	}
 }
 
+// WithAutoAck enable message auto-ack
+func WithAutoAck(val bool) Option {
+	return func(w *options) {
+		w.autoAck = val
+	}
+}
+
 // WithSubj setup the topic
 func WithSubj(subj string) Option {
 	return func(w *options) {
@@ -89,6 +97,7 @@ func newOptions(opts ...Option) options {
 		exchangeName: "test-exchange",
 		exchangeType: "direct",
 		logger:       queue.NewLogger(),
+		autoAck:      false,
 		runFunc: func(context.Context, core.QueuedMessage) error {
 			return nil
 		},

rabbitmq.go

@@ -77,13 +77,13 @@ func (w *Worker) startConsumer() (err error) {
 		}
 
 		w.tasks, err = w.channel.Consume(
-			q.Name,     // queue
-			w.opts.tag, // consumer
-			false,      // auto-ack
-			false,      // exclusive
-			false,      // no-local
-			false,      // no-wait
-			nil,        // args
+			q.Name,         // queue
+			w.opts.tag,     // consumer
+			w.opts.autoAck, // auto-ack
+			false,          // exclusive
+			false,          // no-local
+			false,          // no-wait
+			nil,            // args
 		)
 
 		if err != nil {
@@ -153,21 +153,22 @@ func (w *Worker) Run(task core.QueuedMessage) error {
 }
 
 // Shutdown worker
-func (w *Worker) Shutdown() error {
+func (w *Worker) Shutdown() (err error) {
 	if !atomic.CompareAndSwapInt32(&w.stopFlag, 0, 1) {
 		return queue.ErrQueueShutdown
 	}
 
 	w.stopOnce.Do(func() {
 		close(w.stop)
-		if err := w.channel.Cancel(w.opts.tag, true); err != nil {
-			w.opts.logger.Error(err)
+		if err = w.channel.Cancel(w.opts.tag, true); err != nil {
+			w.opts.logger.Error("consumer cancel failed:", err)
 		}
-		if err := w.conn.Close(); err != nil {
-			w.opts.logger.Error(err)
+		if err = w.conn.Close(); err != nil {
+			w.opts.logger.Error("AMQP connection close error:", err)
 		}
 	})
-	return nil
+
+	return err
 }
 
 // Queue send notification to queue
@@ -214,6 +215,9 @@ loop:
 			}
 			var data queue.Job
 			_ = json.Unmarshal(task.Body, &data)
+			if !w.opts.autoAck {
+				task.Ack(w.opts.autoAck)
+			}
 			return &data, nil
 		case <-time.After(1 * time.Second):
 			if clock == 5 {