func (c *Consumer) ConsumeTimeout(out interface{}, timeout time.Duration) error {
	if c.channel == nil {
		return ChannelClosed
	}
	if c.needAck && !c.conf.EnableMultiAck {
		return fmt.Errorf("Ack required before consume!")
	}
	if c.conf.EnableMultiAck && c.numNoAck >= c.conf.PrefetchCount {
		return fmt.Errorf("Consume will block without Ack!")
	}
	var wait <-chan time.Time
	if timeout >= 0 {
		wait = time.After(timeout)
	}
	var d amqp.Delivery
	var ok bool
	select {
	case d, ok = <-c.deliverChan:
		if !ok {
			return ChannelClosed
		}
	case <-wait:
		return TimedOut
	}
	c.lastMsg = d.DeliveryTag
	c.needAck = true
	c.numNoAck++
	buf := bytes.NewBuffer(d.Body)
	if err := c.conf.Serializer.RelayDecode(buf, out); err != nil {
		c.Nack()
		return fmt.Errorf("Failed to decode message! Got: %s", err)
	}
	return nil
}
Consume will consume the next available message or times out waiting. The
message must be acknowledged with Ack() or Nack() before
the next call to Consume unless EnableMultiAck is true.
package relay
import (
	"bytes"
	"fmt"
	"github.com/streadway/amqp"
	"strings"
	"time"
)
type Consumer struct {
	conf        *Config
	consName    string
	queue       string
	channel     *amqp.Channel
	deliverChan <-chan amqp.Delivery
	lastMsg     uint64  
	numNoAck    int     
	needAck     bool
}