func (r *RabbitMQ) Consume(out chan Message, verify <-chan Verify) {
	go r.ackMultiple(verify)
	deliveries, err := r.channel.Consume(
		r.queue,  
		r.tag,    
		false,    
		false,    
		false,    
		false,    
		nil,      
	)
	if err != nil {
		log.Fatalf("rabbit consumer failed %s", err)
	}
	for d := range deliveries {
		msg := Message{
			Body:        d.Body,
			RoutingKey:  d.RoutingKey,
			Headers:     d.Headers,
			DeliveryTag: d.DeliveryTag,
		}
		out <- msg
	}
	log.Print("All messages consumed")
}
Consume outputs a stream of Message into a channel from rabbit
package rmq
import (
	"log"
	"github.com/streadway/amqp"
)
func NewConsumer(amqpURI, exchange, queue, routingKey, tag string, prefetch int) *RabbitMQ {
	conn, err := amqp.Dial(amqpURI)
	if err != nil {
		log.Fatalf("writer failed to connect to Rabbit: %s", err)
		return nil
	}
	go func() {
		log.Printf("writer closing: %s", <-conn.NotifyClose(make(chan *amqp.Error)))
		log.Printf("writer blocked by rabbit: %v", <-conn.NotifyBlocked(make(chan amqp.Blocking)))
	}()
	channel, err := conn.Channel()
	if err != nil {
		log.Fatalf("writer failed to get a channel from Rabbit: %s", err)
		return nil
	}
	q, err := channel.QueueDeclarePassive(
		queue,  
		true,   
		false,  
		false,  
		false,  
		nil,    
	)
	if err != nil {
		log.Fatalf("Queue Declare: %s", err)
	}
	if q.Messages == 0 {
		log.Fatalf("No messages in RabbitMQ Queue: %s", q.Name)
	}
	if err = channel.QueueBind(
		q.Name,      
		routingKey,  
		exchange,    
		false,       
		nil,         
	); err != nil {
		log.Fatalf("Queue Bind: %s", err)
	}
	r := &RabbitMQ{
		conn:            conn,
		channel:         channel,
		exchange:        exchange,
		contentType:     "application/json",
		contentEncoding: "UTF-8",
	}
	log.Print("RabbitMQ connected: ", amqpURI)
	log.Printf("Bind to Exchange: %q and Queue: %q, Messaging waiting: %d", exchange, queue, q.Messages)
	return r
}
func (r *RabbitMQ) ackMultiple(deliveryTag <-chan Verify) {
	for v := range deliveryTag {
		r.channel.Ack(v.Tag, v.MultiAck)
	}
}