fix(client): fix blocking/deblocking loop

feat(all): add support for headers with ampq.Table arguments
refact(example): refactor basic example to send continuously messages for reconnection testing

Author: sebastienfr

client.go

@@ -134,20 +134,23 @@ func (c *Client) Loop() bool {
 		conn.NotifyClose(chanErr)
 		conn.NotifyBlocked(chanBlocking)
 
-		select {
-		case err1 := <-chanErr:
-			c.reportErr(err1)
-
-			if conn1 := c.conn.Load().(*amqp.Connection); conn1 != nil {
-				c.conn.Store((*amqp.Connection)(nil))
-				conn1.Close()
-			}
-		case blocking := <-chanBlocking:
+		// loop for blocking/deblocking
+		for {
 			select {
-			case c.blocking <- blocking:
-			default:
+			case err1 := <-chanErr:
+				c.reportErr(err1)
+
+				if conn1 := c.conn.Load().(*amqp.Connection); conn1 != nil {
+					c.conn.Store((*amqp.Connection)(nil))
+					conn1.Close()
+				}
+				// return from routine to launch reconnect process
+				return
+			case blocking := <-chanBlocking:
+				c.blocking <- blocking
 			}
 		}
+
 	}()
 
 	ch, err := conn.Channel()

cony.go

@@ -1,6 +1,6 @@
 // Package cony is a high-level wrapper around http://github.com/streadway/amqp library,
 // for working declaratively with AMQP. Cony will manage AMQP
-// connect/reconnect to AMQP brocker, along with recovery of consumers.
+// connect/reconnect to AMQP broker, along with recovery of consumers.
 package cony
 
 import (
@@ -15,6 +15,7 @@ type Queue struct {
 	Durable    bool
 	AutoDelete bool
 	Exclusive  bool
+	Args       amqp.Table
 
 	l sync.Mutex
 }
@@ -25,13 +26,15 @@ type Exchange struct {
 	Kind       string
 	Durable    bool
 	AutoDelete bool
+	Args       amqp.Table
 }
 
-// Binding used to declare bidning between AMQP Queue and AMQP Exchange
+// Binding used to declare binding between AMQP Queue and AMQP Exchange
 type Binding struct {
 	Queue    *Queue
 	Exchange Exchange
 	Key      string
+	Args     amqp.Table
 }
 
 type mqDeleter interface {

declaration.go

@@ -2,7 +2,7 @@ package cony
 
 import "github.com/streadway/amqp"
 
-// Declaration is a callback type to declare AMQP queue/exchange/bidning
+// Declaration is a callback type to declare AMQP queue/exchange/binding
 type Declaration func(Declarer) error
 
 // Declarer is implemented by *amqp.Channel
@@ -20,7 +20,7 @@ func DeclareQueue(q *Queue) Declaration {
 			q.AutoDelete,
 			q.Exclusive,
 			false,
-			nil,
+			q.Args,
 		)
 		q.l.Lock()
 		q.Name = realQ.Name
@@ -38,19 +38,19 @@ func DeclareExchange(e Exchange) Declaration {
 			e.AutoDelete,
 			false,
 			false,
-			nil,
+			e.Args,
 		)
 	}
 }
 
-// DeclareBinding is a way to declare AMQP bidning between AMQP queue and exchange
+// DeclareBinding is a way to declare AMQP binding between AMQP queue and exchange
 func DeclareBinding(b Binding) Declaration {
 	return func(c Declarer) error {
 		return c.QueueBind(b.Queue.Name,
 			b.Key,
 			b.Exchange.Name,
 			false,
-			nil,
+			b.Args,
 		)
 	}
 }

examples/README.md

@@ -2,8 +2,8 @@
 
 ## Basic
 
-This is the base example ment to show how to use cony.
-The producer will exit after each publish action finishes.
+This is the base example meant to show how to use cony.
+The producer will publish a message every second (or until you stop the program with CTRL-C).
 The consumer will run forever (or until you stop the program with CTRL-C).
 Both applications are CLI-based.
 

examples/basic/consumer/consumer.go

@@ -33,16 +33,17 @@ func main() {
 	// The queue name will be supplied by the AMQP server
 	que := &cony.Queue{
 		AutoDelete: true,
+		Name:       "myQueue",
 	}
 	exc := cony.Exchange{
-		Name:       "basic",
+		Name:       "myExc",
 		Kind:       "fanout",
 		AutoDelete: true,
 	}
 	bnd := cony.Binding{
 		Queue:    que,
 		Exchange: exc,
-		Key:      "",
+		Key:      "pubSub",
 	}
 	cli.Declare([]cony.Declaration{
 		cony.DeclareQueue(que),

examples/basic/producer/producer.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/assembla/cony"
 	"github.com/streadway/amqp"
+	"time"
 )
 
 var url = flag.String("url", "amqp://guest:guest@localhost/", "amqp url")
@@ -35,7 +36,7 @@ func main() {
 
 	// Declare the exchange we'll be using
 	exc := cony.Exchange{
-		Name:       "basic",
+		Name:       "myExc",
 		Kind:       "fanout",
 		AutoDelete: true,
 	}
@@ -45,38 +46,37 @@ func main() {
 
 	// Declare and register a publisher
 	// with the cony client
-	pbl := cony.NewPublisher(exc.Name, "")
+	pbl := cony.NewPublisher(exc.Name, "pubSub")
 	cli.Publish(pbl)
-
-	// Chan used for signaling when
-	// publishing has been completed
-	// so the program can exit
-	done := make(chan struct{})
-
 	// Launch a go routine and publish a message.
 	// "Publish" is a blocking method this is why it
 	// needs to be called in its own go routine.
 	//
 	go func() {
-		pbl.Publish(amqp.Publishing{
-			Body: []byte(*body),
-		})
-		// Close done to signal that the program
-		// can exit
-		close(done)
+		ticker := time.NewTicker(time.Second)
+
+		for {
+			select {
+			case <-ticker.C:
+				fmt.Printf("Client publishing\n")
+				err := pbl.Publish(amqp.Publishing{
+					Body: []byte(*body),
+				})
+				if err != nil {
+					fmt.Printf("Client publish error: %v\n", err)
+				}
+			}
+		}
 	}()
 
 	// Client loop sends out declarations(exchanges, queues, bindings
 	// etc) to the AMQP server. It also handles reconnecting.
-	// We use the done channel here to make sure our publishing is
-	// actually published before the programe exists.
-	// We then close the client to exit the Loop.
 	for cli.Loop() {
 		select {
 		case err := <-cli.Errors():
 			fmt.Printf("Client error: %v\n", err)
-		case <-done:
-			cli.Close()
+		case blocked := <-cli.Blocking():
+			fmt.Printf("Client is blocked %v\n", blocked)
 		}
 	}
 }