func (rp *retryPublisher) Close() error {
	pub, err := rp.publisher(false)
	if err != nil || pub == nil {
		return err
	}
	return pub.Close()
}
Close closes the publisher.
package relay
import (
	"errors"
	"log"
	"sync"
	"time"
	"github.com/armon/relay/broker"
)
var (
	errAttemptsRange = errors.New("Attempts must be between 1 and 32")
	errWaitRange = errors.New("Min/Max wait times must be positive")
)
type retryBroker struct {
	broker   *relayBroker
	attempts int
	min      time.Duration
	max      time.Duration
}
func (r *Relay) RetryBroker(attempts int, min, max time.Duration) (*retryBroker, error) {
	if attempts < 1 || attempts > 32 {
		return nil, errAttemptsRange
	}
	if min == 0 || max == 0 {
		return nil, errWaitRange
	}
	return &retryBroker{&relayBroker{r}, attempts, min, max}, nil
}
func (rb *retryBroker) Publisher(queue string) (broker.Publisher, error) {
	return &retryPublisher{
		broker:   rb.broker,
		queue:    queue,
		attempts: rb.attempts,
		min:      rb.min,
		max:      rb.max,
	}, nil
}
func (rb *retryBroker) Consumer(queue string) (broker.Consumer, error) {
	return &retryConsumer{
		broker:   rb.broker,
		queue:    queue,
		attempts: rb.attempts,
		min:      rb.min,
		max:      rb.max,
	}, nil
}
func (rb *retryBroker) Close() error {
	if rb.broker != nil {
		return rb.broker.Close()
	}
	return nil
}
type retryConsumer struct {
	broker   broker.Broker
	cons     broker.Consumer
	queue    string
	attempts int
	min      time.Duration
	max      time.Duration
	l        sync.RWMutex
}
func (rc *retryConsumer) consumer(create bool) (broker.Consumer, error) {
	rc.l.RLock()
	cons := rc.cons
	rc.l.RUnlock()
	if cons != nil || !create {
		return cons, nil
	}
	rc.l.Lock()
	defer rc.l.Unlock()
	cons, err := rc.broker.Consumer(rc.queue)
	if err != nil {
		return nil, err
	}
	rc.cons = cons
	return cons, nil
}
func (rc *retryConsumer) discard(cons broker.Consumer) {
	rc.l.Lock()
	defer rc.l.Unlock()
	if cons == nil {
		return
	}
	cons.Close()
	if cons == rc.cons {
		rc.cons = nil
	}
}
func (rc *retryConsumer) Close() error {
	cons, err := rc.consumer(false)
	if err != nil || cons == nil {
		return err
	}
	return cons.Close()
}
func (rc *retryConsumer) Ack() error {
	cons, err := rc.consumer(false)
	if err != nil || cons == nil {
		return err
	}
	err = cons.Ack()
	if err != nil {
		log.Printf("[ERR] relay: consumer failed ack: %v", err)
		rc.discard(cons)
	}
	return err
}
func (rc *retryConsumer) Nack() error {
	cons, err := rc.consumer(false)
	if err != nil || cons == nil {
		return err
	}
	err = cons.Nack()
	if err != nil {
		log.Printf("[ERR] relay: consumer failed nack: %v", err)
		rc.discard(cons)
	}
	return err
}
func (rc *retryConsumer) ConsumeAck(out interface{}) error {
	if err := rc.Consume(out); err != nil {
		return err
	}
	return rc.Ack()
}
func (rc *retryConsumer) Consume(out interface{}) error {
	return rc.ConsumeTimeout(out, -1)
}
func (rc *retryConsumer) ConsumeTimeout(out interface{}, timeout time.Duration) error {
	var deadline time.Time
	if timeout >= 0 {
		deadline = time.Now().Add(timeout)
	}
	for i := 0; ; i++ {
		cons, err := rc.consumer(true)
		if err != nil {
			goto RETRY
		}
		err = cons.ConsumeTimeout(out, timeout)
		if err == nil {
			break
		}
		if err == TimedOut {
			return err
		}
	RETRY:
		log.Printf("[WARN] relay: consumer got error: %v", err)
		rc.discard(cons)
		if i == rc.attempts {
			log.Printf("[ERR] relay: consumer giving up after %d attempts", i)
			return err
		}
		now := time.Now()
		if !deadline.IsZero() && now.After(deadline) {
			log.Printf("[DEBUG] relay: consumer reached deadline")
			return TimedOut
		}
		wait := rc.min * time.Duration(int64(1)<<uint(i))
		if wait > rc.max {
			wait = rc.max
		}
		if !deadline.IsZero() && now.Add(wait).After(deadline) {
			wait = deadline.Sub(now)
		}
		timeout -= wait
		log.Printf("[DEBUG] relay: consumer retrying in %s", wait)
		time.Sleep(wait)
	}
	return nil
}
type retryPublisher struct {
	broker   broker.Broker
	pub      broker.Publisher
	queue    string
	attempts int
	min      time.Duration
	max      time.Duration
	l        sync.RWMutex
}
func (rp *retryPublisher) publisher(create bool) (broker.Publisher, error) {
	rp.l.RLock()
	pub := rp.pub
	rp.l.RUnlock()
	if pub != nil || !create {
		return pub, nil
	}
	rp.l.Lock()
	defer rp.l.Unlock()
	pub, err := rp.broker.Publisher(rp.queue)
	if err != nil {
		return nil, err
	}
	rp.pub = pub
	return pub, nil
}
func (rp *retryPublisher) discard(pub broker.Publisher) {
	rp.l.Lock()
	defer rp.l.Unlock()
	if pub == nil {
		return
	}
	pub.Close()
	if pub == rp.pub {
		rp.pub = nil
	}
}