func (e *esStateService) readObject(objid string) (o *object, err error) {
	conn := elastigo.NewConn()
	defer conn.Close()
	conn.Domain = e.stateDomain

	template := `{
		"query": {
			"term": {
				"object_id": "%v"
			}
		}
	}`
	tempbuf := fmt.Sprintf(template, objid)
	res, err := conn.Search(e.stateIndex, "geomodel_state", nil, tempbuf)
	if err != nil {
		return o, err
	}
	if res.Hits.Len() == 0 {
		return nil, nil
	}
	if res.Hits.Len() > 1 {
		return nil, fmt.Errorf("consistency failure, more than one object matched")
	}
	o = &object{}
	err = json.Unmarshal(*res.Hits.Hits[0].Source, o)
	if err != nil {
		return nil, err
	}

	return o, nil
}

// Initializes a new elastic search handler.
func NewElastic(host string, index string) (*Elastic, error) {
	c := elastigo.NewConn()
	c.Domain = host
	b := c.NewBulkIndexer(10)
	b.Start()
	return &Elastic{c, b, index}, nil
}

func InitElasticsearch() error {
	es = elastigo.NewConn()
	es.Domain = setting.Config.ElasticsearchDomain // needs to be configurable obviously
	es.Port = strconv.Itoa(setting.Config.ElasticsearchPort)
	if setting.Config.ElasticsearchUser != "" && setting.Config.ElasticsearchPasswd != "" {
		es.Username = setting.Config.ElasticsearchUser
		es.Password = setting.Config.ElasticsearchPasswd
	}
	if exists, err := es.ExistsIndex("definitions", "metric", nil); err != nil && err.Error() != "record not found" {
		return err
	} else {
		if !exists {
			_, err = es.CreateIndex("definitions")
			if err != nil {
				return err
			}
		}
		esopts := elastigo.MappingOptions{}

		err = es.PutMapping("definitions", "metric", MetricDefinition{}, esopts)
		if err != nil {
			return err
		}
	}

	return nil
}

func (c *Configuration) StartElasticSearch() {

	address := c.Json.Elastic.HostAddress

	if !c.Json.RunLocalMode {
		if os.Getenv("CRASHMAT_ELASTICHOSTADDRESS") != "" {
			log.Println("Using environmental for CRASHMAT_ELASTICHOSTADDRESS")
			address = os.Getenv("CRASHMAT_ELASTICHOSTADDRESS")
		}
	}
	if c.Json.Elastic.DropAllOnStartUp {
		log.Println("Purging => ", address+"crashmat")
		out, err := exec.Command("curl", "-XDELETE", address+"crashmat").Output()
		if err != nil {
			log.Println(err)
		}
		fmt.Printf("%s", out)

		log.Println("Creating => ", address+"crashmat")
		out, err = exec.Command("curl", "-XPUT", address+"crashmat").Output()
		if err != nil {
			log.Println(err)
		}
	}

	elasticConnection := elastigo.NewConn()

	elasticConnection.SetFromUrl(address)

	ElasticConnection = elasticConnection
	c.ElasticConnection = elasticConnection
}

// NewServer returns a new instance of Server built from a config.
func NewServer(c *Config, buildInfo *BuildInfo) (*Server, error) {

	elasticsearch := elastigo.NewConn()
	elasticsearch.SetFromUrl(c.Meta.ElasticSearchUrl)

	db := gocql.NewCluster(c.Database.CassandraNodes...)
	db.Keyspace = c.Database.CassandraKeyspace
	db.NumConns = c.Database.CassandraConns
	db.Discovery = gocql.DiscoveryConfig{
		DcFilter:   "",
		RackFilter: "",
		Sleep:      30 * time.Second,
	}

	s := &Server{
		buildInfo: *buildInfo,
		err:       make(chan error),
		closing:   make(chan struct{}),

		Hostname:      c.Meta.Hostname,
		BindAddress:   c.Meta.BindAddress,
		elasticsearch: elasticsearch,
		cassandra:     db,
	}

	// Append services.
	//s.appendMongoService(c.Mongo)

	s.appendMetricsReportingService(c.Meta)
	s.appendHTTPDService(c.HTTPD)
	s.appendRegistrationService(c.Registration, c.RegMeta)
	return s, nil
}

func Init() {
	host := setting.Cfg.Section("elasticsearch").Key("host").MustString("localhost")
	port := setting.Cfg.Section("elasticsearch").Key("port").MustInt64(9200)
	es = elastigo.NewConn()
	es.Domain = host
	es.Port = strconv.FormatInt(port, 10)
}

func main() {
	c := elastigo.NewConn()
	log.SetFlags(log.LstdFlags)
	flag.Parse()

	fmt.Println("host = ", *host)
	// Set the Elasticsearch Host to Connect to
	c.Domain = *host

	// Index a document
	_, err := c.Index("testindex", "user", "docid_1", nil, `{"name":"bob"}`)
	exitIfErr(err)

	// Index a doc using a map of values
	_, err = c.Index("testindex", "user", "docid_2", nil, map[string]string{"name": "venkatesh"})
	exitIfErr(err)

	// Index a doc using Structs
	_, err = c.Index("testindex", "user", "docid_3", nil, MyUser{"wanda", 22})
	exitIfErr(err)

	// Search Using Raw json String
	searchJson := `{
	    "query" : {
	        "term" : { "Name" : "wanda" }
	    }
	}`
	out, err := c.Search("testindex", "user", nil, searchJson)
	if len(out.Hits.Hits) == 1 {
		fmt.Println("%v", out.Hits.Hits[0].Source)
	}
	exitIfErr(err)

}

func main() {
	kingpin.Parse()

	r := Rss2{}

	response, err := http.DefaultClient.Get(*feedurl)
	if err != nil {
		log.Fatal(err)
	}

	xmlContent, err := ioutil.ReadAll(response.Body)
	err = xml.Unmarshal(xmlContent, &r)
	if err != nil {
		log.Fatal(err)
	}

	client := elastigo.NewConn()

	if *esurl != "" {
		client.SetFromUrl(*esurl)
	}

	for _, item := range r.ItemList {
		jsonValue, _ := json.MarshalIndent(item, "", "    ")
		if *esurl != "" {
			client.Index(*esindex, *estype, item.GUID, nil, jsonValue)
		} else {
			fmt.Println(string(jsonValue))
		}
	}
}

// The simplest usage of background bulk indexing
func ExampleBulkIndexer_simple() {
	c := elastigo.NewConn()

	indexer := c.NewBulkIndexerErrors(10, 60)
	indexer.Start()
	indexer.Index("twitter", "user", "1", "", "", nil, `{"name":"bob"}`, true)
	indexer.Stop()
}

func GetInstance(addr string, port int) *ElasticSearchStorage {
	c := elastigo.NewConn()
	c.Domain = addr
	c.Port = strconv.FormatInt(int64(port), 10)

	storage := &ElasticSearchStorage{connection: c}
	return storage
}

// connects to an NSQ topic and emits each message into streamtools.
func (b *ToElasticsearch) Run() {
	var err error
	var esIndex string
	var esType string

	conn := elastigo.NewConn()

	host := "localhost"
	port := "9200"

	for {
		select {
		case ruleI := <-b.inrule:
			host, err = util.ParseString(ruleI, "Host")
			if err != nil {
				b.Error(err)
				break
			}

			port, err = util.ParseString(ruleI, "Port")
			if err != nil {
				b.Error(err)
				break
			}

			esIndex, err = util.ParseString(ruleI, "Index")
			if err != nil {
				b.Error(err)
				break
			}

			esType, err = util.ParseString(ruleI, "Type")
			if err != nil {
				b.Error(err)
				break
			}

			conn.Domain = host
			conn.Port = port

		case msg := <-b.in:
			_, err := conn.Index(esIndex, esType, "", nil, msg)
			if err != nil {
				b.Error(err)
			}
		case <-b.quit:
			return
		case c := <-b.queryrule:
			c <- map[string]interface{}{
				"Host":  host,
				"Port":  port,
				"Index": esIndex,
				"Type":  esType,
			}
		}
	}
}

func (bdb *BloomDatabase) SearchConnection() *elastigo.Conn {
	if bdb.sharedSearch == nil {
		conn := elastigo.NewConn()
		conn.SetHosts(bdb.searchHosts)
		bdb.sharedSearch = conn
	}

	return bdb.sharedSearch
}

func (e *esStateService) writeObject(o object) (err error) {
	conn := elastigo.NewConn()
	defer conn.Close()
	conn.Domain = e.stateDomain
	_, err = conn.Index(e.stateIndex, "geomodel_state", o.ObjectID, nil, o)
	if err != nil {
		return err
	}
	return nil
}

func InitElasticsearch() error {
	es = elastigo.NewConn()
	es.Domain = setting.Config.ElasticsearchDomain // needs to be configurable obviously
	es.Port = strconv.Itoa(setting.Config.ElasticsearchPort)
	if setting.Config.ElasticsearchUser != "" && setting.Config.ElasticsearchPasswd != "" {
		es.Username = setting.Config.ElasticsearchUser
		es.Password = setting.Config.ElasticsearchPasswd
	}

	return nil
}

/*
   Create the index if it does not exist.
   Optionally apply a mapping if mapping file is supplied.
*/
func NewExtendedEssConn(esshost string, essport int) *ExtendedEssConn {

	ed := ExtendedEssConn{
		Conn: elastigo.NewConn(),
	}

	ed.Domain = esshost
	ed.Port = fmt.Sprintf("%d", essport)

	return &ed
}

func main() {
	c := elastigo.NewConn()
	log.SetFlags(log.LstdFlags)
	flag.Parse()

	// Trace all requests
	c.RequestTracer = func(method, url, body string) {
		log.Printf("Requesting %s %s", method, url)
		log.Printf("Request body: %s", body)
	}

	fmt.Println("host = ", *host)
	// Set the Elasticsearch Host to Connect to
	c.Domain = *host

	// Index a document
	_, err := c.Index("oldindex", "product", "docid_1", nil, `{"name":"bob"}`)
	exitIfErr(err)

	// Index a doc using a map of values
	_, err = c.Index("oldindex", "product", "docid_2", nil, map[string]string{"name": "venkatesh"})
	exitIfErr(err)

	// Index a doc using Structs
	_, err = c.Index("testindex", "user", "docid_3", nil, map[string]string{"name": "wena"})
	exitIfErr(err)

	// "name":"peanuts","description":"Honey Roasted peanuts","permalink":"",
	//             "tax_category_id":0,"shipping_category_id":0,"deleted_at":"0001-01-01T00:00:00Z","meta_description":"",
	//             "meta_keywords":"","position":0,"is_featured":false,"can_discount":false,"distributor_only_membership":false

	// product := models.Product{ 59001, "peanuts", "Honey Roasted peanuts", "Good taste food"}
	//
	//
	// // Index a doc using Structs
	// _, err = c.Index("pdtindex", "product", "producid_3", nil, product)
	//     fmt.Println(product)

	//exitIfErr(err)

	// Search Using Raw json String
	searchJson := `{
	    "query" : {
	        "term" : { "Name" : "venkatesh" }
	    }
	}`
	out, err := c.Search("oldindex", "product", nil, searchJson)

	if len(out.Hits.Hits) == 1 {
		fmt.Println("%v", out.Hits.Hits[0].Source)
	}
	exitIfErr(err)

}

func NewESClient() DbClient {
	conn := elastigo.NewConn()
	host := eshost1
	conn.Domain = host

	batchconn := goes.NewConnection(host, "9200")
	return &ESClient{
		conn:      conn,
		batchconn: batchconn,
	}
}

func (elasticSearchClient *ElasticSearchClient) GetConnection() *elastigo.Conn {
	if elasticSearchClient.connection != nil {
		return elasticSearchClient.connection
	} else {
		c := elastigo.NewConn()
		c.SetHosts(elasticSearchClient.host)
		c.SetPort(strconv.Itoa(elasticSearchClient.port))

		elasticSearchClient.connection = c
		return elasticSearchClient.connection
	}
}

// Consumes messages from NSQ and publishes to ElasticSearch
func NSQToES() error {
	nc, err := nsq.NewConsumer(Topic, Channel, nsq.NewConfig())
	if err != nil {
		return err
	}
	tn := &nsqToES{
		conn: el.NewConn(),
	}
	tn.conn.Domain = ESAddress

	nc.AddConcurrentHandlers(tn, NumHandlers)
	return nc.ConnectToNSQD(NSQAddress)
}

// Configure initializes this producer with values from a plugin config.
func (prod *ElasticSearch) Configure(conf core.PluginConfig) error {
	err := prod.ProducerBase.Configure(conf)
	if err != nil {
		return err
	}

	prod.SetStopCallback(prod.close)

	defaultServer := []string{"localhost"}
	numConnections := conf.GetInt("Connections", 6)
	retrySec := conf.GetInt("RetrySec", 5)

	prod.conn = elastigo.NewConn()
	prod.conn.Hosts = conf.GetStringArray("Servers", defaultServer)
	prod.conn.Domain = conf.GetString("Domain", prod.conn.Hosts[0])
	prod.conn.ClusterDomains = prod.conn.Hosts
	prod.conn.Port = strconv.Itoa(conf.GetInt("Port", 9200))
	prod.conn.Username = conf.GetString("User", "")
	prod.conn.Password = conf.GetString("Password", "")

	prod.indexer = prod.conn.NewBulkIndexerErrors(numConnections, retrySec)
	prod.indexer.BufferDelayMax = time.Duration(conf.GetInt("BatchTimeoutSec", 5)) * time.Second
	prod.indexer.BulkMaxBuffer = conf.GetInt("BatchSizeByte", 32768)
	prod.indexer.BulkMaxDocs = conf.GetInt("BatchMaxCount", 128)

	prod.indexer.Sender = func(buf *bytes.Buffer) error {
		_, err := prod.conn.DoCommand("POST", "/_bulk", nil, buf)
		if err != nil {
			Log.Error.Print("ElasticSearch response error - ", err)
		}
		return err
	}

	prod.index = conf.GetStreamMap("Index", "")
	prod.msgType = conf.GetStreamMap("Type", "log")
	prod.msgTTL = conf.GetString("TTL", "")
	prod.dayBasedIndex = conf.GetBool("DayBasedIndex", false)

	prod.counters = make(map[string]*int64)
	prod.lastMetricUpdate = time.Now()

	for _, index := range prod.index {
		shared.Metric.New(elasticMetricMessages + index)
		shared.Metric.New(elasticMetricMessagesSec + index)
		prod.counters[index] = new(int64)
	}

	prod.SetCheckFuseCallback(prod.isClusterUp)
	return nil
}