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
}

func connectToAMQP(uri string) (*amqp.Connection, error) {

	var conn *amqp.Connection
	var err error

	if strings.Contains(uri, "amqps") {
		cfg := new(tls.Config)

		if len(os.Getenv("PMB_SSL_INSECURE_SKIP_VERIFY")) > 0 {
			cfg.InsecureSkipVerify = true
		}

		logrus.Debugf("calling DialTLS")
		conn, err = amqp.DialTLS(uri, cfg)
		logrus.Debugf("Connection obtained")
	} else {
		conn, err = amqp.Dial(uri)
	}

	if err != nil {
		return nil, err
	}

	//logrus.Debugf("Conn: ", conn)
	return conn, nil
}

// Tests that the server has handshaked the connection and seen the client
// protocol announcement.  Does not nest that the connection.open is successful.
func TestTLSHandshake(t *testing.T) {
	srv := startTlsServer()
	defer srv.Close()

	cfg := new(tls.Config)
	cfg.RootCAs = x509.NewCertPool()
	cfg.RootCAs.AppendCertsFromPEM([]byte(caCert))

	cert, _ := tls.X509KeyPair([]byte(clientCert), []byte(clientKey))
	cfg.Certificates = append(cfg.Certificates, cert)

	c, err := amqp.DialTLS(srv.URL, cfg)

	select {
	case <-time.After(10 * time.Millisecond):
		t.Fatalf("did not succeed to handshake the TLS connection after 10ms")
	case header := <-srv.Header:
		if string(header) != "AMQP" {
			t.Fatalf("expected to handshake a TLS connection, got err: %v", err)
		}
	}

	if st := c.ConnectionState(); !st.HandshakeComplete {
		t.Errorf("TLS handshake failed, TLS connection state: %+v", st)
	}
}

func init() {
	//defer debug.PrintStack()
	var err error
	cert, err = tls.LoadX509KeyPair("certs/client0-ca.crt", "certs/client0.key")
	if err != nil {
		log.Fatalf("server: loadkeys: %s", err)
	}
	tmpurl := os.Getenv("CLOUDAMQP_URL")
	if tmpurl != "" {
		url = tmpurl
	}

	connection, err := amqp.DialTLS(url, &tls.Config{Certificates: []tls.Certificate{cert}, PreferServerCipherSuites: true, InsecureSkipVerify: true})
	//connection, err := amqp.Dial(url)
	channel, _ = connection.Channel()
	durable := true
	autoDelete, noWait := false, false
	internal := false
	err = channel.ExchangeDeclarePassive(exchangeName, "topic", durable, autoDelete, internal, noWait, nil)
	if err != nil {
		log.Printf("Exchange does not exist: %s", err)
		channel, _ = connection.Channel()
		err = channel.ExchangeDeclare(exchangeName, "topic", durable, autoDelete, internal, noWait, nil)
		if err != nil {
			log.Fatalf("Could not create Exchange: %s", err)
		}
	}
}

func (r *rabbitMQConn) tryToConnect(secure bool, config *tls.Config) error {
	var err error

	if secure || config != nil {
		if config == nil {
			config = &tls.Config{
				InsecureSkipVerify: true,
			}
		}

		url := strings.Replace(r.url, "amqp://", "amqps://", 1)
		r.Connection, err = amqp.DialTLS(url, config)
	} else {
		r.Connection, err = amqp.Dial(r.url)
	}

	if err != nil {
		return err
	}
	r.Channel, err = newRabbitChannel(r.Connection)
	if err != nil {
		return err
	}
	r.Channel.DeclareExchange(r.exchange)
	return nil
}

// If additional TLS settings were specified, the dialer uses amqp.DialTLS
// instead of amqp.Dial.
func (dialer *AMQPDialer) Dial(url string) (conn AMQPConnection, err error) {
	if dialer.tlsConfig != nil {
		conn, err = amqp.DialTLS(url, dialer.tlsConfig)
	} else {
		conn, err = amqp.Dial(url)
	}
	return
}

func run(config Config) {
	listener := pq.NewListener(config.PostgresURL, 10*time.Second, time.Minute, errorReporter)
	err := listener.Listen("urlwork")
	if err != nil {
		log.Fatal(err)
	}

	rabbitchannel := make(chan string, 100)

	go func() {
		cfg := new(tls.Config)
		cfg.InsecureSkipVerify = true
		conn, err := amqp.DialTLS(config.RabbitMQURL, cfg)
		if err != nil {
			log.Fatal(err)
		}
		defer conn.Close()

		ch, err := conn.Channel()
		if err != nil {
			log.Fatal(err)
		}
		defer ch.Close()

		for {
			payload := <-rabbitchannel
			log.Println(payload)
			err := ch.Publish("urlwork", "todo", false, false, amqp.Publishing{
				ContentType: "text/plain",
				Body:        []byte(payload),
			})
			if err != nil {
				log.Fatal(err)
			}
		}
	}()

	for {
		select {
		case notification := <-listener.Notify:
			rabbitchannel <- notification.Extra
		case <-time.After(90 * time.Second):
			go func() {
				err := listener.Ping()
				if err != nil {
					log.Fatal(err)
				}
			}()
		}
	}
}

func (ac *amqpClient) init() error {
	var err error
	ac.conn, err = amqp.DialTLS(ac.uri, ac.tlsConfig)
	if err != nil {
		return fmt.Errorf("DialTLS: %v", err)
	}

	ac.channel, err = ac.conn.Channel()
	if err != nil {
		return fmt.Errorf("Channel: %v", err)
	}
	if err := ac.channel.Confirm(false); err != nil {
		return fmt.Errorf("Confirm: %v", err)
	}
	ac.acks, ac.nacks = ac.channel.NotifyConfirm(make(chan uint64), make(chan uint64))

	go ac.handleMsgs()

	return nil
}

//  This function will return a amqp channel as well
//  as setup the reconnect and retry logic in case
//  the server its connected to becomes unavailable.
func (t *Tasks) SetupAmqpConnection() *amqp.Connection {
	config := t.GetConfig()
	conn, err := amqp.DialTLS(config.Uri, config.TlsConfig)

	//if err retry until connected.
	if err != nil {
		log.Printf("Error Connecting to amqp: %s\n", err)
		time.Sleep(1 * time.Second)
		conn = t.SetupAmqpConnection()
	} else {
		log.Printf("Connected to Rabbit")
		//Setup Reconnect logic
		go func() {
			log.Printf("closing: %s \n", <-conn.NotifyClose(make(chan *amqp.Error)))
			time.Sleep(1 * time.Second)
			conn = t.SetupAmqpConnection()
		}()
	}

	t.Connection = conn
	// return the connection
	return conn
}

func ExampleDialTLS() {
	// To get started with SSL/TLS follow the instructions for adding SSL/TLS
	// support in RabbitMQ with a private certificate authority here:
	//
	// http://www.rabbitmq.com/ssl.html
	//
	// Then in your rabbitmq.config, disable the plain AMQP port, verify clients
	// and fail if no certificate is presented with the following:
	//
	//   [
	//   {rabbit, [
	//     {tcp_listeners, []},     % listens on 127.0.0.1:5672
	//     {ssl_listeners, [5671]}, % listens on 0.0.0.0:5671
	//     {ssl_options, [{cacertfile,"/path/to/your/testca/cacert.pem"},
	//                    {certfile,"/path/to/your/server/cert.pem"},
	//                    {keyfile,"/path/to/your/server/key.pem"},
	//                    {verify,verify_peer},
	//                    {fail_if_no_peer_cert,true}]}
	//     ]}
	//   ].

	cfg := new(tls.Config)

	// The self-signing certificate authority's certificate must be included in
	// the RootCAs to be trusted so that the server certificate can be verified.
	//
	// Alternatively to adding it to the tls.Config you can add the CA's cert to
	// your system's root CAs.  The tls package will use the system roots
	// specific to each support OS.  Under OS X, add (drag/drop) your cacert.pem
	// file to the 'Certificates' section of KeyChain.app to add and always
	// trust.
	//
	// Or with the command line add and trust the DER encoded certificate:
	//
	//   security add-certificate testca/cacert.cer
	//   security add-trusted-cert testca/cacert.cer
	//
	// If you depend on the system root CAs, then use nil for the RootCAs field
	// so the system roots will be loaded.

	cfg.RootCAs = x509.NewCertPool()

	if ca, err := ioutil.ReadFile("testca/cacert.pem"); err == nil {
		cfg.RootCAs.AppendCertsFromPEM(ca)
	}

	// Move the client cert and key to a location specific to your application
	// and load them here.

	if cert, err := tls.LoadX509KeyPair("client/cert.pem", "client/key.pem"); err == nil {
		cfg.Certificates = append(cfg.Certificates, cert)
	}

	// Server names are validated by the crypto/tls package, so the server
	// certificate must be made for the hostname in the URL.  Find the commonName
	// (CN) and make sure the hostname in the URL matches this common name.  Per
	// the RabbitMQ instructions for a self-signed cert, this defautls to the
	// current hostname.
	//
	//   openssl x509 -noout -in server/cert.pem -subject
	//
	// If your server name in your certificate is different than the host you are
	// connecting to, set the hostname used for verification in
	// ServerName field of the tls.Config struct.

	conn, err := amqp.DialTLS("amqps://server-name-from-certificate/", cfg)

	log.Printf("conn: %v, err: %v", conn, err)
}

// Connects to the message queue, opens a channel, declares a queue
func (amqpBackend *AMQPBackend) open(taskUUID string) (*amqp.Connection, *amqp.Channel, amqp.Queue, <-chan amqp.Confirmation, error) {
	var (
		conn    *amqp.Connection
		channel *amqp.Channel
		queue   amqp.Queue
		err     error
	)

	// Connect
	// From amqp docs: DialTLS will use the provided tls.Config when it encounters an amqps:// scheme
	// and will dial a plain connection when it encounters an amqp:// scheme.
	conn, err = amqp.DialTLS(amqpBackend.config.Broker, amqpBackend.config.TLSConfig)
	if err != nil {
		return conn, channel, queue, nil, fmt.Errorf("Dial: %s", err)
	}

	// Open a channel
	channel, err = conn.Channel()
	if err != nil {
		return conn, channel, queue, nil, fmt.Errorf("Channel: %s", err)
	}

	// Declare an exchange
	err = channel.ExchangeDeclare(
		amqpBackend.config.Exchange,     // name of the exchange
		amqpBackend.config.ExchangeType, // type
		true,  // durable
		false, // delete when complete
		false, // internal
		false, // noWait
		nil,   // arguments
	)
	if err != nil {
		return conn, channel, queue, nil, fmt.Errorf("Exchange Declare: %s", err)
	}

	// Declare a queue
	arguments := amqp.Table{
		"x-message-ttl": int32(amqpBackend.getExpiresIn()),
	}
	queue, err = channel.QueueDeclare(
		taskUUID, // name
		false,    // durable
		true,     // delete when unused
		false,    // exclusive
		false,    // no-wait
		arguments,
	)
	if err != nil {
		return conn, channel, queue, nil, fmt.Errorf("Queue Declare: %s", err)
	}

	// Bind the queue
	if err := channel.QueueBind(
		queue.Name,                  // name of the queue
		taskUUID,                    // binding key
		amqpBackend.config.Exchange, // source exchange
		false, // noWait
		nil,   // arguments
	); err != nil {
		return conn, channel, queue, nil, fmt.Errorf("Queue Bind: %s", err)
	}

	// Enable publish confirmations
	if err := channel.Confirm(false); err != nil {
		return conn, channel, queue, nil, fmt.Errorf("Channel could not be put into confirm mode: %s", err)
	}

	return conn, channel, queue, channel.NotifyPublish(make(chan amqp.Confirmation, 1)), nil
}

func run(config Config) {
	session, err := r.Connect(r.ConnectOpts{
		Address:  config.RethinkDBAddress,
		Database: config.RethinkDBDatabase,
		AuthKey:  config.RethinkDBAuthkey,
		TLSConfig: &tls.Config{
			InsecureSkipVerify: true,
		},
	})
	if err != nil {
		log.Fatal(session, err)
	}

	rabbitchannel := make(chan []byte, 100)

	go func() {
		cfg := new(tls.Config)
		cfg.InsecureSkipVerify = true
		conn, err := amqp.DialTLS(config.RabbitMQURL, cfg)
		if err != nil {
			log.Fatal(err)
		}
		defer conn.Close()

		ch, err := conn.Channel()
		if err != nil {
			log.Fatal(err)
		}
		defer ch.Close()

		for {
			payload := <-rabbitchannel
			log.Println(string(payload))
			err := ch.Publish("urlwork", "todo", false, false, amqp.Publishing{
				ContentType: "text/plain",
				Body:        []byte(payload),
			})
			if err != nil {
				log.Fatal(err)
			}
		}
	}()

	res, err := r.Table("urls").Changes().Run(session)
	if err != nil {
		log.Fatalln(err)
	}

	var value interface{}

	for res.Next(&value) {
		mapval := value.(map[string]interface{})
		if mapval["new_val"] != nil && mapval["old_val"] == nil {
			jsonbytes, err := json.Marshal(mapval["new_val"])
			if err != nil {
				log.Fatal(err)
			}
			rabbitchannel <- jsonbytes
		}
	}
}

// makeAmqpChannel sets an AMQP connection up using SSL if configuration is provided
func makeAmqpChannel(conf *cmd.AMQPConfig) (*amqp.Channel, error) {
	var conn *amqp.Connection
	var err error

	log := blog.Get()

	serverURL, err := conf.ServerURL()
	if err != nil {
		return nil, err
	}

	if conf.Insecure == true {
		// If the Insecure flag is true, then just go ahead and connect
		conn, err = amqp.Dial(serverURL)
	} else {
		// The insecure flag is false or not set, so we need to load up the options
		log.Info("AMQPS: Loading TLS Options.")

		if strings.HasPrefix(serverURL, "amqps") == false {
			err = fmt.Errorf("AMQPS: Not using an AMQPS URL. To use AMQP instead of AMQPS, set insecure=true")
			return nil, err
		}

		if conf.TLS == nil {
			err = fmt.Errorf("AMQPS: No TLS configuration provided. To use AMQP instead of AMQPS, set insecure=true")
			return nil, err
		}

		cfg := new(tls.Config)

		// If the configuration specified a certificate (or key), load them
		if conf.TLS.CertFile != nil || conf.TLS.KeyFile != nil {
			// But they have to give both.
			if conf.TLS.CertFile == nil || conf.TLS.KeyFile == nil {
				err = fmt.Errorf("AMQPS: You must set both of the configuration values AMQP.TLS.KeyFile and AMQP.TLS.CertFile")
				return nil, err
			}

			cert, err := tls.LoadX509KeyPair(*conf.TLS.CertFile, *conf.TLS.KeyFile)
			if err != nil {
				err = fmt.Errorf("AMQPS: Could not load Client Certificate or Key: %s", err)
				return nil, err
			}

			log.Info("AMQPS: Configured client certificate for AMQPS.")
			cfg.Certificates = append(cfg.Certificates, cert)
		}

		// If the configuration specified a CA certificate, make it the only
		// available root.
		if conf.TLS.CACertFile != nil {
			cfg.RootCAs = x509.NewCertPool()

			ca, err := ioutil.ReadFile(*conf.TLS.CACertFile)
			if err != nil {
				err = fmt.Errorf("AMQPS: Could not load CA Certificate: %s", err)
				return nil, err
			}
			cfg.RootCAs.AppendCertsFromPEM(ca)
			log.Info("AMQPS: Configured CA certificate for AMQPS.")
		}

		conn, err = amqp.DialTLS(serverURL, cfg)
	}

	if err != nil {
		return nil, err
	}

	return conn.Channel()
}

func (i *CLI) setupJobQueueAndCanceller() error {
	switch i.Config.QueueType {
	case "amqp":
		var amqpConn *amqp.Connection
		var err error
		if i.Config.AmqpInsecure {
			amqpConn, err = amqp.DialTLS(
				i.Config.AmqpURI,
				&tls.Config{InsecureSkipVerify: true},
			)
		} else {
			amqpConn, err = amqp.Dial(i.Config.AmqpURI)
		}
		if err != nil {
			i.logger.WithField("err", err).Error("couldn't connect to AMQP")
			return err
		}

		go i.amqpErrorWatcher(amqpConn)

		i.logger.Debug("connected to AMQP")

		canceller := NewAMQPCanceller(i.ctx, amqpConn)
		i.logger.WithFields(logrus.Fields{
			"canceller": fmt.Sprintf("%#v", canceller),
		}).Debug("built")

		i.Canceller = canceller

		go canceller.Run()

		jobQueue, err := NewAMQPJobQueue(amqpConn, i.Config.QueueName)
		if err != nil {
			return err
		}

		jobQueue.DefaultLanguage = i.Config.DefaultLanguage
		jobQueue.DefaultDist = i.Config.DefaultDist
		jobQueue.DefaultGroup = i.Config.DefaultGroup
		jobQueue.DefaultOS = i.Config.DefaultOS

		i.JobQueue = jobQueue
		return nil
	case "file":
		canceller := NewFileCanceller(i.ctx, i.Config.BaseDir)
		go canceller.Run()

		i.Canceller = canceller

		jobQueue, err := NewFileJobQueue(i.Config.BaseDir, i.Config.QueueName, i.Config.FilePollingInterval)
		if err != nil {
			return err
		}

		jobQueue.DefaultLanguage = i.Config.DefaultLanguage
		jobQueue.DefaultDist = i.Config.DefaultDist
		jobQueue.DefaultGroup = i.Config.DefaultGroup
		jobQueue.DefaultOS = i.Config.DefaultOS

		i.JobQueue = jobQueue
		return nil
	}

	return fmt.Errorf("unknown queue type %q", i.Config.QueueType)
}

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
	var err error
	if q.SslCert != "" && q.SslKey != "" {
		// make new tls config
		cfg := new(tls.Config)
		if q.SslCa != "" {
			// create ca pool
			cfg.RootCAs = x509.NewCertPool()

			// add self-signed cert
			if ca, err := ioutil.ReadFile(q.SslCa); err == nil {
				cfg.RootCAs.AppendCertsFromPEM(ca)
			} else {
				log.Println(err)
			}
		}
		if cert, err := tls.LoadX509KeyPair(q.SslCert, q.SslKey); err == nil {
			cfg.Certificates = append(cfg.Certificates, cert)
		} else {
			log.Println(err)
		}
		connection, err = amqp.DialTLS(q.URL, cfg)

	} 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
}

func ExampleDialTLS() {
	// This example assume you have a RabbitMQ node running on localhost
	// with TLS enabled.
	//
	// The easiest way to create the CA, certificates and keys required for these
	// examples is by using tls-gen: https://github.com/michaelklishin/tls-gen
	//
	// A comprehensive RabbitMQ TLS guide can be found at
	// http://www.rabbitmq.com/ssl.html
	//
	// Once you have the required TLS files in place, use the following
	// rabbitmq.config example for the RabbitMQ node that you will run on
	// localhost:
	//
	//   [
	//   {rabbit, [
	//     {tcp_listeners, []},     % listens on 127.0.0.1:5672
	//     {ssl_listeners, [5671]}, % listens on 0.0.0.0:5671
	//     {ssl_options, [{cacertfile,"/path/to/your/testca/cacert.pem"},
	//                    {certfile,"/path/to/your/server/cert.pem"},
	//                    {keyfile,"/path/to/your/server/key.pem"},
	//                    {verify,verify_peer},
	//                    {fail_if_no_peer_cert,true}]}
	//     ]}
	//   ].
	//
	//
	// In the above rabbitmq.config example, we are disabling the plain AMQP port
	// and verifying that clients and fail if no certificate is presented.
	//
	// The self-signing certificate authority's certificate (cacert.pem) must be
	// included in the RootCAs to be trusted, otherwise the server certificate
	// will fail certificate verification.
	//
	// Alternatively to adding it to the tls.Config. you can add the CA's cert to
	// your system's root CAs.  The tls package will use the system roots
	// specific to each support OS.  Under OS X, add (drag/drop) cacert.pem
	// file to the 'Certificates' section of KeyChain.app to add and always
	// trust.  You can also add it via the command line:
	//
	//   security add-certificate testca/cacert.pem
	//   security add-trusted-cert testca/cacert.pem
	//
	// If you depend on the system root CAs, then use nil for the RootCAs field
	// so the system roots will be loaded instead.
	//
	// Server names are validated by the crypto/tls package, so the server
	// certificate must be made for the hostname in the URL.  Find the commonName
	// (CN) and make sure the hostname in the URL matches this common name.  Per
	// the RabbitMQ instructions (or tls-gen) for a self-signed cert, this defaults to the
	// current hostname.
	//
	//   openssl x509 -noout -in /path/to/certificate.pem -subject
	//
	// If your server name in your certificate is different than the host you are
	// connecting to, set the hostname used for verification in
	// ServerName field of the tls.Config struct.
	cfg := new(tls.Config)

	// see at the top
	cfg.RootCAs = x509.NewCertPool()

	if ca, err := ioutil.ReadFile("testca/cacert.pem"); err == nil {
		cfg.RootCAs.AppendCertsFromPEM(ca)
	}

	// Move the client cert and key to a location specific to your application
	// and load them here.

	if cert, err := tls.LoadX509KeyPair("client/cert.pem", "client/key.pem"); err == nil {
		cfg.Certificates = append(cfg.Certificates, cert)
	}

	// see a note about Common Name (CN) at the top
	conn, err := amqp.DialTLS("amqps://server-name-from-certificate/", cfg)

	log.Printf("conn: %v, err: %v", conn, err)
}