chore(queue): add routing key

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

Author: appleboy

options.go

@@ -23,6 +23,8 @@ type options struct {
 	//  Exchange Types: Direct, Fanout, Topic and Headers
 	exchangeType string
 	autoAck      bool
+	// AMQP routing key
+	routingKey string
 }
 
 // WithAddr setup the URI
@@ -54,10 +56,17 @@ func WithExchangeType(val string) Option {
 	}
 }
 
+// WithRoutingKey setup  AMQP routing key
+func WithRoutingKey(val string) Option {
+	return func(w *options) {
+		w.routingKey = val
+	}
+}
+
 // WithAddr setup the tag
-func WithTag(tag string) Option {
+func WithTag(val string) Option {
 	return func(w *options) {
-		w.tag = tag
+		w.tag = val
 	}
 }
 
@@ -69,9 +78,9 @@ func WithAutoAck(val bool) Option {
 }
 
 // WithSubj setup the topic
-func WithSubj(subj string) Option {
+func WithSubj(val string) Option {
 	return func(w *options) {
-		w.subj = subj
+		w.subj = val
 	}
 }
 
@@ -96,6 +105,7 @@ func newOptions(opts ...Option) options {
 		tag:          "golang-queue",
 		exchangeName: "test-exchange",
 		exchangeType: "direct",
+		routingKey:   "test-key",
 		logger:       queue.NewLogger(),
 		autoAck:      false,
 		runFunc: func(context.Context, core.QueuedMessage) error {

rabbitmq.go

@@ -177,26 +177,18 @@ func (w *Worker) Queue(job core.QueuedMessage) error {
 		return queue.ErrQueueShutdown
 	}
 
-	q, err := w.channel.QueueDeclare(
-		w.opts.subj, // name
-		true,        // durable
-		false,       // delete when unused
-		false,       // exclusive
-		false,       // no-wait
-		nil,         // arguments
-	)
-	if err != nil {
-		return err
-	}
-
-	err = w.channel.Publish(
-		"",     // exchange
-		q.Name, // routing key
-		false,  // mandatory
-		false,  // immediate
+	err := w.channel.Publish(
+		w.opts.exchangeName, // exchange
+		w.opts.routingKey,   // routing key
+		false,               // mandatory
+		false,               // immediate
 		amqp.Publishing{
-			ContentType: "text/plain",
-			Body:        job.Bytes(),
+			Headers:         amqp.Table{},
+			ContentType:     "text/plain",
+			ContentEncoding: "",
+			Body:            job.Bytes(),
+			DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent
+			Priority:        0,              // 0-9
 		})
 
 	return err