func (c *consumer) FactoryName() string {
	return c.factoryName
}
FactoryName is the name of the factory responsible for this consumer.
package rabbit
import (
	"context"
	"errors"
	"time"
	"gopkg.in/tomb.v2"
	"github.com/leandro-lugaresi/hub"
	"github.com/leandro-lugaresi/message-cannon/runner"
	"github.com/streadway/amqp"
)
type consumer struct {
	runner      runner.Runnable
	hash        string
	name        string
	queue       string
	workerPool  pool
	timeout     time.Duration
	factoryName string
	opts        Options
	channel     *amqp.Channel
	t           tomb.Tomb
	hub         *hub.Hub
}
func (c *consumer) Run() {
	c.t.Go(func() error {
		defer func() {
			err := c.channel.Close()
			if err != nil {
				c.hub.Publish(hub.Message{
					Name:   "rabbit.consumer.error",
					Body:   []byte("Error closing the consumer channel"),
					Fields: hub.Fields{"error": err},
				})
			}
		}()
		d, err := c.channel.Consume(c.queue, "rabbitmq-"+c.name+"-"+c.hash,
			c.opts.AutoAck,
			c.opts.Exclusive,
			c.opts.NoLocal,
			c.opts.NoWait,
			c.opts.Args)
		if err != nil {
			c.hub.Publish(hub.Message{
				Name:   "rabbit.consumer.error",
				Body:   []byte("Failed to start consume"),
				Fields: hub.Fields{"error": err},
			})
			return err
		}
		dying := c.t.Dying()
		closed := c.channel.NotifyClose(make(chan *amqp.Error))
		ctx, cancel := context.WithCancel(context.Background())
		defer cancel()
		for {
			select {
			case <-dying:
				c.workerPool.Wait()
				return nil
			case err := <-closed:
				return err
			case msg, ok := <-d:
				if !ok {
					c.hub.Publish(hub.Message{
						Name:   "rabbit.consumer.error",
						Body:   []byte("receive an empty delivery. closing consumer"),
						Fields: hub.Fields{},
					})
					return errors.New("receive an empty delivery")
				}
				c.workerPool.Acquire()
				go func(msg amqp.Delivery) {
					nctx := ctx
					if c.timeout >= time.Second {
						var canc context.CancelFunc
						nctx, canc = context.WithTimeout(ctx, c.timeout)
						defer canc()
					}
					c.processMessage(nctx, msg)
					c.workerPool.Release()
				}(msg)
			}
		}
	})
}
func (c *consumer) Kill() {
	c.t.Kill(nil)
	<-c.t.Dead()
}
func (c *consumer) Alive() bool {
	return c.t.Alive()
}
func (c *consumer) Name() string {
	return c.name
}