func MakeQueue(c *amqp.Channel) amqp.Queue {
	q, err := c.QueueDeclare("stress-test-exchange", true, false, false, false, nil)
	if err != nil {
		log.Fatal(err)
	}
	return q
}

// Function to send keep alive message over specified channel
func sendKeepAlive(channel *amqp.Channel, client ClientConfig) error {
	// Create a keepalive struct to send to server
	body := &KeepAlive{
		Name:          client.Name,
		Address:       client.Address,
		Subscriptions: client.Subscriptions,
		Timestamp:     time.Now().Unix(),
	}

	// Parse the keepalive struct to json
	bodyJson, err := json.Marshal(body)
	if err != nil {
		return err
	}

	// Create the amqp message to publish
	msg := amqp.Publishing{
		ContentType:  "application/octet-stream",
		DeliveryMode: amqp.Persistent,
		Priority:     0,
		Body:         bodyJson,
	}

	// Publish message to amqp server
	if err := channel.Publish("keepalives", "", false, false, msg); err != nil {
		return err
	}

	// Returns nil as error if message was sent successfully
	return nil
}

func MakeQueue(c *amqp.Channel, queueName string, queueDurability bool) amqp.Queue {
	q, err2 := c.QueueDeclare(queueName, queueDurability, false, false, false, nil)
	if err2 != nil {
		panic(err2)
	}
	return q
}

func SendEmail(req *http.Request, mail Email, r render.Render, ch *amqp.Channel) {
	msg, _ := json.Marshal(mail)

	q, _ := ch.QueueDeclare(
		"email", // name
		true,    // durable
		false,   // delete when usused
		false,   // exclusive
		false,   // no-wait
		nil,     // arguments
	)

	err := ch.Publish(
		"",     // exchange
		q.Name, // routing key
		false,  // mandatory
		false,  // immediate
		amqp.Publishing{
			ContentType: "json/plain",
			Body:        []byte(msg),
		})

	if err == nil {
		r.JSON(http.StatusOK, map[string]interface{}{"Email": mail})
	} else {
		r.JSON(http.StatusInternalServerError, err)
	}
}

/*
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 handleChange(channel *amqp.Channel, ev *fsnotify.FileEvent) error {
	// check to see if the file is locked. If it is then break
	path, _ := filepath.Rel(rootDir, ev.Name)
	if FileIsLocked(path) {
		return nil
	}
	log.Printf("PUBLISHER: publishing %dB path (%q)", len(path), path)

	var change = &Change{
		IsCreate: ev.IsCreate(),
		IsDelete: ev.IsDelete(),
		IsMod:    ev.IsModify(),
		IsMove:   ev.IsRename(),
		ModDate:  time.Now().UTC(),
		Path:     path,
		Source:   hostname,
		Checksum: Checksum(ev.Name),
	}

	msg, err := json.Marshal(change)
	if err != nil {
		log.Fatalf("ERROR marshaling msg %s", change)
	}

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

				ContentType     string    // MIME content type
				ContentEncoding string    // MIME content encoding
				DeliveryMode    uint8     // Transient (0 or 1) or Persistent (2)
				Priority        uint8     // 0 to 9
				CorrelationId   string    // correlation identifier
				ReplyTo         string    // address to to reply to (ex: RPC)
				Expiration      string    // message expiration spec
				MessageId       string    // message identifier
				Timestamp       time.Time // message timestamp
				Type            string    // message type name
				UserId          string    // creating user id - ex: "guest"
				AppId           string    // creating application id
			*/
		},
	); err != nil {
		log.Fatalf("Exchange Publish: %s", err)
		panic(err)
	}

	return nil
}

func (rc requestCh) publish() error {
	var (
		err error
		out *amqp.Channel
	)
	for r := range rc {
		if r == nil {
			break
		}
		for out == nil {
			out, err = dial()
			if err != nil {
				out = nil
				log.Println(err)
				time.Sleep(*backoff)
			}
		}

		err = out.Publish(*exchange, *key, false, false,
			amqp.Publishing{
				Headers:      amqp.Table(r.headers),
				DeliveryMode: amqp.Persistent,
				Body:         []byte(r.body),
			},
		)
		if err != nil {
			return err
		}
	}
	return nil
}

func (rc RequestChan) publish() error {
	var (
		err error
		out *amqp.Channel
	)
	for r := range rc {
		if r == nil {
			break
		}
		for out == nil {
			out, err = dial()
			if err != nil {
				out = nil
				log.Println(err)
				time.Sleep(*backoff)
			}
		}

		err = out.Publish(*exchange, *key, false, false,
			amqp.Publishing{
				Headers:     amqp.Table(r.Headers),
				ContentType: r.ContentType,
				Body:        r.Body,
			},
		)
		if err != nil {
			return err
		}
	}
	return nil
}

func produce(conn *amqp.Connection, channel *amqp.Channel, val *interface{}) {

	if val == nil {
		log.Warn("the redis json is nil")
		return
	}

	body, err := json.Marshal(val)

	if err != nil || body == nil {
		log.Error("redis event to json error: %s , oplog is : %s ", err, string(body))
	} else {

		routingKey := "redis.event"
		log.Info("routing key is : %s ", routingKey)

		err = channel.Publish(
			EXCHANGE_KEY, // exchange
			routingKey,   // routing key
			false,        // mandatory
			false,        // immediate
			amqp.Publishing{
				ContentType: "text/plain",
				Body:        body,
			})
		if err != nil {
			log.Error("publish message err : %s ", err)
		}

		//TODO recreate channel ?
	}
}

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 cmdMain(ch *amqp.Channel) {
	fmt.Println(len(os.Args))
	if len(os.Args) != 4 {
		cmdUsage()
	}
	name := os.Args[2]
	val, _ := strconv.Atoi(os.Args[3])
	fmt.Println("ARGS", os.Args, name, val)

	m := UpdateMsg{name, val}

	msg, err := json.Marshal(m)
	if err != nil {
		log.Fatal("Barf!", err)
	}
	err = ch.Publish(
		"updateMsgs", // exchange
		"",           // routing key
		false,        // mandatory
		false,        // immediate
		amqp.Publishing{
			ContentType: "text/json",
			Body:        msg,
		})
	failOnError(err, "Failed to publish a message")
	log.Printf("Sending msg -- %s", msg)
	os.Exit(0)

}

func StartSubscribers(amqpChan *amqp.Channel, esClient *elastic.Client, config *config.Config) {
	log.Println("Starting RabbitMQ subscribers")
	msgChan, err := amqpChan.Consume(config.AmqpQueues[0], "", true, false, false, false, nil)
	utils.CheckFatalError(err)
	go incomingDocConsumer(msgChan, esClient, config)
	log.Println("Started RabbitMQ subscribers")
}

func MakeQueue(c *amqp.Channel) amqp.Queue {
	q, err2 := c.QueueDeclare("stress-test-queue", true, false, false, false, nil)
	if err2 != nil {
		panic(err2)
	}
	return q
}

func amqpEventReceiver(ch *amqp.Channel, q amqp.Queue, c chan Event) {
	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer
		true,   // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)
	if err != nil {
		//fixme send error to something ?
	}
	for d := range msgs {
		var ev Event
		ev.Headers = d.Headers
		ev.Headers["_transport-exchange"] = d.Exchange
		ev.Headers["_transport-RoutingKey"] = d.RoutingKey
		ev.Headers["_transport-ContentType"] = d.ContentType
		ev.Body = d.Body
		c <- ev
	}
	c <- Event{
		Body: []byte("dc1?"),
	}
	os.Exit(1)
}

func main() {

	log.Println("Starting elasticgin")

	//Config fetch
	config := config.GetConfFromJSONFile("config.json")

	//ES init
	esClient := initESClient(config.ElasticURL, config.Indices, config.SniffCluster)
	defer esClient.Stop()

	//Rabbitmq init
	var amqpChannel *amqp.Channel

	if config.UseAMQP {

		amqpConnection, amqpChannel := initAMQPClient(config)
		defer amqpConnection.Close()
		defer amqpChannel.Close()

		rabbitmq.StartSubscribers(amqpChannel, esClient, config)
	} else {
		amqpChannel = nil
	}

	api := api.SetupAPI(esClient, amqpChannel, config)
	api.Run(config.ApiURL)
}

func (r *Rabbitmq) Publish(ch *amqp.Channel, msg *amqp.Publishing) (err error) {
	try_count := 3
	if ch == nil {
		log.Printf("[WARNING] send to mq failed: Channel must not be empty\n")
		return errors.New("ampq channel is nil")
	}

	defer func() {
		if f := recover(); f != nil {
			log.Printf("[WARNING] send to mq failed:%v\n", f)
			err = f.(error)
		}
	}()

	for i := 0; i < try_count; i++ {

		err = ch.Publish(r.Exchange, r.RoutingKey, false, false, *msg)

		if err != nil {
			conn_err := r.Reconn()
			ch1, _ := r.Channel()
			*ch = *ch1
			log.Printf("[WARNING] RabbitMQ Reconn err is %v send to bnow failed:%v try Count:%v\n", conn_err, err, i)
			time.Sleep(time.Second)
			continue
		}
		break
	}
	return
}

func (s *Server) run() error {
	ch, err := s.dial()
	if err != nil {
		return err
	}

	for name, sink := range s.Sinks {
		if err := ch.ExchangeDeclare(name, sink.Type, true, false, false, false, nil); err != nil {
			return err
		}

		go func(name string, sink *Sink) {
			var (
				ch  *amqp.Channel
				elt []byte
				err error
			)
			for s.alive {
				if err != nil {
					ch = nil
					log.Println(err)
					err = nil
					time.Sleep(*backoff)
					continue
				}
				if len(elt) == 0 {
					elt = <-sink.ch
				}
				if ch == nil {
					if ch, err = s.dial(); err != nil {
						continue
					}
				}
				if ch != nil {
					err = ch.Publish(name, name, false, false, amqp.Publishing{Body: elt})
					if err == nil {
						elt = nil
					}
				}
			}
		}(name, sink)
	}

	for i := 0; i < *threads; i++ {
		go func() {
			for s.alive {
				elt := <-s.ch
				for _, sink := range s.Sinks {
					if sink.regex != nil && !sink.regex.Match(elt) {
						continue
					}
					sink.ch <- elt
				}
			}
		}()
	}
	go s.consumer()
	return nil
}

func NewDispatcher(jobQueue <-chan amqp.Delivery, worker Worker, replyChannel *amqp.Channel) *Dispatcher {
	return &Dispatcher{
		jobQueue:     jobQueue,
		worker:       worker,
		replyChannel: replyChannel,
		confirms:     replyChannel.NotifyPublish(make(chan amqp.Confirmation, 1)),
	}
}

func QueueBind(q *amqp.Queue, exchange string, ch *amqp.Channel) error {
	return ch.QueueBind(
		q.Name,   // queue name
		"",       // routing key
		exchange, // exchange
		false,
		nil)
}

/*
Run ...
*/
func (c *Consumer) Run(ch *amqp.Channel) chan []common.MapStr {
	events := make(chan []common.MapStr)
	// TODO: make exchange part of config
	deliveries, err := ch.Consume("test", *c.cfg.Name, false, false, false, false, nil)
	utils.FailOnError(err, "Failed to register consumer")
	go c.consume(deliveries, events)
	return events
}