func (l *RedisInput) Init(config inputs.MothershipConfig) error {

	l.Config = config

	if config.Host == "" {
		return errors.New("No Input Host specified")
	}
	l.Host = config.Host

	if config.Port == 0 {
		return errors.New("No Input Port specified")
	}
	l.Port = config.Port

	l.DB = config.DB

	if config.Key == "" {
		return errors.New("No Input Key specified")
	}
	l.Key = strings.TrimSpace(config.Key)

	if config.Type == "" {
		return errors.New("No Event Type specified")
	}
	l.Type = strings.TrimSpace(config.Type)

	logp.Debug("redisinput", "Using Host %s", l.Host)
	logp.Debug("redisinput", "Using Port %d", l.Port)
	logp.Debug("redisinput", "Using Database %d", l.DB)
	logp.Debug("redisinput", "Using Key %s", l.Key)
	logp.Debug("redisinput", "Adding Event Type %s", l.Type)

	return nil
}

func (l *RedisInput) averageSortedEvents(sorted_events map[string][]common.MapStr) ([]common.MapStr, error) {
	var output_events []common.MapStr
	var merged_event common.MapStr
	var metric_value_string string
	//var metric_value_bytes []byte
	metric_value := 0.0
	for _, events := range sorted_events {
		metric_value = 0.0
		merged_event = common.MapStr{}
		for _, event := range events {
			merged_event.Update(event)
			logp.Debug("groupstuff", "metric value: %v", event["metric_value"])
			metric_value_string = event["metric_value"].(string)
			//			metric_value_bytes = []byte(metric_value_string)
			//			metric_value += float64(common.Bytes_Ntohll(metric_value_bytes))
			metric_value_float, err := strconv.ParseFloat(metric_value_string, 65)
			if err != nil {
				logp.Err("Error parsing metric_value: %v", err)
			}
			metric_value += metric_value_float
		}
		logp.Debug("groupstuff", "the summed values is %v", metric_value)
		logp.Debug("groupstuff", "the length is %v", float64(len(events)))
		metric_value = metric_value / float64(len(events))
		logp.Debug("groupstuff", "the avg value is %v", metric_value)
		merged_event["metric_value"] = metric_value
		output_events = append(output_events, merged_event)
	}
	return output_events, nil
}

func FiltersRun(config common.MapStr, plugins map[Filter]FilterPlugin,
	next chan common.MapStr, stopCb func()) (input chan common.MapStr, err error) {

	logp.Debug("filters", "Initializing filters plugins")

	for filter, plugin := range plugins {
		Filters.Register(filter, plugin)
	}
	filters_plugins, err :=
		LoadConfiguredFilters(config)
	if err != nil {
		return nil, fmt.Errorf("Error loading filters plugins: %v", err)
	}
	logp.Debug("filters", "Filters plugins order: %v", filters_plugins)

	if len(filters_plugins) > 0 {
		runner := NewFilterRunner(next, filters_plugins)
		go func() {
			err := runner.Run()
			if err != nil {
				logp.Critical("Filters runner failed: %v", err)
				// shutting down
				stopCb()
			}
		}()
		input = runner.FiltersQueue
	} else {
		input = next
	}

	return input, nil
}

//TODO: Check for Errors Here
func (jsonexpander *JSONExpander) Filter(event common.MapStr) (common.MapStr, error) {
	text := event["message"]
	text_string := text.(*string)
	logp.Debug("jsonexpander", "Attempting to expand: %v", event)

	if isJSONString(*text_string) {
		data := []byte(*text_string)
		err := json.Unmarshal(data, &event)
		if err != nil {
			logp.Err("jsonexpander", "Could not expand json data")
			return event, nil
		}
	} else {
		logp.Debug("jsonexpander", "Message does not appear to be JSON data: %s", text_string)
	}

	now := func() time.Time {
		t := time.Now()
		return t
	}

	event.EnsureTimestampField(now)

	logp.Debug("jsonexpander", "Final Event: %v", event)
	return event, nil
}

// Each shipper publishes a list of IPs together with its name to Elasticsearch
func (out *ElasticsearchOutput) PublishIPs(name string, localAddrs []string) error {
	if !out.ttlEnabled {
		logp.Debug("output_elasticsearch", "Not publishing IPs because TTL was not yet confirmed to be enabled")
		return nil
	}

	logp.Debug("output_elasticsearch", "Publish IPs %s with expiration time %d", localAddrs, out.TopologyExpire)
	params := map[string]string{
		"ttl":     fmt.Sprintf("%dms", out.TopologyExpire),
		"refresh": "true",
	}
	_, err := out.Conn.Index(
		".packetbeat-topology", /*index*/
		"server-ip",            /*type*/
		name,                   /* id */
		params,                 /* parameters */
		PublishedTopology{name, strings.Join(localAddrs, ",")} /* body */)

	if err != nil {
		logp.Err("Fail to publish IP addresses: %s", err)
		return err
	}

	out.UpdateLocalTopologyMap()

	return nil
}

func (out *KafkaOutput) SendMessagesGoroutine() {

	for {
		select {

		case queueMsg := <-out.sendingQueue:

			if !out.connected {
				logp.Debug("output_kafka", "Droping pkt ...")
				continue
			}
			logp.Debug("output_kafka", "Send event to kafka")

			out.Producer.Input() <- &sarama.ProducerMessage{
				Topic: out.Topic,
				Key:   nil,
				Value: &queueMsg,
			}

		case err := <-out.Producer.Errors():
			logp.Err("Failed to publish event to kafka: %s", err)
			out.connected = false
			out.Close()
			go out.Reconnect()
			return
		}
	}
}

// Create an HTTP request and send it to Elasticsearch. The request is retransmitted max_retries
// before returning an error.
func (es *Elasticsearch) Request(method string, path string,
	params map[string]string, body interface{}) ([]byte, error) {

	var errors []error

	for attempt := 0; attempt < es.MaxRetries; attempt++ {

		conn := es.connectionPool.GetConnection()
		logp.Debug("elasticsearch", "Use connection %s", conn.Url)

		url := conn.Url + path
		if len(params) > 0 {
			url = url + "?" + UrlEncode(params)
		}

		logp.Debug("elasticsearch", "%s %s %s", method, url, body)

		var obj []byte
		var err error
		if body != nil {
			obj, err = json.Marshal(body)
			if err != nil {
				return nil, fmt.Errorf("Fail to JSON encode the body: %s", err)
			}
		} else {
			obj = nil
		}
		req, err := http.NewRequest(method, url, bytes.NewReader(obj))
		if err != nil {
			return nil, fmt.Errorf("NewRequest fails: %s", err)
		}

		resp, retry, err := es.PerformRequest(conn, req)
		if retry == true {
			// retry
			if err != nil {
				errors = append(errors, err)
			}
			continue
		}
		if err != nil {
			return nil, err
		}
		return resp, nil

	}

	logp.Warn("Request fails to be send after %d retries", es.MaxRetries)

	return nil, fmt.Errorf("Request fails after %d retries. Errors: %v", es.MaxRetries, errors)
}

func (l *RedisInput) Run(output chan common.MapStr) error {
	logp.Debug("redisinput", "Running Redis Input")
	var keysScript = redis.NewScript(1, `return redis.call('KEYS', KEYS[1])`)

	go func() {
		redisURL := fmt.Sprintf("redis://%s:%d/%d", l.Host, l.Port, l.DB)
		dialConnectTimeout := redis.DialConnectTimeout(3 * time.Second)
		dialReadTimeout := redis.DialReadTimeout(10 * time.Second)
		var backOffCount = 0
		var backOffDuration time.Duration = 5 * time.Second
		for {
			logp.Debug("redisinput", "Connecting to: %s", redisURL)
			server, err := redis.DialURL(redisURL, dialConnectTimeout, dialReadTimeout)
			if err != nil {
				logp.Err("couldn't start listening: " + err.Error())
				return
			}
			logp.Debug("redisinput", "Connected to Redis Server")

			reply, err := keysScript.Do(server, "*")
			if err != nil {
				logp.Err("An error occured while executing KEYS command: %s\n", err)
				return
			}

			keys, err := redis.Strings(reply, err)
			if err != nil {
				logp.Err("An error occured while converting reply to String: %s\n", err)
				return
			}

			for _, key := range keys {
				logp.Debug("redisinput", "key is %s", key)
				lineCount, err := l.handleConn(server, output, key)
				if err == nil {
					logp.Debug("redisinput", "Read %v events", lineCount)
					backOffCount = 0
					backOffDuration = time.Duration(backOffCount) * time.Second
					time.Sleep(backOffDuration)
				} else {
					backOffCount++
					backOffDuration = time.Duration(backOffCount) * time.Second
					time.Sleep(backOffDuration)
				}
			}
			defer server.Close()
		}
	}()
	return nil
}

// Create a HTTP request containing a bunch of operations and send them to Elasticsearch.
// The request is retransmitted up to max_retries before returning an error.
func (es *Elasticsearch) BulkRequest(method string, path string,
	params map[string]string, body chan interface{}) ([]byte, error) {

	var buf bytes.Buffer
	enc := json.NewEncoder(&buf)
	for obj := range body {
		enc.Encode(obj)
	}

	if buf.Len() == 0 {
		logp.Debug("elasticsearch", "Empty channel. Wait for more data.")
		return nil, nil
	}

	var errors []error

	for attempt := 0; attempt < es.MaxRetries; attempt++ {

		conn := es.connectionPool.GetConnection()
		logp.Debug("elasticsearch", "Use connection %s", conn.Url)

		url := conn.Url + path
		if len(params) > 0 {
			url = url + "?" + UrlEncode(params)
		}
		logp.Debug("elasticsearch", "Sending bulk request to %s", url)

		req, err := http.NewRequest(method, url, &buf)
		if err != nil {
			return nil, fmt.Errorf("NewRequest fails: %s", err)
		}

		resp, retry, err := es.PerformRequest(conn, req)
		if retry == true {
			// retry
			if err != nil {
				errors = append(errors, err)
			}
			continue
		}
		if err != nil {
			return nil, fmt.Errorf("PerformRequest fails: %s", err)
		}
		return resp, nil
	}

	logp.Warn("Request fails to be send after %d retries", es.MaxRetries)

	return nil, fmt.Errorf("Request fails after %d retries. Errors: %v", es.MaxRetries, errors)
}

func (l *StdinInput) doStuff(output chan common.MapStr) {
	reader := bufio.NewReader(os.Stdin)
	buffer := new(bytes.Buffer)

	var source string = fmt.Sprintf("%s:%s", os.Getenv("REMOTE_HOST"), os.Getenv("REMOTE_PORT"))
	var ssl_client_dn string = os.Getenv("SSL_CLIENT_DN")
	var offset int64 = 0
	var line uint64 = 0
	var read_timeout = 10 * time.Second

	logp.Debug("stdinput", "Handling New Connection from %s", source)

	now := func() time.Time {
		t := time.Now()
		return t
	}

	for {
		text, bytesread, err := l.readline(reader, buffer, read_timeout)

		if err != nil {
			logp.Info("Unexpected state reading from %v; error: %s\n", os.Getenv("SSL_CLIENT_DN"), err)
			return
		}

		logp.Debug("stdinputlines", "New Line: %s", &text)

		line++

		event := common.MapStr{}
		event["ssl_client_dn"] = &ssl_client_dn
		event["source"] = &source
		event["offset"] = offset
		event["line"] = line
		event["message"] = text
		event["type"] = l.Type

		event.EnsureTimestampField(now)
		event.EnsureCountField()

		offset += int64(bytesread)

		logp.Debug("stdinput", "InputEvent: %v", event)
		output <- event // ship the new event downstream
		os.Stdout.Write([]byte("OK"))
	}
	logp.Debug("stdinput", "Closed Connection from %s", source)
}

// If a connection fails, it will be marked as dead and put on timeout.
// timeout = default_timeout * 2 ** (fail_count - 1)
// When the timeout is over, the connection will be resurrected and
// returned to the live pool
func (pool *ConnectionPool) MarkDead(conn *Connection) error {

	if !conn.dead {
		logp.Debug("elasticsearch", "Mark dead %s", conn.Url)
		conn.dead = true
		conn.dead_count = conn.dead_count + 1
		timeout := pool.Dead_timeout * time.Duration(math.Pow(2, float64(conn.dead_count)-1))
		conn.timer = time.AfterFunc(timeout*time.Second, func() {
			// timeout expires
			conn.dead = false
			logp.Debug("elasticsearch", "Timeout expired. Mark it as alive: %s", conn.Url)
		})
	}

	return nil
}

// Handles OS signals that ask the service/daemon to stop.
// The stopFunction should break the loop in the Beat so that
// the service shut downs gracefully.
func HandleSignals(stopFunction func()) {
	// On ^C or SIGTERM, gracefully stop the sniffer
	sigc := make(chan os.Signal, 1)
	signal.Notify(sigc, syscall.SIGINT, syscall.SIGTERM)
	go func() {
		<-sigc
		logp.Debug("service", "Received sigterm/sigint, stopping")
		stopFunction()
	}()

	// Handle the Windows service events
	go ProcessWindowsControlEvents(func() {
		logp.Debug("service", "Received svc stop/shutdown request")
		stopFunction()
	})
}

func (l *ProcfsInput) Run(output chan common.MapStr) error {
	logp.Debug("[procfsinput]", "Starting up Procfs Input")

	go inputs.PeriodicTaskRunner(l, output, runTick, runMinor, runMajor)

	return nil
}

// Update the local topology map
func (out *ElasticsearchOutput) UpdateLocalTopologyMap() {

	// get all shippers IPs from Elasticsearch
	TopologyMapTmp := make(map[string]string)

	res, err := out.Conn.SearchUri(".packetbeat-topology", "server-ip", nil)
	if err == nil {
		for _, obj := range res.Hits.Hits {
			var result QueryResult
			err = json.Unmarshal(obj, &result)
			if err != nil {
				return
			}

			var pub PublishedTopology
			err = json.Unmarshal(result.Source, &pub)
			if err != nil {
				logp.Err("json.Unmarshal fails with: %s", err)
			}
			// add mapping
			ipaddrs := strings.Split(pub.IPs, ",")
			for _, addr := range ipaddrs {
				TopologyMapTmp[addr] = pub.Name
			}
		}
	} else {
		logp.Err("Getting topology map fails with: %s", err)
	}

	// update topology map
	out.TopologyMap = TopologyMapTmp

	logp.Debug("output_elasticsearch", "Topology map %s", out.TopologyMap)
}

// Publish an event by adding it to the queue of events.
func (out *ElasticsearchOutput) PublishEvent(ts time.Time, event common.MapStr) error {

	out.sendingQueue <- EventMsg{Ts: ts, Event: event}

	logp.Debug("output_elasticsearch", "Publish event: %s", event)
	return nil
}

func (l *TailInput) doStuff(output chan common.MapStr) {
	now := func() time.Time {
		t := time.Now()
		return t
	}

	var line uint64 = 0
	var read_timeout = 30 * time.Second

	// open file
	// basic error handling, if we hit an error, log and return
	// this ends the currently running thread without impacting other threads
	f, err := os.Open(l.FileName)
	if err != nil {
		logp.Err("Error opening file " + err.Error())
		return
	}
	l.FileP = f

	// seek to end
	// for offset, we use the actual file offset
	// we initialize it to the end of the file at time of open
	l.offset, err = l.FileP.Seek(0, 2)
	if err != nil {
		logp.Err("Error seeking in file " + err.Error())
		return
	}
	l.LastOpen = time.Now()

	buffer := new(bytes.Buffer)
	reader := bufio.NewReader(l.FileP)

	for {
		l.CheckReopen()
		text, bytesread, err := readline(reader, buffer, read_timeout)
		if err != nil && err != io.EOF {
			// EOF errors are expected, since we are tailing the file
			logp.Err("Error reading file " + err.Error())
			return
		}

		if bytesread > 0 {
			l.offset += int64(bytesread)
			line++

			event := common.MapStr{}
			event["filename"] = l.FileName
			event["line"] = line
			event["message"] = text
			event["offset"] = l.offset
			event["type"] = l.Type

			event.EnsureTimestampField(now)
			event.EnsureCountField()

			logp.Debug("tailinput", "InputEvent: %v", event)
			output <- event // ship the new event downstream
		}
	}
}

func (l *TcpInput) handleConn(client net.Conn, output chan common.MapStr) {
	reader := bufio.NewReader(client)
	buffer := new(bytes.Buffer)

	var source string = client.RemoteAddr().String()
	var offset int64 = 0
	var line uint64 = 0
	var read_timeout = 10 * time.Second

	logp.Debug("tcpinput", "Handling New Connection from %s", source)

	now := func() time.Time {
		t := time.Now()
		return t
	}

	for {
		text, bytesread, err := l.readline(reader, buffer, read_timeout)

		if err != nil {
			logp.Info("Unexpected state reading from %v; error: %s\n", client.RemoteAddr().String, err)
			return
		}

		logp.Debug("tcpinputlines", "New Line: %s", &text)

		line++

		event := common.MapStr{}
		event["source"] = &source
		event["offset"] = offset
		event["line"] = line
		event["message"] = text
		event["type"] = l.Type

		event.EnsureTimestampField(now)
		event.EnsureCountField()

		offset += int64(bytesread)

		logp.Debug("tcpinput", "InputEvent: %v", event)
		output <- event // ship the new event downstream
		client.Write([]byte("OK"))
	}
	logp.Debug("tcpinput", "Closed Connection from %s", source)
}

// If you had a periodic type input, use the below as the "Run" method instead of the above "Run"
func (l *NullInput) RunPeriodic(output chan common.MapStr) error {
	logp.Debug("[nullinput]", "Starting up Null Input")

	// use the runTick for tick interval, empty functions for minor and major
	go inputs.PeriodicTaskRunner(l, output, l.doStuff, inputs.EmptyFunc, inputs.EmptyFunc)

	return nil
}

func PrintPublishEvent(event common.MapStr) {
	json, err := json.MarshalIndent(event, "", "  ")
	if err != nil {
		logp.Err("json.Marshal: %s", err)
	} else {
		logp.Debug("publish", "Publish: %s", string(json))
	}
}

func (reader *ReaderType) PrintReaderEvent(event common.MapStr) {
	json, err := json.MarshalIndent(event, "", "  ")
	if err != nil {
		logp.Err("json.Marshal: %s", err)
	} else {
		logp.Debug("reader", "Reader: %s", string(json))
	}
}