Merge branch 'release/1.0.0'

.gitignore

@@ -1 +1,3 @@
-vendor
+vendor
+cmd
+bin

README.md

@@ -1,6 +1,7 @@
 # Hop
 
 [![GoDoc](https://godoc.org/github.com/mazingstudio/hop?status.svg)](https://godoc.org/github.com/mazingstudio/hop)
+[![Go Report Card](https://goreportcard.com/badge/github.com/mazingstudio/hop)](https://goreportcard.com/report/github.com/mazingstudio/hop)
 [![license](https://img.shields.io/github/license/mashape/apistatus.svg)]()
 
 _An AMQP client wrapper that provides easy work queue semantics_
@@ -46,19 +47,37 @@ func main() {
 		log.Fatalf("error getting topic: %s", err)
 	}
 
-	// Put a job into the "tasks" Topic
+	// Put a Job into the "tasks" Topic
 	err = tasks.Put([]byte("Hello"))
 	if err != nil {
 		log.Fatalf("error putting: %s", err)
 	}
 
-	// Pull the job from the Topic
+	// Pull the Job from the Topic
 	hello, err := tasks.Pull()
 	if err != nil {
 		log.Fatalf("error pulling: %s", err)
 	}
-
 	// This should print "hello"
 	log.Infof("job body: %s", hello.Body())
+
+	// Mark the Job as failed and requeue
+	hello.Fail(true)
+
+	// Pull the Job again
+	hello2, err := tasks.Pull()
+	if err != nil {
+		log.Fatalf("error pulling: %s", err)
+	}
+	log.Infof("job body: %s", hello.Body())
+
+	// Mark the Job as done
+	hello2.Done()
 }
-```
+```
+
+## License & Third Party Code
+
+Hop uses [`streadway/amqp`](https://github.com/streadway/amqp) internally.
+
+Hop is distributed under the MIT License. Please refer to `LICENSE` for more details.

config.go

@@ -1,5 +1,11 @@
 package hop
 
+import (
+	"time"
+
+	"github.com/streadway/amqp"
+)
+
 // Config represents a WorkQueue configuration.
 type Config struct {
 	// ExchangeName is used when naming the underlying exchange for the work
@@ -10,9 +16,19 @@ type Config struct {
 	// Persist indicates whether the underlying queues and messages should be
 	// saved to disk to survive server restarts and crashes
 	Persistent bool
+	// MaxConnectionRetry indicates for how long dialing should be retried
+	// during connection recovery
+	MaxConnectionRetry time.Duration
+	// AMQPConfig is optionally used to set up the internal connection to the
+	// AMQP broker. If nil, default values (determined by the underlying
+	// library) are used.
+	AMQPConfig *amqp.Config
 }
 
+// DefaultConfig are the default values used when initializing the default
+// queue.
 var DefaultConfig = &Config{
-	ExchangeName: "hop.exchange",
-	Persistent:   false,
+	ExchangeName:       "hop.exchange",
+	Persistent:         false,
+	MaxConnectionRetry: 15 * time.Minute,
 }

connection.go

@@ -0,0 +1,74 @@
+package hop
+
+import (
+	"sync"
+
+	"github.com/cenkalti/backoff"
+	"github.com/pkg/errors"
+	"github.com/streadway/amqp"
+)
+
+type connection struct {
+	addr   string
+	config *Config
+	conn   *amqp.Connection
+	mu     sync.Mutex
+}
+
+func newConnection(addr string, config *Config) *connection {
+	return &connection{
+		addr:   addr,
+		config: config,
+	}
+}
+
+func (c *connection) connect() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	conn, err := c.dialWithBackOff()
+	if err != nil {
+		return errors.Wrap(err, "queue connection failed permanently")
+	}
+	c.conn = conn
+	return nil
+}
+
+func (c *connection) dialWithBackOff() (*amqp.Connection, error) {
+	var conn *amqp.Connection
+	connect := func() error {
+		var err error
+		if c.config.AMQPConfig != nil {
+			conn, err = amqp.DialConfig(c.addr, *c.config.AMQPConfig)
+			return err
+		}
+		conn, err = amqp.Dial(c.addr)
+		return err
+	}
+	b := backoff.NewExponentialBackOff()
+	b.MaxElapsedTime = c.config.MaxConnectionRetry
+	err := backoff.Retry(connect, b)
+	if err != nil {
+		return nil, errors.Wrap(err, "error dialing")
+	}
+	return conn, nil
+}
+
+func (c *connection) newChannel() interface{} {
+	var (
+		ch  *amqp.Channel
+		err error
+	)
+	for err == nil {
+		ch, err = c.conn.Channel()
+		if err != nil {
+			err = c.connect()
+			continue
+		}
+		return ch
+	}
+	return errors.Wrap(err, "failed to initialize channel")
+}
+
+func (c *connection) Close() error {
+	return c.conn.Close()
+}

job.go

@@ -9,13 +9,13 @@ import (
 type Job struct {
 	topic *Topic
 	del   *amqp.Delivery
+	ch    *amqp.Channel
 }
 
 // Done marks the Job for queue removal.
 func (j *Job) Done() error {
-	// We ack directly on the parent channel, in case we need to migrate
-	// to a different one, in the case of a channel failure
-	err := j.topic.ch.Ack(j.del.DeliveryTag, false)
+	defer j.topic.queue.putChannel(j.ch)
+	err := j.ch.Ack(j.del.DeliveryTag, false)
 	if err != nil {
 		return errors.Wrap(err, "error acknowledging delivery")
 	}
@@ -25,7 +25,8 @@ func (j *Job) Done() error {
 // Fail marks the Job as failed. If requeue is true, the Job will be added
 // back to the queue; otherwise, it will be dropped.
 func (j *Job) Fail(requeue bool) error {
-	err := j.topic.ch.Reject(j.del.DeliveryTag, requeue)
+	defer j.topic.queue.putChannel(j.ch)
+	err := j.ch.Nack(j.del.DeliveryTag, false, requeue)
 	if err != nil {
 		return errors.Wrap(err, "error rejecting delivery")
 	}
@@ -38,7 +39,7 @@ func (j *Job) Body() []byte {
 }
 
 // Delivery returns the Job's underlying delivery. Use this if you need more
-// control over the AMQP messages.
+// control over the AMQP message.
 func (j *Job) Delivery() *amqp.Delivery {
 	return j.del
 }

queue.go

@@ -1,15 +1,18 @@
 package hop
 
 import (
+	"sync"
+
 	"github.com/pkg/errors"
 	"github.com/streadway/amqp"
 )
 
 // WorkQueue is an abstraction over an AMQP connection that automatically
 // performs management and configuration to provide easy work queue semantics.
 type WorkQueue struct {
-	conn   *amqp.Connection
-	config *Config
+	conn        *connection
+	config      *Config
+	channelPool sync.Pool
 }
 
 const exchangeKind = "direct"
@@ -21,23 +24,23 @@ func DefaultQueue(addr string) (*WorkQueue, error) {
 }
 
 // ConnectQueue dials using the provided address string and creates a queue
-// with the passed configuration. If you need to manage your own connection,
-// use NewQueue.
+// with the passed configuration.
 func ConnectQueue(addr string, config *Config) (*WorkQueue, error) {
-	conn, err := amqp.Dial(addr)
+	conn := newConnection(addr, config)
+	err := conn.connect()
 	if err != nil {
-		return nil, errors.Wrap(err, "error dialing")
+		return nil, errors.Wrap(err, "error connecting")
 	}
-	return NewQueue(conn, config)
+	return newQueue(conn, config)
 }
 
-// NewQueue allows you to manually construct a WorkQueue with the provided
-// parameters. Due to the slight verbosity of this process, DefaultQueue and
-// ConnectQueue are generally recommended instead.
-func NewQueue(conn *amqp.Connection, config *Config) (*WorkQueue, error) {
+func newQueue(conn *connection, config *Config) (*WorkQueue, error) {
 	q := &WorkQueue{
 		conn:   conn,
 		config: config,
+		channelPool: sync.Pool{
+			New: conn.newChannel,
+		},
 	}
 	if err := q.init(); err != nil {
 		return nil, errors.Wrap(err, "failed to initialize queue")
@@ -46,26 +49,42 @@ func NewQueue(conn *amqp.Connection, config *Config) (*WorkQueue, error) {
 }
 
 func (q *WorkQueue) init() error {
-	ch, err := q.conn.Channel()
+	ch, err := q.getChannel()
 	if err != nil {
 		return errors.Wrap(err, "error getting management channel")
 	}
-	defer ch.Close()
+	defer q.putChannel(ch)
 
 	err = ch.ExchangeDeclare(q.config.ExchangeName, exchangeKind,
-		false, false, false, false, nil)
+		q.config.Persistent, false, false, false, nil)
 	if err != nil {
 		return errors.Wrap(err, "error declaring exchange")
 	}
 	return nil
 }
 
-// Close gracefully closes the connection to the message broker. DO NOT use this
-// if you're managing your AMQP connection manually.
+// Close gracefully closes the connection to the message broker.
 func (q *WorkQueue) Close() error {
 	err := q.conn.Close()
 	if err != nil {
 		return errors.Wrap(err, "error closing connection")
 	}
 	return nil
 }
+
+// --- Channels ---
+
+func (q *WorkQueue) getChannel() (*amqp.Channel, error) {
+	res := q.channelPool.Get()
+	switch v := res.(type) {
+	case error:
+		return nil, errors.Wrap(v, "channel retrieval failed permanently")
+	case *amqp.Channel:
+		return v, nil
+	}
+	return nil, nil
+}
+
+func (q *WorkQueue) putChannel(ch *amqp.Channel) {
+	q.channelPool.Put(ch)
+}