func (q *AMQP) Connect() error {
	q.Lock()
	defer q.Unlock()

	q.headers = amqp.Table{
		"precision":        q.Precision,
		"database":         q.Database,
		"retention_policy": q.RetentionPolicy,
	}

	var connection *amqp.Connection
	// make new tls config
	tls, err := internal.GetTLSConfig(
		q.SSLCert, q.SSLKey, q.SSLCA, q.InsecureSkipVerify)
	if err != nil {
		return err
	}

	if tls != nil {
		connection, err = amqp.DialTLS(q.URL, tls)
	} else {
		connection, err = amqp.Dial(q.URL)
	}
	if err != nil {
		return err
	}
	channel, err := connection.Channel()
	if err != nil {
		return fmt.Errorf("Failed to open a channel: %s", err)
	}

	err = channel.ExchangeDeclare(
		q.Exchange, // name
		"topic",    // type
		true,       // durable
		false,      // delete when unused
		false,      // internal
		false,      // no-wait
		nil,        // arguments
	)
	if err != nil {
		return fmt.Errorf("Failed to declare an exchange: %s", err)
	}
	q.channel = channel
	go func() {
		log.Printf("Closing: %s", <-connection.NotifyClose(make(chan *amqp.Error)))
		log.Printf("Trying to reconnect")
		for err := q.Connect(); err != nil; err = q.Connect() {
			log.Println(err)
			time.Sleep(10 * time.Second)
		}

	}()
	return nil
}

//NewServerCodec returns a new rpc.ClientCodec using AMQP on conn. serverRouting is the routing
//key with with RPC calls are received, encodingCodec is an EncodingCoding implementation. This package provdes JSONCodec and GobCodec for the JSON and Gob encodings respectively.
func NewServerCodec(conn *amqp.Connection, serverRouting string, encodingCodec EncodingCodec) (rpc.ServerCodec, error) {
	channel, err := conn.Channel()
	if err != nil {
		return nil, err
	}

	queue, err := channel.QueueDeclare(serverRouting, false, false, false, false, nil)
	if err != nil {
		return nil, err
	}

	messages, err := channel.Consume(queue.Name, "", true, false, false, false, nil)
	if err != nil {
		return nil, err
	}

	server := &serverCodec{
		codec: &codec{
			conn:    conn,
			channel: channel,
			routing: queue.Name,

			codec:   encodingCodec,
			message: messages,
		},
		lock:  new(sync.RWMutex),
		calls: make(map[uint64]route),
		seq:   0,
	}

	return server, err
}

func (server *Server) ServeConn(conn *amqp.Connection, queue string) {
	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	q, err := ch.QueueDeclare(
		queue, // name
		false, // durable
		false, // delete when usused
		false, // exclusive
		false, // noWait
		nil,   // arguments
	)
	failOnError(err, "Failed to declare a queue")
	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer
		true,   // autoAck
		false,  // exclusive
		false,  // noLocal
		false,  // noWait
		nil,    // arguments
	)
	failOnError(err, "Failed to register a consumer")
	codec := &serverCodec{
		conn:    conn,
		ch:      ch,
		msgs:    msgs,
		pending: make(map[uint64]prop),
	}
	server.s.ServeCodec(codec)
}

//NewClientCodec returns a new rpc.ClientCodec using AMQP on conn. serverRouting is the routing
//key with with RPC calls are sent, it should be the same routing key used with NewServerCodec.
//encodingCodec is an EncodingCoding implementation. This package provdes JSONCodec and GobCodec
//for the JSON and Gob encodings respectively.
func NewClientCodec(conn *amqp.Connection, serverRouting string, encodingCodec EncodingCodec) (rpc.ClientCodec, error) {
	channel, err := conn.Channel()
	if err != nil {
		return nil, err
	}

	queue, err := channel.QueueDeclare("", false, true, false, false, nil)
	if err != nil {
		return nil, err
	}

	serverQueue, err := channel.QueueDeclare(serverRouting, false, false, false, false, nil)
	if err != nil {
		return nil, err
	}
	if serverQueue.Consumers == 0 {
		return nil, ErrNoConsumers
	}

	message, err := channel.Consume(queue.Name, "", true, false, false, false, nil)
	client := &clientCodec{
		codec: &codec{
			conn:    conn,
			channel: channel,
			routing: queue.Name,
			codec:   encodingCodec,
			message: message,
		},

		serverRouting: serverRouting,
	}

	return client, err
}

func monitoring(uri string, queueName string, mgmtUri string, prefix string) {
	var (
		queueConn *amqp.Connection
		queueChan *amqp.Channel
		err       error
	)
	queueConn, queueChan, err = rabbitmqConnect(uri, queueName)
	if err != nil {
		return
	}
	for {
		log.Printf("fetch rabbitmq stats")
		var metrics []graphite.Metric
		for _, metric := range fetchQueueMetrics(mgmtUri, prefix) {
			metrics = append(metrics, metric)
		}
		for _, metric := range fetchExchangeMetrics(mgmtUri, prefix) {
			metrics = append(metrics, metric)
		}
		for _, metric := range metrics {
			body := []byte(metric.Name + "\t" + metric.Value + "\t" + strconv.FormatInt(metric.Timestamp, 10))
			msg := amqp.Publishing{ContentType: "text/plain", Body: body}
			err = queueChan.Publish("", queueName, false, false, msg)
			if err != nil {
				log.Printf("publish err: %s", err)
				return
			}
			//log.Printf("metric\t%s\t\t%s", metric.Name, metric.Value)
		}
		time.Sleep(time.Second * 5)
	}
	queueChan.Close()
	queueConn.Close()
}

func CreateHelloTopology(connection *amqp.Connection) (*amqp.Channel, error) {
	channel, err := connection.Channel()
	if err != nil {
		log.Println("Failed to get channel!: ", err)
		return nil, err
	}

	err = channel.ExchangeDeclare("hello-exchange", "direct", true, false, false, false, nil)
	if err != nil {
		log.Println("Failed to declare exchange!: ", err)
		return nil, err
	}

	_, err = channel.QueueDeclare("hello-queue", false, false, false, false, nil)
	if err != nil {
		log.Println("Failed to declare queue!: ", err)
		return nil, err
	}

	err = channel.QueueBind("hello-queue", "hola", "hello-exchange", false, nil)
	if err != nil {
		log.Println("Failed to bind to queue!: ", err)
		return nil, err
	}

	return channel, nil
}

func newConsumeChannel(conn *amqp.Connection, queueName string) (<-chan amqp.Delivery, error) {
	ch, err := conn.Channel()
	if err != nil {
		return nil, err
	}
	return ch.Consume(queueName, "", false, false, false, false, nil)
}

/*
If ch is nil, a new Connection and Channel will be created, and this publisher
will 'own' the connection. A call to close() will close both channel and connection.

If ch is provided, then this publisher will reuse the channel and
calls to close() will do nothing.
*/
func newPublisher(serverURI string, cfg *ChannelConfig, ch *amqp.Channel) *Publisher {

	var conn *amqp.Connection
	var err error

	if ch == nil {
		conn, err = amqp.Dial(serverURI)
		if err != nil {
			panic(fmt.Errorf("Failed to connect to RabbitMQ: %v", err))
		}

		ch, err = conn.Channel()
		if err != nil {
			panic(fmt.Errorf("Failed to open a channel: %v", err))
		}
	}

	_, err = ch.QueueDeclare(*cfg.Name, *cfg.Durable, *cfg.AutoDelete, *cfg.Exclusive, false, *cfg.Args)
	if err != nil {
		panic(fmt.Errorf("Failed to declare queue %s: %v", cfg.Name, err))
	}

	ch.QueuePurge(*cfg.Name, true)
	return &Publisher{exch: "", routingKey: *cfg.Name, conn: conn, ch: ch, typeTag: *cfg.TypeTag}
}

// openChannel opens up a RabbitMQ channel.
func openChannel(conn *amqp.Connection) *amqp.Channel {
	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("Failed to open a channel: %s", err)
	}
	return ch
}

func Consume(connection *amqp.Connection, queue string, outChannel chan *protobuf.Notification) {
	// create a channel on this connection
	channel, err := connection.Channel()
	if err != nil {
		panic(err)
	}
	defer channel.Close()

	// start consuming data
	consumerTag := queue + "-consumer"
	deliveries, err := channel.Consume(
		queue,       // name
		consumerTag, // consumerTag
		false,       // noAck
		false,       // exclusive
		false,       // noLocal
		false,       // noWait
		nil,         // arguments
	)
	if err != nil {
		panic(err)
	}
	defer channel.Cancel(consumerTag, false)

	for delivery := range deliveries {
		notif := &protobuf.Notification{}
		proto.Unmarshal(delivery.Body, notif)

		outChannel <- notif

		if err := delivery.Ack(false); err != nil {
			panic(err)
		}
	}
}

func connect(conn *amqp.Connection, exchange string, wg *sync.WaitGroup) {
	ch, err := conn.Channel()
	utils.FailOnError(err, "Failed to open a channel")
	defer conn.Close()

	err = ch.ExchangeDeclare(
		exchange, // name
		"direct", // type
		true,     // durable
		false,    // auto-deleted
		false,    // internal
		false,    // no-wait
		nil,      // arguments
	)
	utils.FailOnError(err, "Failed to declare an exchange")

	keys := []string{"create", "update"}
	for {
		for _, key := range keys {
			go connectKey(ch, key, exchange)
		}
		log.Printf("Break \n \n")
		time.Sleep(40 * time.Second)
	}
}

func Log(connection *amqp.Connection, inChannel chan *protobuf.Log) {
	// create a channel on this connection
	channel, err := connection.Channel()
	if err != nil {
		panic(err)
	}
	defer channel.Close()

	for log := range inChannel {
		body, err := proto.Marshal(log)
		if err != nil {
			panic(err)
		}

		err = channel.Publish(
			exchange,   // publish to an exchange
			routingKey, // routing to 0 or more queues
			false,      // mandatory
			false,      // immediate
			amqp.Publishing{
				Headers:         amqp.Table{},
				ContentType:     "",
				ContentEncoding: "",
				Body:            body,
				DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent
				Priority:        0,              // 0-9
				// a bunch of application/implementation-specific fields
			},
		)
		if err != nil {
			panic(err)
		}
	}
}

func receive(conn *amqp.Connection, exchange string, key string, wg *sync.WaitGroup) {
	ch, err := conn.Channel()
	utils.FailOnError(err, "Failed to open a channel")
	defer ch.Close()

	err = ch.ExchangeDeclare(
		exchange, // name
		"direct", // type
		true,     // durable
		false,    // auto-deleted
		false,    // internal
		false,    // no-wait
		nil,      // arguments
	)
	utils.FailOnError(err, "Failed to declare an exchange")

	q, err := ch.QueueDeclare(
		"",    // name
		false, // durable
		false, // delete when usused
		true,  // exclusive
		false, // no-wait
		nil,   // arguments
	)
	utils.FailOnError(err, "Failed to declare a queue")

	err = ch.QueueBind(
		q.Name,   // queue name
		key,      // routing key
		exchange, // exchange
		false,
		nil)
	utils.FailOnError(err, "Failed to bind a queue")

	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer
		true,   // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)
	utils.FailOnError(err, "Failed to register a consumer")

	forever := make(chan bool)

	go func() {
		for msg := range msgs {
			log.Printf(" [i] %s", key)
			go processMessage(msg.Body)
		}
	}()

	log.Printf(" [*] Waiting for logs. To exit press CTRL+C")
	<-forever
}

// shutdown is a general closer function for handling close gracefully
// Mostly here for both consumers and producers
// After a reconnection scenerio we are gonna call shutdown before connection
func shutdown(conn *amqp.Connection) error {
	if err := conn.Close(); err != nil {
		if amqpError, isAmqpError := err.(*amqp.Error); isAmqpError && amqpError.Code != 504 {
			return fmt.Errorf("AMQP connection close error: %s", err)
		}
	}

	return nil
}

/**
 * Retrieves a channel from the given connection, ConnectionDefinition
 */
func OpenChannel(connection *amqp.Connection, cd ConnectionDefinition) error {
	channel, err := connection.Channel()
	if err != nil {
		return err
	}

	Channel[cd.Vhost] = channel
	return nil
}

func (i *CLI) amqpErrorWatcher(amqpConn *amqp.Connection) {
	errChan := make(chan *amqp.Error)
	errChan = amqpConn.NotifyClose(errChan)

	err, ok := <-errChan
	if ok {
		i.logger.WithField("err", err).Error("amqp connection errored, terminating")
		i.cancel()
	}
}

// CreateChannel is a wrapper to simplify creating a channel with its attendant
// exchange.
func CreateChannel(conn *amqp.Connection, exchange, exchangeType string) (*amqp.Channel, error) {
	ch, err := conn.Channel()
	if err != nil {
		return nil, err
	}

	if err = ch.ExchangeDeclare(exchange, exchangeType, true, false, false, false, nil); err != nil {
		return nil, err
	}
	return ch, nil
}

// Closes the connection
func (amqpBackend *AMQPBackend) close(channel *amqp.Channel, conn *amqp.Connection) error {
	if err := channel.Close(); err != nil {
		return fmt.Errorf("Channel Close: %s", err)
	}

	if err := conn.Close(); err != nil {
		return fmt.Errorf("Connection Close: %s", err)
	}

	return nil
}

func Publish(messages chan *amqp.Publishing, connectionUri, exchange,
	routingKey string, mandatory, immediate bool, results chan *PublishResult) {

	var err error
	var conn *amqp.Connection
	var channel *amqp.Channel

	defer close(results)

	if conn, err = amqp.Dial(connectionUri); err != nil {
		results <- &PublishResult{"Failed to connect", err, true}
		return
	}

	defer conn.Close()

	if channel, err = conn.Channel(); err != nil {
		results <- &PublishResult{"Failed to get channel", err, true}
		return
	}

	pubAcks, pubNacks := channel.NotifyConfirm(make(chan uint64), make(chan uint64))
	chanClose := channel.NotifyClose(make(chan *amqp.Error))
	if err = channel.Confirm(false); err != nil {
		results <- &PublishResult{
			"Failed to put channel into confirm mode",
			err,
			true,
		}
		return
	}

	for message := range messages {
		err = channel.Publish(exchange, routingKey, mandatory, immediate, *message)
		if err != nil {
			results <- &PublishResult{"Failed to publish message", err, false}
			continue
		}

		select {
		case err = <-chanClose:
			results <- &PublishResult{"Channel closed!", err, true}
		case <-pubAcks:
			results <- &PublishResult{
				fmt.Sprintf("Published to exchange '%s' routing key '%v': %+v", exchange, routingKey, message),
				nil,
				false,
			}
		case <-pubNacks:
			results <- &PublishResult{"Received basic.nack for message", errors.New("'basic.nack'"), false}
		}
	}
}

// Fire is called when an event should be sent to the message broker.k
func (q amqpConn) Fire(entry *logrus.Entry) error {
	ch, err := q.Connection.Channel()
	if err != nil {
		// Any other error other than connection closed, return.
		if err != amqp.ErrClosed {
			return err
		}
		// Attempt to connect again.
		var conn *amqp.Connection
		conn, err = amqp.Dial(q.params.URL)
		if err != nil {
			return err
		}
		ch, err = conn.Channel()
		if err != nil {
			return err
		}
	}
	defer ch.Close()

	err = ch.ExchangeDeclare(
		q.params.Exchange,
		q.params.ExchangeType,
		q.params.Durable,
		q.params.AutoDeleted,
		q.params.Internal,
		q.params.NoWait,
		nil,
	)
	if err != nil {
		return err
	}

	body, err := entry.String()
	if err != nil {
		return err
	}

	err = ch.Publish(
		q.params.Exchange,
		q.params.RoutingKey,
		q.params.Mandatory,
		q.params.Immediate,
		amqp.Publishing{
			ContentType: "application/json",
			Body:        []byte(body),
		})
	if err != nil {
		return err
	}

	return nil
}