fix: generic pubsub

Author: ochom

pubsub/consumer.go

@@ -4,19 +4,15 @@ import "fmt"
 
 // Consumer ...
 type Consumer struct {
-	url         string
-	exchange    string
-	queue       string
-	enableDelay bool
+	url          string
+	exchange     string
+	queue        string
+	exchangeType string
 }
 
 // Create a new consumer instance
-func NewConsumer(rabbitURL, exchange, queue string, config ...Config) *Consumer {
-	if len(config) > 0 {
-		return &Consumer{rabbitURL, exchange, queue, config[0].EnableDelay}
-	}
-
-	return &Consumer{rabbitURL, exchange, queue, false}
+func NewConsumer(rabbitURL, exchange, queue string, exchangeType ExchangeType) *Consumer {
+	return &Consumer{rabbitURL, exchange, queue, string(exchangeType)}
 }
 
 // Consume consume messages from the channels
@@ -28,7 +24,7 @@ func (c *Consumer) Consume(workerFunc func([]byte)) error {
 	defer ch.Close()
 	defer conn.Close()
 
-	if err := initPubSub(ch, c.exchange, c.queue, c.enableDelay); err != nil {
+	if err := initPubSub(ch, c.exchange, c.queue, c.exchangeType); err != nil {
 		return fmt.Errorf("failed to initialize a pubsub: %s", err.Error())
 	}
 

pubsub/publisher.go

@@ -9,19 +9,15 @@ import (
 
 // publisher ...
 type publisher struct {
-	url         string
-	exchange    string
-	queue       string
-	enableDelay bool
+	url          string
+	exchange     string
+	queue        string
+	exchangeType string
 }
 
 // NewPublisher  creates a new publisher to rabbit
-func NewPublisher(rabbitURL, exchange, queue string, config ...Config) *publisher {
-	if len(config) > 0 {
-		return &publisher{rabbitURL, exchange, queue, config[0].EnableDelay}
-	}
-
-	return &publisher{rabbitURL, exchange, queue, false}
+func NewPublisher(rabbitURL, exchange, queue string, exchangeType ExchangeType) *publisher {
+	return &publisher{rabbitURL, exchange, queue, string(exchangeType)}
 }
 
 // publish ...
@@ -33,7 +29,7 @@ func (p *publisher) publish(body []byte, delay time.Duration) error {
 	defer ch.Close()
 	defer conn.Close()
 
-	if err := initPubSub(ch, p.exchange, p.queue, p.enableDelay); err != nil {
+	if err := initPubSub(ch, p.exchange, p.queue, p.exchangeType); err != nil {
 		return fmt.Errorf("failed to initialize a pubsub: %s", err.Error())
 	}
 

pubsub/pubsub.go

@@ -6,10 +6,16 @@ import (
 	"github.com/streadway/amqp"
 )
 
-// Config  for publisher and consumer
-type Config struct {
-	EnableDelay bool
-}
+// ExchangeType ...
+type ExchangeType string
+
+var (
+	Direct  ExchangeType = "direct"
+	Topic   ExchangeType = "topic"
+	Fanout  ExchangeType = "fanout"
+	Headers ExchangeType = "headers"
+	Delayed ExchangeType = "x-delayed-message"
+)
 
 func initQ(url string) (*amqp.Connection, *amqp.Channel, error) {
 	conn, err := amqp.Dial(url)
@@ -31,12 +37,7 @@ func initQ(url string) (*amqp.Connection, *amqp.Channel, error) {
 }
 
 // initPubSub ...
-func initPubSub(ch *amqp.Channel, exchangeName, queueName string, enableDelay bool) error {
-	exchangeType := "direct"
-	if enableDelay {
-		exchangeType = "x-delayed-message"
-	}
-
+func initPubSub(ch *amqp.Channel, exchangeName, queueName, exchangeType string) error {
 	err := ch.ExchangeDeclare(
 		exchangeName, // name
 		exchangeType, // type