fix: generic pubsub

Author: ochom

makefile

@@ -8,4 +8,4 @@ test:
 
 lint:
 	@echo "Running linter..."
-	@golangci-lint run
+	@golangci-lint run

pubsub/consumer.go

@@ -9,16 +9,13 @@ type Consumer struct {
 	queue    string
 }
 
-// newConsumer ...
-func newConsumer(rabbitURL, queueName string) *Consumer {
-	exchange := fmt.Sprintf("%s-%s-exchange", queuePrefix, queueName)
-	queueName = fmt.Sprintf("%s-%s", queuePrefix, queueName)
-	return &Consumer{rabbitURL, exchange, queueName}
+// Create a new consumer instance
+func NewConsumer(rabbitURL, exchange, queue string) *Consumer {
+	return &Consumer{rabbitURL, exchange, queue}
 }
 
 // Consume consume messages from the channels
-func Consume(queueName string, workerFunc func([]byte)) error {
-	c := newConsumer(rabbitURL, queueName)
+func (c *Consumer) Consume(workerFunc func([]byte)) error {
 	conn, ch, err := initQ(c.url)
 	if err != nil {
 		return fmt.Errorf("failed to initialize a connection: %s", err.Error())

pubsub/publisher.go

@@ -1,7 +1,6 @@
 package pubsub
 
 import (
-	"fmt"
 	"time"
 
 	"github.com/streadway/amqp"
@@ -14,10 +13,9 @@ type publisher struct {
 	queue    string
 }
 
-// newPublisher ...
-func newPublisher(queueName string) *publisher {
-	exchange := fmt.Sprintf("%s-exchange", queueName)
-	return &publisher{rabbitURL, exchange, queueName}
+// NewPublisher  creates a new publisher to rabbit
+func NewPublisher(rabbitURL, exchange, queue string) *publisher {
+	return &publisher{rabbitURL, exchange, queue}
 }
 
 // publish ...
@@ -54,13 +52,11 @@ func (p *publisher) publish(body []byte, delay time.Duration) error {
 }
 
 // PublishWithDelay ...
-func PublishWithDelay(queueName string, body []byte, delay time.Duration) error {
-	p := newPublisher(fmt.Sprintf("%s-%s", queuePrefix, queueName))
+func (p *publisher) PublishWithDelay(queueName string, body []byte, delay time.Duration) error {
 	return p.publish(body, delay)
 }
 
 // Publish ...
-func Publish(queueName string, body []byte) error {
-	p := newPublisher(fmt.Sprintf("%s-%s", queuePrefix, queueName))
+func (p *publisher) Publish(queueName string, body []byte) error {
 	return p.publish(body, 0)
 }

pubsub/pubsub.go

@@ -3,18 +3,9 @@ package pubsub
 import (
 	"fmt"
 
-	"github.com/ochom/gutils/helpers"
 	"github.com/streadway/amqp"
 )
 
-var (
-	// rabbitURL is the URL of the RabbitMQ server
-	rabbitURL = helpers.GetEnv("RABBIT_URL", "amqp://guest:guest@localhost:5672/")
-
-	// queuePrefix used to prefix the queue name to avoid conflict with other services
-	queuePrefix = helpers.GetEnv("QUEUE_PREFIX", "dev")
-)
-
 func initQ(url string) (*amqp.Connection, *amqp.Channel, error) {
 	conn, err := amqp.Dial(url)
 	if err != nil {