func handlePacket(buffer []byte) {
	parser := rfc5424.NewParser(buffer)
	err := parser.Parse()

	if err != nil {
		fmt.Printf("Error reading syslog message %s", err)
		return
	}

	log := parser.Dump()
	log["@timestamp"] = log["timestamp"]
	log["facility_label"] = FACILITY_LABELS[(log["facility"]).(int)]
	log["severity_label"] = SEVERITY_LABELS[(log["severity"]).(int)]
	log["type"] = "syslog"

	now := time.Now()
	index := "logstash-" + now.Format("2006.01.02")

	_, err = elasticSearch.Index(true, index, "logs", "", log)
	if err != nil {
		fmt.Printf("Error indexing message %s", err)
		return
	}
	fmt.Println("Logged")
}

func main() {
	core.DebugRequests = true
	log.SetFlags(log.LstdFlags)
	flag.Parse()

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

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

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

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

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

}

func main() {

	type Address struct {
		ProvinceName string `json:"province_name"`
	}

	api.Domain = "localhost"
	//api.Port = "9300"

	core.Index("data", "address", "1", nil, Address{"ประเทศไทย"})
	core.Index("data", "address", "2", nil, Address{"เพลงไทย"})

}

func (publisher *PublisherType) PublishMysqlTransaction(t *MysqlTransaction) error {
	// Set the Elasticsearch Host to Connect to
	api.Domain = publisher.mother_host
	api.Port = publisher.mother_port

	index := fmt.Sprintf("packetbeat-%d.%02d.%02d", t.ts.Year(), t.ts.Month(), t.ts.Day())

	status := t.Mysql["error_message"].(string)
	if len(status) == 0 {
		status = "OK"
	}

	src_server := publisher.GetServerName(t.Src.Ip)
	dst_server := publisher.GetServerName(t.Dst.Ip)

	// add Mysql transaction
	_, err := core.Index(index, "mysql", "", nil, Event{
		t.ts, "mysql", t.Src.Ip, t.Src.Port, t.Src.Proc, "", src_server,
		t.Dst.Ip, t.Dst.Port, t.Dst.Proc, dst_server,
		t.ResponseTime, status, t.Request_raw, t.Response_raw,
		t.Mysql, nil, nil})

	DEBUG("publish", "Sent MySQL transaction [%s->%s]:\n%s", t.Src.Proc, t.Dst.Proc, t.Mysql)

	return err

}

func (publisher *PublisherType) PublishRedisTransaction(t *RedisTransaction) error {
	var err error
	index := fmt.Sprintf("packetbeat-%d.%02d.%02d", t.ts.Year(), t.ts.Month(), t.ts.Day())

	status := "OK"

	src_server := publisher.GetServerName(t.Src.Ip)
	dst_server := publisher.GetServerName(t.Dst.Ip)

	if _Config.Agent.Ignore_outgoing && dst_server != "" &&
		dst_server != publisher.name {
		// duplicated transaction -> ignore it
		DEBUG("publish", "Ignore duplicated REDIS transaction on %s: %s -> %s", publisher.name, src_server, dst_server)
		return nil
	}

	event := Event{
		t.ts, "redis", t.Src.Ip, t.Src.Port, t.Src.Proc, "", src_server,
		t.Dst.Ip, t.Dst.Port, t.Dst.Proc, dst_server,
		t.ResponseTime, status, t.Request_raw, t.Response_raw,
		nil, nil, t.Redis}

	if IS_DEBUG("publish") {
		PrintPublishEvent(&event)
	}

	// add Redis transaction
	if !publisher.disabled {
		_, err = core.Index(index, "redis", "", nil, event)
	}

	return err
}

func main() {
	api.Domain = "localhost"

	// connect to ZeroMQ
	subscriber, _ := zmq.NewSocket(zmq.SUB)
	defer subscriber.Close()
	subscriber.Connect("tcp://38.86.38.213:12777")
	// subscriber.Connect("tcp://127.0.0.1:12777")
	subscriber.SetSubscribe("")

	// endless looping!
	for {
		var p Packet

		// get stuff from ZeroMQ
		msg, _ := subscriber.Recv(0)
		// fmt.Println(msg)

		// decode the JSON aprs packet in to struct p
		err := json.Unmarshal([]byte(msg), &p)
		log.Println(err)

		// re-encode digipeters back to json
		// digipeters, err := json.Marshal(p.Digipeaters)
		// wx, err := json.Marshal(p.Wx)

		t := time.Now()
		response, _ := core.Index("packets", "packet", "", t, p)

		fmt.Println(response)
	}
}

func CreateGeoJotHandler(w http.ResponseWriter, r *http.Request) {
	// Parse the incoming geojot from the request body
	var geojotJSON GeoJotJSON
	err := json.NewDecoder(r.Body).Decode(&geojotJSON)
	if err != nil {
		serveError(w, err)
	}

	geojot := geojotJSON.GeoJot

	// add single go struct entity
	response, _ := core.Index(true, es_index, es_index_type, "", geojot)

	// Grab the Id for the client
	geojot.Id = response.Id

	// Serialize the modified geojot to JSON
	j, err := json.Marshal(GeoJotJSON{GeoJot: geojot})
	if err != nil {
		serveError(w, err)
	}

	w.Header().Set("Content-Type", "application/json")
	w.Write(j)
}

// for testing
func main() {
	flag.Parse()
	log.SetFlags(log.Ltime | log.Lshortfile)
	api.Domain = *eshost
	response, _ := core.Index(true, "twitter", "tweet", "1", NewTweet("kimchy", "Search is cool"))
	indices.Flush()
	log.Printf("Index OK: %v", response.Ok)
	searchresponse, err := core.SearchRequest(true, "twitter", "tweet", "{\"query\" : {\"term\" : { \"user\" : \"kimchy\" }}}", "", 0)
	if err != nil {
		log.Println("error during search:" + err.Error())
		log.Fatal(err)
	}
	// try marshalling to tweet type
	var t Tweet
	json.Unmarshal(searchresponse.Hits.Hits[0].Source, t)
	log.Printf("Search Found: %s", t)
	response, _ = core.Get(true, "twitter", "tweet", "1")
	log.Printf("Get: %v", response.Exists)
	exists, _ := core.Exists(true, "twitter", "tweet", "1")
	log.Printf("Exists: %v", exists)
	indices.Flush()
	countResponse, _ := core.Count(true, "twitter", "tweet")
	log.Printf("Count: %v", countResponse.Count)
	response, _ = core.Delete(true, "twitter", "tweet", "1", -1, "")
	log.Printf("Delete OK: %v", response.Ok)
	response, _ = core.Get(true, "twitter", "tweet", "1")
	log.Printf("Get: %v", response.Exists)

	healthResponse, _ := cluster.Health(true)
	log.Printf("Health: %v", healthResponse.Status)

	cluster.State("transient", "discovery.zen.minimum_master_nodes", 2)

}

// Run is the block's main loop. Here we listen on the different channels we set up.
// This block posts a message to a specified Elasticsearch index with the given type.
func (b *ToElasticsearch) Run() {
	var err error
	var index string
	var indextype string

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

	for {
		select {
		case msgI := <-b.inrule:
			host, err = util.ParseString(msgI, "Host")
			if err != nil {
				b.Error(err)
				continue
			}
			port, err = util.ParseString(msgI, "Port")
			if err != nil {
				b.Error(err)
				continue
			}
			index, err = util.ParseString(msgI, "Index")
			if err != nil {
				b.Error(err)
				continue
			}
			indextype, err = util.ParseString(msgI, "IndexType")
			if err != nil {
				b.Error(err)
				continue
			}

			// Set the Elasticsearch Host/Port to Connect to
			api.Domain = host
			api.Port = port

		case MsgChan := <-b.queryrule:
			// deal with a query request
			MsgChan <- map[string]interface{}{
				"Host":      host,
				"Port":      port,
				"Index":     index,
				"IndexType": indextype,
			}
		case <-b.quit:
			// quit the block
			return
		case msg := <-b.in:
			var args map[string]interface{}
			_, err := core.Index(index, indextype, "", args, msg)
			if err != nil {
				b.Error(err)
			}
		}
	}
}

// value should be JSON serializable.
func (d *Datastore) Put(key ds.Key, value interface{}) (err error) {
	id := d.KeyHash(key)
	res, err := core.Index(false, d.Index(key), key.Type(), id, value)
	if err != nil {
		return err
	}
	if !res.Ok {
		return fmt.Errorf("Elasticsearch response: NOT OK. %v", res)
	}
	return nil
}

func DispatchLogs(es eventsource.EventSource) {
	for {
		log := <-LogPipe
		resp, err := core.Index(true, "logs", "log", "", log)
		if err != nil {
			fmt.Println("EINDEXING: ", err)
		} else {
			es.SendMessage(log.URL, "", resp.Id) // I should add a type
			indices.Flush()
		}
	}
}

// doSyncUsers synchronize the content of the specified user file with the
// Elastic Search backend.
func doSyncUsers(c *cli.Context) {
	_ = ParseConfigOrDie(c.GlobalString("config"))

	var userData map[string]UserData
	if _, err := toml.DecodeFile(c.String("file"), &userData); err != nil {
		log.Fatal(err)
	}

	for login, data := range userData {
		fmt.Printf("Saving data for %q: %#v\n", login, data)
		if _, err := core.Index(UserIndex, UserType, login, nil, data); err != nil {
			log.Errorf("indexing data for %q; %v", login, err)
		}
	}
}

func (publisher *PublisherType) PublishHttpTransaction(t *HttpTransaction) error {
	// Set the Elasticsearch Host to Connect to
	api.Domain = publisher.mother_host
	api.Port = publisher.mother_port

	index := fmt.Sprintf("packetbeat-%d.%02d.%02d", t.ts.Year(), t.ts.Month(), t.ts.Day())

	status := t.Http["response"].(bson.M)["phrase"].(string)

	src_server := publisher.GetServerName(t.Src.Ip)
	dst_server := publisher.GetServerName(t.Dst.Ip)

	if _Config.Agent.Ignore_outgoing && dst_server != "" &&
		dst_server != publisher.name {
		// duplicated transaction -> ignore it
		DEBUG("publish", "Ignore duplicated HTTP transaction on %s: %s -> %s", publisher.name, src_server, dst_server)
		return nil
	}

	var src_country = ""
	if _GeoLite != nil {
		if len(src_server) == 0 { // only for external IP addresses
			loc := _GeoLite.GetLocationByIP(t.Src.Ip)
			if loc != nil {
				src_country = loc.CountryCode
			}
		}
	}

	event := Event{
		t.ts, "http", t.Src.Ip, t.Src.Port, t.Src.Proc, src_country, src_server,
		t.Dst.Ip, t.Dst.Port, t.Dst.Proc, dst_server,
		t.ResponseTime, status, t.Request_raw, t.Response_raw,
		nil, t.Http, nil}

	if IS_DEBUG("publish") {
		PrintPublishEvent(&event)
	}

	// add Http transaction
	_, err := core.Index(index, "http", "", nil, event)

	return err

}

// Posts a message to a specified Elasticsearch index with the given type.
func ToElasticsearch(b *Block) {

	type toElasticsearchRule struct {
		Host  string
		Port  string
		Index string
		Type  string
	}

	var rule *toElasticsearchRule

	// TODO check the endpoint for happiness
	for {
		select {
		case msg := <-b.Routes["set_rule"]:
			if rule == nil {
				rule = &toElasticsearchRule{}
			}
			unmarshal(msg, rule)
			// Set the Elasticsearch Host/Port to Connect to
			api.Domain = rule.Host
			api.Port = rule.Port

		case msg := <-b.Routes["get_rule"]:
			if rule == nil {
				marshal(msg, &toElasticsearchRule{})
			} else {
				marshal(msg, rule)
			}
		case <-b.QuitChan:
			quit(b)
			return
		case msg := <-b.InChan:
			if rule == nil {
				break
			}

			_, err := core.Index(true, rule.Index, rule.Type, "", msg.Msg)
			if err != nil {
				log.Println(err.Error())
				break
			}
		}
	}
}

func (publisher *PublisherType) PublishEvent(ts time.Time, src *Endpoint, dst *Endpoint, event *Event) error {
	index := fmt.Sprintf("packetbeat-%d.%02d.%02d", ts.Year(), ts.Month(), ts.Day())

	event.Src_server = publisher.GetServerName(src.Ip)
	event.Dst_server = publisher.GetServerName(dst.Ip)

	if _Config.Agent.Ignore_outgoing && event.Dst_server != "" &&
		event.Dst_server != publisher.name {
		// duplicated transaction -> ignore it
		DEBUG("publish", "Ignore duplicated REDIS transaction on %s: %s -> %s", publisher.name, event.Src_server, event.Dst_server)
		return nil
	}

	event.Timestamp = ts
	event.Src_ip = src.Ip
	event.Src_port = src.Port
	event.Src_proc = src.Proc
	event.Dst_ip = dst.Ip
	event.Dst_port = dst.Port
	event.Dst_proc = dst.Proc

	// set src_country if no src_server is set
	event.Src_country = ""
	if _GeoLite != nil {
		if len(event.Src_server) == 0 { // only for external IP addresses
			loc := _GeoLite.GetLocationByIP(src.Ip)
			if loc != nil {
				event.Src_country = loc.CountryCode
			}
		}
	}

	if IS_DEBUG("publish") {
		PrintPublishEvent(event)
	}

	// add Redis transaction
	var err error
	if !publisher.disabled {
		_, err = core.Index(index, event.Type, "", nil, event)
	}

	return err
}

// for testing
func main() {
	flag.Parse()
	log.SetFlags(log.Ltime | log.Lshortfile)
	api.Domain = *eshost
	core.VerboseLogging = true
	response, _ := core.Index("twitter", "tweet", "1", nil, NewTweet("kimchy", "Search is cool"))
	indices.Flush()
	log.Printf("Index OK: %v", response.Ok)
	searchresponse, err := core.SearchRequest("twitter", "tweet", nil, "{\"query\" : {\"term\" : { \"user\" : \"kimchy\" }}}")
	if err != nil {
		log.Println("error during search:" + err.Error())
		log.Fatal(err)
	}
	// try marshalling to tweet type
	var t Tweet
	bytes, err := searchresponse.Hits.Hits[0].Source.MarshalJSON()
	if err != nil {
		log.Fatalf("err calling marshalJson:%v", err)
	}
	json.Unmarshal(bytes, t)
	log.Printf("Search Found: %s", t)
	response, _ = core.Get("twitter", "tweet", "1", nil)
	log.Printf("Get: %v", response.Exists)
	exists, _ := core.Exists("twitter", "tweet", "1", nil)
	log.Printf("Exists: %v", exists)
	indices.Flush()
	countResponse, _ := core.Count("twitter", "tweet", nil)
	log.Printf("Count: %v", countResponse.Count)
	response, _ = core.Delete("twitter", "tweet", "1", map[string]interface{}{"version": -1, "routing": ""})
	log.Printf("Delete OK: %v", response.Ok)
	response, _ = core.Get("twitter", "tweet", "1", nil)
	log.Printf("Get: %v", response.Exists)

	healthResponse, _ := cluster.Health()
	log.Printf("Health: %v", healthResponse.Status)

	cluster.UpdateSettings("transient", "discovery.zen.minimum_master_nodes", 2)

}

func (publisher *PublisherType) PublishMysqlTransaction(t *MysqlTransaction) error {
	// Set the Elasticsearch Host to Connect to
	api.Domain = publisher.mother_host
	api.Port = publisher.mother_port

	index := fmt.Sprintf("packetbeat-%d.%02d.%02d", t.ts.Year(), t.ts.Month(), t.ts.Day())

	status := t.Mysql["error_message"].(string)
	if len(status) == 0 {
		status = "OK"
	}

	src_server := publisher.GetServerName(t.Src.Ip)
	dst_server := publisher.GetServerName(t.Dst.Ip)

	if _Config.Agent.Ignore_outgoing && dst_server != "" &&
		dst_server != publisher.name {
		// duplicated transaction -> ignore it
		DEBUG("publish", "Ignore duplicated MySQL transaction on %s: %s -> %s", publisher.name, src_server, dst_server)
		return nil
	}

	event := Event{
		t.ts, "mysql", t.Src.Ip, t.Src.Port, t.Src.Proc, "", src_server,
		t.Dst.Ip, t.Dst.Port, t.Dst.Proc, dst_server,
		t.ResponseTime, status, t.Request_raw, t.Response_raw,
		t.Mysql, nil, nil}

	if IS_DEBUG("publish") {
		PrintPublishEvent(&event)
	}

	// add Mysql transaction
	_, err := core.Index(index, "mysql", "", nil, event)

	return err

}

// Run is the block's main loop. Here we listen on the different channels we set up.
// This block posts a message to a specified Elasticsearch index with the given type.
func (b *ToElasticsearch) Run() {
	for {
		select {
		case msgI := <-b.inrule:
			host, _ := util.ParseString(msgI, "Host")
			port, _ := util.ParseString(msgI, "Port")
			index, _ := util.ParseString(msgI, "Index")
			indextype, _ := util.ParseString(msgI, "IndexType")

			// Set the Elasticsearch Host/Port to Connect to
			api.Domain = host
			api.Port = port

			b.host = host
			b.port = port
			b.index = index
			b.indextype = indextype

		case respChan := <-b.queryrule:
			// deal with a query request
			respChan <- map[string]interface{}{
				"Host":      b.host,
				"Port":      b.port,
				"Index":     b.index,
				"IndexType": b.indextype,
			}
		case <-b.quit:
			// quit the block
			return
		case msg := <-b.in:
			var args map[string]interface{}
			_, err := core.Index(b.index, b.indextype, "", args, msg)
			if err != nil {
				b.Error(err)
			}
		}
	}
}

func (publisher *PublisherType) PublishRedisTransaction(t *RedisTransaction) error {
	// Set the Elasticsearch Host to Connect to
	api.Domain = publisher.mother_host
	api.Port = publisher.mother_port

	index := fmt.Sprintf("packetbeat-%d.%02d.%02d", t.ts.Year(), t.ts.Month(), t.ts.Day())

	status := "OK"

	src_server := publisher.GetServerName(t.Src.Ip)
	dst_server := publisher.GetServerName(t.Dst.Ip)

	// add Redis transaction
	_, err := core.Index(index, "redis", "", nil, Event{
		t.ts, "redis", t.Src.Ip, t.Src.Port, t.Src.Proc, "", src_server,
		t.Dst.Ip, t.Dst.Port, t.Dst.Proc, dst_server,
		t.ResponseTime, status, t.Request_raw, t.Response_raw,
		nil, nil, t.Redis})

	DEBUG("publish", "Sent Redis transaction [%s->%s]:\n%s", t.Src.Proc, t.Dst.Proc, t.Redis)
	return err

}

func (publisher *PublisherType) PublishTopology(params ...string) error {

	var localAddrs []string = params

	if len(params) == 0 {
		addrs, err := LocalAddrs()
		if err != nil {
			ERR("Getting local IP addresses fails with: %s", err)
			return err
		}
		localAddrs = addrs
	}

	// delete old IP addresses
	searchJson := fmt.Sprintf("{query: {term: {name: %s}}}", strconv.Quote(publisher.name))
	res, err := core.SearchRequest("packetbeat-topology", "server-ip", nil, searchJson)
	if err == nil {
		for _, server := range res.Hits.Hits {

			var top Topology
			err = json.Unmarshal([]byte(*server.Source), &top)
			if err != nil {
				ERR("Failed to unmarshal json data: %s", err)
			}
			if !stringInSlice(top.Ip, localAddrs) {
				res, err := core.Delete("packetbeat-topology", "server-ip" /*id*/, top.Ip, nil)
				if err != nil {
					ERR("Failed to delete the old IP address from packetbeat-topology")
				}
				if !res.Ok {
					ERR("Fail to delete old topology entry")
				}
			}

		}
	}

	// add new IP addresses
	for _, addr := range localAddrs {

		// check if the IP is already in the elasticsearch, before adding it
		found, err := core.Exists("packetbeat-topology", "server-ip" /*id*/, addr, nil)
		if err != nil {
			ERR("core.Exists fails with: %s", err)
		} else {

			if !found {
				res, err := core.Index("packetbeat-topology", "server-ip" /*id*/, addr, nil,
					Topology{publisher.name, addr})
				if err != nil {
					return err
				}
				if !res.Ok {
					ERR("Fail to add new topology entry")
				}
			}
		}
	}

	DEBUG("publish", "Topology: name=%s, ips=%s", publisher.name, strings.Join(localAddrs, " "))

	// initialize local topology map
	publisher.TopologyMap = make(map[string]string)

	return nil
}