func TestRPC(b *testing.T) {
	conn, err := amqp.Dial(*url)
	if err != nil {
		b.Fatal(err)
	}

	serverCodec, err := NewServerCodec(conn, *queue, JSONCodec{})
	if err != nil {
		b.Fatal(err)
	}

	server := rpc.NewServer()

	err = server.Register(new(RPC))
	if err != nil {
		b.Fatal(err)
	}

	go func() { server.ServeCodec(serverCodec) }()

	var clientCodecs []rpc.ClientCodec
	var clients []*rpc.Client
	wait := new(sync.WaitGroup)
	mu := new(sync.Mutex)

	wait.Add(10)

	for i := 0; i < 10; i++ {
		go func() {
			conn, err := amqp.Dial(*url)
			if err != nil {
				b.Fatal(err)
			}

			codec, err := NewClientCodec(conn, *queue, JSONCodec{})
			if err != nil {
				b.Fatal(err)
			}

			mu.Lock()
			clientCodecs = append(clientCodecs, codec)
			clients = append(clients, rpc.NewClientWithCodec(codec))
			mu.Unlock()
			wait.Done()
		}()
	}

	wait.Wait()

	for i := 0; i < 10; i++ {
		wait.Add(10)
		go func() {
			for _, client := range clients {
				go doCall(b, client, wait)
			}
		}()
	}

	wait.Wait()
}

/*Redial continually connects to the URL, returns no longer possible
 *no guarantee on the number of sessions returned on close.
 *==============
 *URL reference
 *amqp://user:[email protected]:port/vhost
 *a different URL-structure will not work
 *==============
 */
func (r Rabbit) Redial(ctx context.Context, url string) chan Session {
	sessions := make(chan Session)
	go func() {
		defer close(sessions)
		for {
			select {
			default:
				log.Info("Dialing")
			case <-ctx.Done():
				log.Infof("Shutting down session factory")
				return
			}
			conn, err := amqp.Dial(url)
			if err != nil {
				log.Warnf("Can't dial. Waiting 10 seconds...")
				time.Sleep(10 * time.Second)
				conn, err = amqp.Dial(url)
				if err != nil {
					log.Errorf("cannot (re)dial: %v: %q", err, url)
					return
				}
			}
			ch, err := conn.Channel()
			if err != nil {
				log.Errorf("cannot create channel %v: %v", r.Exchange, err)
				return
			}
			// idempotent declaration
			if err := r.DeclareExc(ch); err != nil {
				log.Errorf("cannot declare %v exchange: %v", r.ExchangeType, err)
				return
			}
			//Deliveries on the returned chan will be buffered indefinitely.  To limit memory
			//of this buffer, use the Channel.Qos method to limit the amount of
			//unacknowledged/buffered deliveries the server will deliver on this Channel.
			err = ch.Qos(
				r.QoS, // prefetch count
				0,     // prefetch size
				false) // global
			if err != nil {
				log.Errorf("Error setting Qos %v", err)
			}
			select {
			// this will block here if the subscriber is not using the session
			case sessions <- Session{conn, ch}:
				log.Info("New session has been initialized.")
			case <-ctx.Done():
				log.Infof("Shutting down session")
				return
			}
		}
	}()
	return sessions
}

func publish(ch *amqp.Channel, q amqp.Queue, c <-chan string) {
	for {
		connectionStatus := <-c
		if connectionStatus == "lost" {
			for i := 0; i < 40; i++ {
				conn, err := amqp.Dial("amqp://guest:[email protected]:5672/")
				if err != nil {
					fmt.Println("Failed to connect to RabbitMQ")
					time.Sleep(1 * time.Second)
					continue
				} else {
					defer conn.Close()
				}
			}
		}
		if connectionStatus == "ok" {
			conn, err := amqp.Dial("amqp://guest:[email protected]:5672/")
			if err != nil {
				fmt.Println("Failed to connect to RabbitMQ")
				time.Sleep(1 * time.Second)
				continue
			} else {
				defer conn.Close()
			}
			ch, _ := conn.Channel()
			defer ch.Close()
			body := bodyFrom(os.Args)
			q, err := ch.QueueDeclare(
				"task_queue", // name
				true,         // durable
				false,        // delete when unused
				false,        // exclusive
				false,        // no-wait
				nil,          // arguments
			)
			err = ch.Publish(
				"",     // exchange
				q.Name, // routing key
				false,  // mandatory
				false,
				amqp.Publishing{
					DeliveryMode: amqp.Persistent,
					ContentType:  "text/plain",
					Body:         []byte(body),
				})
			if err == nil {
				log.Printf(" [x] Sent %s", body)
			}
		}
		time.Sleep(1 * time.Second)
	}
}

func main() {
	conn, err := amqp.Dial("amqp://guest:[email protected]:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

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

	body := "hello"
	err = ch.Publish(
		"",     // exchange
		q.Name, // routing key
		false,  // mandatory
		false,  // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		})
	log.Printf(" [x] Sent %s", body)
	failOnError(err, "Failed to publish a message")
}

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

func (qi *amqpInstance) dial(queueName string) (*amqp.Channel, error) {

	/*  amqpAddr, err := config.GetString("amqp:url") //setup on cloudifice config - cfs.yml
	if err != nil {
		amqpAddr = "amqp://172.17.0.5:5672/"
	} */

	conn, err := amqp.Dial(qi.RmqAddress)
	if err != nil {
		return nil, err
	}

	log.Printf(" [QS] Dialed to (%s)", qi.RmqAddress)

	channel, err := conn.Channel()

	if err != nil {
		return nil, err
	}

	//NOTE: This is a passive call.
	//TODO: Does everycall require a check?
	q, err := channel.QueueInspect(queueName)
	if err != nil {
		return nil, err
	}

	log.Printf(" [x] Connection successful to  (%s,%s)", qi.RmqAddress, q.Name)
	return channel, err
}

func main() {
	conn, err := amqp.Dial("amqp://guest:[email protected]:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	err = ch.ExchangeDeclare(
		"logs",   // name
		"fanout", // type
		true,     // durable
		false,    // auto-deleted
		false,    // internal
		false,    // no-wait
		nil,      // arguments
	)
	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
	)
	failOnError(err, "Failed to declare a queue")

	err = ch.QueueBind(
		q.Name, // queue name
		"",     // routing key
		"logs", // exchange
		false,
		nil)
	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
	)
	failOnError(err, "Failed to register a consumer")

	forever := make(chan bool)

	go func() {
		for d := range msgs {
			log.Printf(" [x] %s", d.Body)
		}
	}()

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

func main() {
	conn, err := amqp.Dial("amqp://localhost:5672")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	err = ch.ExchangeDeclare(
		"logs",   // name
		"fanout", // kind
		true,     // durable
		false,    // auto-delete
		false,    // internal
		false,    // no-wait
		nil,      // args
	)
	failOnError(err, "Failed to declare an exchange")

	body := bodyFrom(os.Args[1:])
	err = ch.Publish(
		"logs", // exchange
		"",     // key
		false,  // mandatory
		false,  // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		})
	failOnError(err, "Failed to publishing message")

	// time.Sleep(10 * time.Second)
}

func publish(amqpURI, exchange, exchangeType, routingKey, body string, reliable bool) error {

	// This function dials, connects, declares, publishes, and tears down,
	// all in one go. In a real service, you probably want to maintain a
	// long-lived connection as state, and publish against that.

	log.Printf("dialing %s", amqpURI)
	connection, err := amqp.Dial(amqpURI)
	if err != nil {
		return fmt.Errorf("Dial: %s", err)
	}
	defer connection.Close()

	log.Printf("got Connection, getting Channel")
	channel, err := connection.Channel()
	if err != nil {
		return fmt.Errorf("Channel: %s", err)
	}

	log.Printf("got Channel, declaring %q Exchange (%s)", exchangeType, exchange)
	if err := channel.ExchangeDeclare(
		exchange,     // name
		exchangeType, // type
		false,        // durable
		false,        // auto-deleted
		false,        // internal
		false,        // noWait
		nil,          // arguments
	); err != nil {
		return fmt.Errorf("Exchange Declare: %s", err)
	}

	// Prepare this message to be persistent.  Your publishing requirements may
	// be different.
	msg := amqp.Publishing{
		//DeliveryMode: amqp.Persistent,
		//Timestamp:    time.Now(),
		//ContentType:  "text/plain",
		//Body:         []byte(body),
		Headers:         amqp.Table{},
		ContentType:     "text/plain",
		ContentEncoding: "",
		Body:            []byte(body),
		DeliveryMode:    1, // 1=non-persistent, 2=persistent
		Priority:        0, // 0-9
	}

	log.Printf("declared Exchange, publishing %dB body (%s)", len(body), body)
	if err = channel.Publish(
		exchange,   // publish to an exchange
		routingKey, // routing to 0 or more queues
		false,      // mandatory
		false,      // immediate
		msg,
	); err != nil {
		return fmt.Errorf("Exchange Publish: %s", err)
	}

	return nil
}

func New(config *config.Config) *Broker {
	broker := new(Broker)
	broker.cfg = config

	// Connect to the message queue server
	address := fmt.Sprintf("amqp://%s:%[email protected]%s:%d/", config.GetQueueUsername(),
		config.GetQueuePassword(),
		config.GetQueueHost(),
		config.GetQueuePort())

	conn, err := amqp.Dial(address)
	if err != nil {
		fmt.Printf("There was an error connecting to the broker")
		panic(err)
	}
	broker.conn = conn

	// Create a new channel
	channel, err := conn.Channel()
	if err != nil {
		panic(err)
	}
	broker.channel = channel

	broker.initializeQueues()
	return broker
}

// NewRabbitMQQueueLenCheck returns a check function that check if queue have more pending messages than a given limit
func NewRabbitMQQueueLenCheck(host, service, amqpuri, queue string, max int) CheckFunction {
	return func() Event {
		result := Event{Host: host, Service: service}

		conn, err := amqp.Dial(amqpuri)
		if err != nil {
			result.State = "critical"
			result.Description = err.Error()
			return result
		}

		ch, err := conn.Channel()
		if err != nil {
			result.State = "critical"
			result.Description = err.Error()
			return result
		}
		defer ch.Close()
		defer conn.Close()

		queueInfo, err := ch.QueueInspect(queue)
		if err != nil {
			result.State = "critical"
			result.Description = err.Error()
			return result
		}

		var state = "critical"
		if queueInfo.Messages <= max {
			state = "ok"
		}
		return Event{Host: host, Service: service, State: state, Metric: float32(queueInfo.Messages)}
	}
}

// Connects to the RabbitMQ host and sends a message containing the stock
// information for the stock symbols in the config file.
func sendMessageToRabbitMQ(message []byte) {
	var amqpAddress = "amqp://" + Config.Rabbitmq_user + ":" + Config.Rabbitmq_pass + "@" + Config.Rabbitmq_host + ":" + strconv.Itoa(Config.Rabbitmq_port)
	conn, err := amqp.Dial(amqpAddress)
	failOnError(err, "Failed to connect to RabbitMQ")

	ch, err := conn.Channel()
	failOnError(err, "Failed to open channel")
	defer ch.Close()

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

	err = ch.Publish(
		"amq.topic",
		q.Name,
		false,
		false,
		amqp.Publishing{
			ContentType: "application/json",
			Body:        (message),
		})
	failOnError(err, "Failed to publish stock update")
}

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
}

func main() {
	conn, err := amqp.Dial("amqp://localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	err = ch.ExchangeDeclare(
		"logs_topic", // name
		"topic",      // kind
		true,         // durable
		false,        // auto-delete
		false,        // internal
		false,        // no-wait
		nil,          // args
	)
	failOnError(err, "Failed to declare an exchange")

	body := bodyFrom(os.Args)
	severity := severityFrom(os.Args)
	err = ch.Publish(
		"logs_topic", // exchage
		severity,     // key
		false,        // mandatory
		false,        // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		})
	log.Printf(" [x] Sent <%s> %s", severity, body)
}

func fibonacciRPC(n int) (res int, err error) {
	conn, err := amqp.Dial("amqp://guest:[email protected]:5672")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to declare a channel")
	defer ch.Close()

	q, err := ch.QueueDeclare("", false, false, true, false, nil)
	failOnError(err, "Failed to declare a queue")

	msgs, err := ch.Consume(q.Name, "", true, false, false, false, nil)
	failOnError(err, "Failed to register a consumer")

	corrId := randomString(32)

	err = ch.Publish("", "rpc_queue", false, false, amqp.Publishing{
		ContentType:   "text/plain",
		CorrelationId: corrId,
		ReplyTo:       q.Name,
		Body:          []byte(strconv.Itoa(n)),
	})
	failOnError(err, "Failed to publish a message")

	// Waiting on response
	for d := range msgs {
		if corrId == d.CorrelationId {
			res, err = strconv.Atoi(string(d.Body))
			failOnError(err, "Failed to convert body to integer")
			break
		}
	}
	return
}

func main() {
	conn, err := amqp.Dial("amqp://localhost:5672/")
	failOnError(err, "Failed to connect to Rabbitmq")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

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

	body := bodyFrom(os.Args)
	err = ch.Publish(
		"",     // exchange
		q.Name, // routing key
		false,  // mandatory
		false,  // immediate
		amqp.Publishing{ // msg
			DeliveryMode: amqp.Persistent,
			ContentType:  "text/plain",
			Body:         []byte(body),
		})
	failOnError(err, "Failed to publish a message")
	log.Printf(" [x] Sent %s", body)
}

func init() {
	c, err := amqp.Dial("amqp://guest:[email protected]:5672/")
	if err != nil {
		panic(err)
	}
	conn = c
}

func init() {
	wg := sync.WaitGroup{}
	wg.Add(2)

	go func() {
		var err error
		amqpConnection, err = amqp.Dial("amqp://rabbitmq:[email protected]:5672/")
		if err != nil {
			log.Fatal("can't connect to rabbitmq (", err, ")")
		}

		wg.Done()
	}()

	go func() {
		redisClient = redis.NewClient(&redis.Options{
			Addr:     "localhost:6379",
			Password: "",
			DB:       0,
		})

		_, err := redisClient.Ping().Result()
		if err != nil {
			log.Fatal("can't connect to redis (", err, ")")
		}

		wg.Done()
	}()

	wg.Wait()
}

func main() {
	conn, err := amqp.Dial("amqp://guest:[email protected]:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

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

	body := bodyFrom(os.Args)
	err = ch.Publish(
		"logs", // exchange
		"",     // routing key
		false,  // mandatory
		false,  // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		})
	failOnError(err, "Failed to publish a message")

	log.Printf(" [x] Sent %s", body)
}

func main() {
	conn, err := amqp.Dial("amqp://guest:[email protected]:15672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

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

	forever := make(chan bool)

	go func() {
		for d := range msgs {
			log.Printf("Received a message: %s", d.Body)
		}
	}()

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