chore(queue): "add exchange name and type default is direct"

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

Author: appleboy

options.go

@@ -10,12 +10,18 @@ import (
 // Option for queue system
 type Option func(*options)
 
+// AMQP 0-9-1 Model Explained
+// ref: https://www.rabbitmq.com/tutorials/amqp-concepts.html
 type options struct {
 	runFunc func(context.Context, core.QueuedMessage) error
 	logger  queue.Logger
 	addr    string
 	subj    string
 	tag     string
+	// Durable AMQP exchange name
+	exchangeName string
+	//  Exchange Types: Direct, Fanout, Topic and Headers
+	exchangeType string
 }
 
 // WithAddr setup the URI
@@ -25,6 +31,28 @@ func WithAddr(addr string) Option {
 	}
 }
 
+// WithExchangeName setup the Exchange name
+// Exchanges are AMQP 0-9-1 entities where messages are sent to.
+// Exchanges take a message and route it into zero or more queues.
+func WithExchangeName(val string) Option {
+	return func(w *options) {
+		w.exchangeName = val
+	}
+}
+
+// WithExchangeType setup the Exchange type
+// The routing algorithm used depends on the exchange type and rules called bindings.
+// AMQP 0-9-1 brokers provide four exchange types:
+// Direct exchange	(Empty string) and amq.direct
+// Fanout exchange	amq.fanout
+// Topic exchange	amq.topic
+// Headers exchange	amq.match (and amq.headers in RabbitMQ)
+func WithExchangeType(val string) Option {
+	return func(w *options) {
+		w.exchangeType = val
+	}
+}
+
 // WithAddr setup the tag
 func WithTag(tag string) Option {
 	return func(w *options) {
@@ -55,10 +83,12 @@ func WithLogger(l queue.Logger) Option {
 
 func newOptions(opts ...Option) options {
 	defaultOpts := options{
-		addr:   "amqp://guest:guest@localhost:5672/",
-		subj:   "queue",
-		tag:    "golang-queue",
-		logger: queue.NewLogger(),
+		addr:         "amqp://guest:guest@localhost:5672/",
+		subj:         "golang-queue",
+		tag:          "golang-queue",
+		exchangeName: "test-exchange",
+		exchangeType: "direct",
+		logger:       queue.NewLogger(),
 		runFunc: func(context.Context, core.QueuedMessage) error {
 			return nil
 		},

rabbitmq.go

@@ -46,6 +46,18 @@ func NewWorker(opts ...Option) *Worker {
 		panic(err)
 	}
 
+	if err := w.channel.ExchangeDeclare(
+		w.opts.exchangeName, // name
+		w.opts.exchangeType, // type
+		true,                // durable
+		false,               // auto-deleted
+		false,               // internal
+		false,               // noWait
+		nil,                 // arguments
+	); err != nil {
+		panic(err)
+	}
+
 	return w
 }
 
@@ -54,7 +66,7 @@ func (w *Worker) startConsumer() (err error) {
 		var err error
 		q, err := w.channel.QueueDeclare(
 			w.opts.subj, // name
-			false,       // durable
+			true,        // durable
 			false,       // delete when unused
 			false,       // exclusive
 			false,       // no-wait
@@ -167,7 +179,7 @@ func (w *Worker) Queue(job core.QueuedMessage) error {
 
 	q, err := w.channel.QueueDeclare(
 		w.opts.subj, // name
-		false,       // durable
+		true,        // durable
 		false,       // delete when unused
 		false,       // exclusive
 		false,       // no-wait