resolve #33 (#34)

Added cluster dsn and cluster connection

.gitignore

@@ -1,2 +1,3 @@
 .idea/
 vendor/
+go.sum

README.md

@@ -42,7 +42,7 @@ Supported configuration tags:
 Available options:
 
 ```yaml
-dsn: "amqp://login:password@host:port/virtual_host"
+dsn: "amqp://login:password@host:port/virtual_host" # Use comma separated list for cluster connection
 reconnect_delay: 5s                     # Interval between connection tries. Check https://golang.org/pkg/time/#ParseDuration for details.
 test_mode: false                        # Switches library to use mocked broker. Defaults to false.
 exchanges:

config.go

@@ -1,6 +1,7 @@
 package mq
 
 import (
+	"strings"
 	"time"
 
 	"github.com/streadway/amqp"
@@ -24,14 +25,21 @@ type Config struct {
 	Queues         Queues        `mapstructure:"queues" json:"queues" yaml:"queues"`
 	Producers      Producers     `mapstructure:"producers" json:"producers" yaml:"producers"`
 	Consumers      Consumers     `mapstructure:"consumers" json:"consumers" yaml:"consumers"`
+	dsnList        []string
 }
 
 // Traverses the config tree and fixes option keys name.
-func (config Config) normalize() {
+func (config *Config) normalize() {
 	config.Exchanges.normalize()
 	config.Queues.normalize()
 	config.Producers.normalize()
 	config.Consumers.normalize()
+	config.dsnClusterParse()
+}
+
+// Parse DSN to cluster list
+func (config *Config) dsnClusterParse() {
+	config.dsnList = strings.Split(config.DSN, ",")
 }
 
 // Exchanges describes configuration list for exchanges.

mq.go

@@ -4,6 +4,7 @@ package mq
 import (
 	"fmt"
 	"net"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -55,6 +56,10 @@ type mq struct {
 	consumers            *consumersRegistry
 	producers            *producersRegistry
 	reconnectStatus      int32 // Defines whether client is trying to reconnect or not.
+	cluster              struct {
+		sync.Once
+		currentNode int32
+	}
 }
 
 // New initializes AMQP connection to the message broker
@@ -169,11 +174,18 @@ func (mq *mq) connect() error {
 }
 
 func (mq *mq) createConnection() (conn, error) {
+	mq.cluster.Do(func() { atomic.StoreInt32(&mq.cluster.currentNode, -1) })
+	atomic.AddInt32(&mq.cluster.currentNode, 1)
+	if int(mq.cluster.currentNode) >= len(mq.config.dsnList) {
+		atomic.StoreInt32(&mq.cluster.currentNode, 0)
+	}
+	dsn := mq.config.dsnList[mq.cluster.currentNode]
+
 	if brokerIsMocked || mq.config.TestMode {
-		return amqptest.Dial(mq.config.DSN)
+		return amqptest.Dial(dsn)
 	}
 
-	return amqp.Dial(mq.config.DSN)
+	return amqp.Dial(dsn)
 }
 
 // Register close handler.

mq_test.go

@@ -493,6 +493,43 @@ func TestMq_New_InvalidConsumerConfiguration(t *testing.T) {
 	}
 }
 
+func Test_mq_createConnection(t *testing.T) {
+	cases := []struct {
+		name      string
+		dsn       string
+		checkList []string
+	}{
+		{name: "single dsn", dsn: dsnForTests, checkList: []string{dsnForTests}},
+		{name: "cluster dsn", dsn: "amqp://guest:guest@localhost:5672/1,amqp://guest:guest@localhost:5672/2,amqp://guest:guest@localhost:5672/3",
+			checkList: []string{"amqp://guest:guest@localhost:5672/1", "amqp://guest:guest@localhost:5672/2", "amqp://guest:guest@localhost:5672/3", "amqp://guest:guest@localhost:5672/1"}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			cfg := newDefaultConfig()
+			cfg.DSN = tc.dsn
+			cfg.TestMode = true
+			cfg.normalize()
+
+			mq := &mq{
+				config:               cfg,
+				errorChannel:         make(chan error),
+				internalErrorChannel: make(chan error),
+				consumers:            newConsumersRegistry(len(cfg.Consumers)),
+				producers:            newProducersRegistry(len(cfg.Producers)),
+			}
+			defer mq.Close()
+			for _, dsn := range tc.checkList {
+				_, _ = mq.createConnection()
+				curBroker := mq.config.dsnList[mq.cluster.currentNode]
+				if curBroker != dsn {
+					t.Errorf("createConnection() current broker %v, expected broker %v", curBroker, dsn)
+					return
+				}
+			}
+		})
+	}
+}
+
 func assertNoMqError(t *testing.T, mq MQ) {
 	select {
 	case err := <-mq.Error():