local dev/test env setup for rmq/amqp, build out events/publisher for rabbitMQ, add a subscriber for rabbitMQ

Author: ramin

build/environments/development/rabbit/.gitignore

@@ -0,0 +1 @@
+.data

build/environments/development/rabbit/docker-compose.yml

@@ -0,0 +1,10 @@
+services:
+  rabbitmq:
+    image: rabbitmq:3-management-alpine
+    container_name: 'rabbitmq'
+    ports:
+        - 5672:5672
+        - 15672:15672
+    volumes:
+        - ~/.data/:/var/lib/rabbitmq/
+        - ~/.data/:/var/log/rabbitmq

config/config.go

@@ -89,6 +89,7 @@ type RabbitMQConfig struct {
 	Immediate    bool   `mapstructure:"Immediate"`
 	DeliveryMode int    `mapstructure:"DeliveryMode"`
 	ContentType  string `mapstructure:"ContentType"`
+	AutoAck      bool   `mapstructure:"AutoAck"`
 
 	// Optional
 	Durable    bool `mapstructure:"Durable"`
@@ -183,16 +184,17 @@ func Read() *Config {
 		viper.SetDefault("notifications.EventsQueue", "https://sqs.us-east-1.amazonaws.com/351249512935/demo-queue")
 
 		// Default settings for RabbitMQ
-		viper.SetDefault("rabbitMQ.URL", "amqp://guest:guest@localhost:5672/")
-		viper.SetDefault("rabbitMQ.QueueName", "defaultQueue")
-		viper.SetDefault("rabbitMQ.ExchangeName", "defaultExchange")
-		viper.SetDefault("rabbitMQ.Durable", true)
-		viper.SetDefault("rabbitMQ.AutoDelete", false)
-		viper.SetDefault("rabbitMQ.Exclusive", false)
-		viper.SetDefault("rabbitMQ.NoWait", false)
-		viper.SetDefault("rabbitMQ.Mandatory", false)
-		viper.SetDefault("rabbitMQ.Immediate", false)
-		viper.SetDefault("rabbitMQ.ContentType", "text/plain")
+		viper.SetDefault("rmq.URL", "amqp://guest:guest@localhost:5672/")
+		viper.SetDefault("rmq.QueueName", "defaultQueue")
+		viper.SetDefault("rmq.ExchangeName", "")
+		viper.SetDefault("rmq.AutoAck", true)
+		viper.SetDefault("rmq.Durable", true)
+		viper.SetDefault("rmq.AutoDelete", false)
+		viper.SetDefault("rmq.Exclusive", false)
+		viper.SetDefault("rmq.NoWait", false)
+		viper.SetDefault("rmq.Mandatory", false)
+		viper.SetDefault("rmq.Immediate", false)
+		viper.SetDefault("rmq.ContentType", "text/plain")
 
 		// StreamConfig
 		viper.SetDefault("streams.zeroMQListenAddr", "tcp://127.0.0.1:5558")

events/publishers/rmq/rmq.go

@@ -1,6 +1,8 @@
 package rmq
 
 import (
+	"fmt"
+
 	"github.com/edobtc/cloudkit/config"
 	"github.com/streadway/amqp"
 )
@@ -14,6 +16,8 @@ type Publisher struct {
 func NewPublisher() (*Publisher, error) {
 	queueName := config.Read().RabbitMQ.QueueName
 
+	fmt.Println(config.Read().RabbitMQ.URL)
+
 	conn, err := amqp.Dial(config.Read().RabbitMQ.URL)
 	if err != nil {
 		return nil, err
@@ -43,6 +47,24 @@ func NewPublisher() (*Publisher, error) {
 	}, nil
 }
 
+func (r *Publisher) Listen() (<-chan amqp.Delivery, error) {
+	queueName := config.Read().RabbitMQ.QueueName
+
+	msgs, err := r.channel.Consume(
+		queueName,                        // queue
+		"",                               // consumer
+		true,                             //config.Read().RabbitMQ.AutoAck,  // auto-ack
+		config.Read().RabbitMQ.Exclusive, // exclusive
+		false,                            // no-local
+		config.Read().RabbitMQ.NoWait,    // no-wait
+		nil,                              // args
+	)
+	if err != nil {
+		return nil, err
+	}
+	return msgs, nil
+}
+
 func (r *Publisher) Send(data []byte) error {
 	return r.channel.Publish(
 		config.Read().RabbitMQ.ExchangeName, // exchange

events/subscribers/autoload/autoload.go

@@ -7,12 +7,14 @@ import (
 
 	// conforming implementations
 	eclair "github.com/edobtc/cloudkit/events/subscribers/lightning/eclair/ws"
+	rmq "github.com/edobtc/cloudkit/events/subscribers/rmq"
 	zmq "github.com/edobtc/cloudkit/events/subscribers/zmq"
 )
 
 // Ensure that the Subscriber interface is properly implemented by the autoloaded subscribers
 var _ subscribers.Subscriber = &eclair.EclairSubscriber{}
 var _ subscribers.Subscriber = &zmq.Subscriber{}
+var _ subscribers.Subscriber = &rmq.RMQSubscriber{}
 
 var (
 	ErrAdapterNotFound = errors.New("adapter by name not found")
@@ -24,7 +26,10 @@ func NewSubscriber(adapter string) (subscribers.Subscriber, error) {
 		return eclair.NewSubscriber(), nil
 	case "zmq":
 		return zmq.NewSubscriber(), nil
-
+	case "rmq":
+		// NewSubscriber returns an err along
+		// with the subscriber
+		return rmq.NewSubscriber()
 	default:
 		return nil, ErrAdapterNotFound
 	}

events/subscribers/rmq/rmq.go

@@ -0,0 +1,102 @@
+package rmq
+
+import (
+	"github.com/edobtc/cloudkit/config"
+	"github.com/streadway/amqp"
+)
+
+type RMQSubscriber struct {
+	connection *amqp.Connection
+	channel    *amqp.Channel
+	queueName  string
+}
+
+func NewSubscriber() (*RMQSubscriber, error) {
+	queueName := config.Read().RabbitMQ.QueueName
+
+	conn, err := amqp.Dial(config.Read().RabbitMQ.URL)
+	if err != nil {
+		return nil, err
+	}
+
+	ch, err := conn.Channel()
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = ch.QueueDeclare(
+		queueName,
+		config.Read().RabbitMQ.Durable,
+		config.Read().RabbitMQ.AutoDelete,
+		config.Read().RabbitMQ.Exclusive,
+		config.Read().RabbitMQ.NoWait,
+		nil,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	return &RMQSubscriber{
+		connection: conn,
+		channel:    ch,
+		queueName:  queueName,
+	}, nil
+}
+
+func (r *RMQSubscriber) Start() chan bool {
+	// Start consuming messages
+	msgs, err := r.channel.Consume(
+		r.queueName,
+		"",
+		config.Read().RabbitMQ.AutoAck,
+		config.Read().RabbitMQ.Exclusive,
+		false,
+		config.Read().RabbitMQ.NoWait,
+		nil,
+	)
+	if err != nil {
+		panic(err)
+	}
+
+	done := make(chan bool)
+	go func() {
+		for d := range msgs {
+			// Process message
+			// Placeholder: Print message to console
+			println("Received message: ", string(d.Body))
+		}
+		done <- true
+	}()
+	return done
+}
+
+func (r *RMQSubscriber) Detach() error {
+	if err := r.channel.Close(); err != nil {
+		return err
+	}
+	return r.connection.Close()
+}
+
+func (r *RMQSubscriber) Listen() <-chan interface{} {
+	msgs, err := r.channel.Consume(
+		r.queueName,
+		"",
+		config.Read().RabbitMQ.AutoAck,
+		config.Read().RabbitMQ.Exclusive,
+		false,
+		config.Read().RabbitMQ.NoWait,
+		nil,
+	)
+	if err != nil {
+		panic(err)
+	}
+
+	output := make(chan interface{})
+	go func() {
+		for d := range msgs {
+			output <- d.Body
+		}
+		close(output)
+	}()
+	return output
+}

events/subscribers/subscriber.go

@@ -3,5 +3,10 @@ package subscribers
 type Subscriber interface {
 	Start() chan bool
 	Detach() error
+
+	// we need to replace this return interface{}
+	// with a specific type or a channel that will
+	// return a structured piece of data eventually
+	// (io.Reader or a Proto)
 	Listen() <-chan interface{}
 }