chore(queue): support custom consumer name

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

Author: appleboy

options.go

@@ -15,16 +15,24 @@ type options struct {
 	logger  queue.Logger
 	addr    string
 	subj    string
+	tag     string
 }
 
-// WithAddr setup the addr of NATS
+// WithAddr setup the URI
 func WithAddr(addr string) Option {
 	return func(w *options) {
 		w.addr = "nats://" + addr
 	}
 }
 
-// WithSubj setup the subject of NATS
+// WithAddr setup the tag
+func WithTag(tag string) Option {
+	return func(w *options) {
+		w.tag = tag
+	}
+}
+
+// WithSubj setup the topic
 func WithSubj(subj string) Option {
 	return func(w *options) {
 		w.subj = subj
@@ -49,6 +57,7 @@ func newOptions(opts ...Option) options {
 	defaultOpts := options{
 		addr:   "amqp://guest:guest@localhost:5672/",
 		subj:   "queue",
+		tag:    "golang-queue",
 		logger: queue.NewLogger(),
 		runFunc: func(context.Context, core.QueuedMessage) error {
 			return nil

rabbitmq.go

@@ -17,7 +17,7 @@ var _ core.Worker = (*Worker)(nil)
 
 // Worker for NSQ
 type Worker struct {
-	client    *amqp.Connection
+	conn      *amqp.Connection
 	channel   *amqp.Channel
 	stop      chan struct{}
 	stopFlag  int32
@@ -36,12 +36,12 @@ func NewWorker(opts ...Option) *Worker {
 		tasks: make(chan amqp.Delivery),
 	}
 
-	w.client, err = amqp.Dial(w.opts.addr)
+	w.conn, err = amqp.Dial(w.opts.addr)
 	if err != nil {
 		panic(err)
 	}
 
-	w.channel, err = w.client.Channel()
+	w.channel, err = w.conn.Channel()
 	if err != nil {
 		panic(err)
 	}
@@ -66,13 +66,13 @@ func (w *Worker) startConsumer() (err error) {
 		}
 
 		w.tasks, err = w.channel.Consume(
-			q.Name, // queue
-			"",     // consumer
-			true,   // auto-ack
-			false,  // exclusive
-			false,  // no-local
-			false,  // no-wait
-			nil,    // args
+			q.Name,     // queue
+			w.opts.tag, // consumer
+			true,       // auto-ack
+			false,      // exclusive
+			false,      // no-local
+			false,      // no-wait
+			nil,        // args
 		)
 
 		if err != nil {
@@ -152,7 +152,7 @@ func (w *Worker) Shutdown() error {
 		if err := w.channel.Cancel("", true); err != nil {
 			w.opts.logger.Error(err)
 		}
-		if err := w.client.Close(); err != nil {
+		if err := w.conn.Close(); err != nil {
 			w.opts.logger.Error(err)
 		}
 	})