func routeEvent(msg *events.Envelope) {

	eventType := msg.GetEventType()

	if selectedEvents[eventType.String()] {
		var event Event
		switch eventType {
		case events.Envelope_Heartbeat:
			event = Heartbeat(msg)
		case events.Envelope_HttpStart:
			event = HttpStart(msg)
		case events.Envelope_HttpStop:
			event = HttpStop(msg)
		case events.Envelope_HttpStartStop:
			event = HttpStartStop(msg)
		case events.Envelope_LogMessage:
			event = LogMessage(msg)
		case events.Envelope_ValueMetric:
			event = ValueMetric(msg)
		case events.Envelope_CounterEvent:
			event = CounterEvent(msg)
		case events.Envelope_Error:
			event = ErrorEvent(msg)
		case events.Envelope_ContainerMetric:
			event = ContainerMetric(msg)
		}

		event.AnnotateWithAppData()
		event.AnnotateWithMetaData()
		event.ShipEvent()
	}
}

func (p *CounterProcessor) Process(e *events.Envelope) []metrics.WMetric {
	processedMetrics := make([]metrics.WMetric, 1)
	counterEvent := e.GetCounterEvent()

	processedMetrics[0] = *p.ProcessCounter(counterEvent)

	return processedMetrics
}

func (p *ValueMetricProcessor) Process(e *events.Envelope) []metrics.WMetric {
	processedMetrics := make([]metrics.WMetric, 1)
	valueMetricEvent := e.GetValueMetric()

	processedMetrics[0] = *p.ProcessValueMetric(valueMetricEvent, e.GetOrigin())

	return processedMetrics
}

func (p *ContainerMetricProcessor) Process(e *events.Envelope) []metrics.WMetric {
	processedMetrics := make([]metrics.WMetric, 3)
	containerMetricEvent := e.GetContainerMetric()

	processedMetrics[0] = *p.ProcessContainerMetricCPU(containerMetricEvent)
	processedMetrics[1] = *p.ProcessContainerMetricMemory(containerMetricEvent)
	processedMetrics[2] = *p.ProcessContainerMetricDisk(containerMetricEvent)

	return processedMetrics
}

func (p *HttpStartStopProcessor) Process(e *events.Envelope) []metrics.Metric {
	processedMetrics := make([]metrics.Metric, 4)
	httpStartStopEvent := e.GetHttpStartStop()

	processedMetrics[0] = metrics.Metric(p.ProcessHttpStartStopResponseTime(httpStartStopEvent))
	processedMetrics[1] = metrics.Metric(p.ProcessHttpStartStopStatusCodeCount(httpStartStopEvent))
	processedMetrics[2] = metrics.Metric(p.ProcessHttpStartStopHttpErrorCount(httpStartStopEvent))
	processedMetrics[3] = metrics.Metric(p.ProcessHttpStartStopHttpRequestCount(httpStartStopEvent))

	return processedMetrics
}

func (p *HeartbeatProcessor) Process(e *events.Envelope) []metrics.WMetric {
	processedMetrics := make([]metrics.WMetric, 4)
	heartbeat := e.GetHeartbeat()
	origin := e.GetOrigin()

	processedMetrics[0] = *p.ProcessHeartbeatCount(heartbeat, origin)
	processedMetrics[1] = *p.ProcessHeartbeatEventsSentCount(heartbeat, origin)
	processedMetrics[2] = *p.ProcessHeartbeatEventsReceivedCount(heartbeat, origin)
	processedMetrics[3] = *p.ProcessHeartbeatEventsErrorCount(heartbeat, origin)

	return processedMetrics
}

func (c *Client) AddMetric(envelope *events.Envelope) {
	c.totalMessagesReceived++
	eventType := envelope.GetEventType()

	processedMetrics := []metrics.WMetric{}

	// epagent-nozzle can handle CounterEvent, ContainerMetric, Heartbeat,
	// HttpStartStop and ValueMetric events
	switch eventType {
	case events.Envelope_ContainerMetric:
		processedMetrics = c.containerMetricProcessor.Process(envelope)
	case events.Envelope_CounterEvent:
		processedMetrics = c.counterProcessor.Process(envelope)
	case events.Envelope_Heartbeat:
		processedMetrics = c.heartbeatProcessor.Process(envelope)
	case events.Envelope_HttpStartStop:
		processedMetrics = c.httpStartStopProcessor.Process(envelope)
	case events.Envelope_ValueMetric:
		processedMetrics = c.valueMetricProcessor.Process(envelope)
	default:
		// do nothing
	}

	if len(processedMetrics) > 0 {
		for _, metric := range processedMetrics {
			if metric.Type == "PerintervalCounter" {
				_, ok := c.metricPoints[metric.Name]
				if ok {
					//log.Println("incrementing counter")
					oldmetric := c.metricPoints[metric.Name]
					oldint, _ := strconv.ParseInt(oldmetric.Value, 10, 64)
					newint, _ := strconv.ParseInt(metric.Value, 10, 64)
					newint = oldint + newint

					metric.Value = strconv.FormatInt(newint, 10)
				}
			}
			c.metricPoints[metric.Name] = metric
			//log.Printf("metric: %s:::%s\n", metric.Name, metric.Value)
		}
	}

}

func HttpStartStop(msg *events.Envelope) Event {
	httpStartStop := msg.GetHttpStartStop()

	fields := logrus.Fields{
		"origin":            msg.GetOrigin(),
		"cf_app_id":         httpStartStop.GetApplicationId(),
		"content_length":    httpStartStop.GetContentLength(),
		"instance_id":       httpStartStop.GetInstanceId(),
		"instance_index":    httpStartStop.GetInstanceIndex(),
		"method":            httpStartStop.GetMethod(),
		"parent_request_id": httpStartStop.GetParentRequestId(),
		"peer_type":         httpStartStop.GetPeerType(),
		"remote_addr":       httpStartStop.GetRemoteAddress(),
		"request_id":        httpStartStop.GetRequestId(),
		"start_timestamp":   httpStartStop.GetStartTimestamp(),
		"status_code":       httpStartStop.GetStatusCode(),
		"stop_timestamp":    httpStartStop.GetStopTimestamp(),
		"uri":               httpStartStop.GetUri(),
		"user_agent":        httpStartStop.GetUserAgent(),
		"duration_ms":       (((httpStartStop.GetStopTimestamp() - httpStartStop.GetStartTimestamp()) / 1000) / 1000),
	}

	return Event{
		Fields: fields,
		Msg:    "",
		Type:   msg.GetEventType().String(),
	}
}

// RecentLogs connects to traffic controller via its 'recentlogs' http(s) endpoint and returns a slice of recent messages.
// It does not guarantee any order of the messages; they are in the order returned by traffic controller.
//
// The SortRecent method is provided to sort the data returned by this method.
func (cnsmr *Consumer) RecentLogs(appGuid string, authToken string) ([]*events.LogMessage, error) {
	resp, err := cnsmr.makeHttpRequestToTrafficController(appGuid, authToken, "recentlogs")

	if err != nil {
		return nil, errors.New(fmt.Sprintf("Error dialing traffic controller server: %s.\nPlease ask your Cloud Foundry Operator to check the platform configuration (traffic controller endpoint is %s).", err.Error(), cnsmr.trafficControllerUrl))
	}

	defer resp.Body.Close()

	err = checkForErrors(resp)
	if err != nil {
		return nil, err
	}

	reader, err := checkContentsAndFindBoundaries(resp)
	if err != nil {
		return nil, err
	}

	var buffer bytes.Buffer
	messages := make([]*events.LogMessage, 0, 200)

	for part, loopErr := reader.NextPart(); loopErr == nil; part, loopErr = reader.NextPart() {
		buffer.Reset()

		msg := new(events.Envelope)
		_, err := buffer.ReadFrom(part)
		if err != nil {
			break
		}
		proto.Unmarshal(buffer.Bytes(), msg)
		messages = append(messages, msg.GetLogMessage())
	}

	return messages, err
}

func ErrorEvent(msg *events.Envelope) Event {
	errorEvent := msg.GetError()

	fields := logrus.Fields{
		"origin": msg.GetOrigin(),
		"code":   errorEvent.GetCode(),
		"delta":  errorEvent.GetSource(),
	}

	return Event{
		Fields: fields,
		Msg:    errorEvent.GetMessage(),
		Type:   msg.GetEventType().String(),
	}
}

func CounterEvent(msg *events.Envelope) Event {
	counterEvent := msg.GetCounterEvent()

	fields := logrus.Fields{
		"origin": msg.GetOrigin(),
		"name":   counterEvent.GetName(),
		"delta":  counterEvent.GetDelta(),
		"total":  counterEvent.GetTotal(),
	}

	return Event{
		Fields: fields,
		Msg:    "",
		Type:   msg.GetEventType().String(),
	}
}

func ValueMetric(msg *events.Envelope) Event {
	valMetric := msg.GetValueMetric()

	fields := logrus.Fields{
		"origin": msg.GetOrigin(),
		"name":   valMetric.GetName(),
		"unit":   valMetric.GetUnit(),
		"value":  valMetric.GetValue(),
	}

	return Event{
		Fields: fields,
		Msg:    "",
		Type:   msg.GetEventType().String(),
	}
}

func ContainerMetric(msg *events.Envelope) Event {
	containerMetric := msg.GetContainerMetric()

	fields := logrus.Fields{
		"origin":         msg.GetOrigin(),
		"cf_app_id":      containerMetric.GetApplicationId(),
		"cpu_percentage": containerMetric.GetCpuPercentage(),
		"disk_bytes":     containerMetric.GetDiskBytes(),
		"instance_index": containerMetric.GetInstanceIndex(),
		"memory_bytes":   containerMetric.GetMemoryBytes(),
	}

	return Event{
		Fields: fields,
		Msg:    "",
		Type:   msg.GetEventType().String(),
	}
}

func LogMessage(msg *events.Envelope) Event {
	logMessage := msg.GetLogMessage()

	fields := logrus.Fields{
		"origin":          msg.GetOrigin(),
		"cf_app_id":       logMessage.GetAppId(),
		"timestamp":       logMessage.GetTimestamp(),
		"source_type":     logMessage.GetSourceType(),
		"message_type":    logMessage.GetMessageType().String(),
		"source_instance": logMessage.GetSourceInstance(),
	}

	return Event{
		Fields: fields,
		Msg:    string(logMessage.GetMessage()),
		Type:   msg.GetEventType().String(),
	}
}

func HttpStop(msg *events.Envelope) Event {
	httpStop := msg.GetHttpStop()

	fields := logrus.Fields{
		"origin":         msg.GetOrigin(),
		"cf_app_id":      httpStop.GetApplicationId(),
		"content_length": httpStop.GetContentLength(),
		"peer_type":      httpStop.GetPeerType(),
		"request_id":     httpStop.GetRequestId(),
		"status_code":    httpStop.GetStatusCode(),
		"timestamp":      httpStop.GetTimestamp(),
		"uri":            httpStop.GetUri(),
	}

	return Event{
		Fields: fields,
		Msg:    "",
		Type:   msg.GetEventType().String(),
	}
}

func HttpStart(msg *events.Envelope) Event {
	httpStart := msg.GetHttpStart()

	fields := logrus.Fields{
		"origin":            msg.GetOrigin(),
		"cf_app_id":         httpStart.GetApplicationId(),
		"instance_id":       httpStart.GetInstanceId(),
		"instance_index":    httpStart.GetInstanceIndex(),
		"method":            httpStart.GetMethod(),
		"parent_request_id": httpStart.GetParentRequestId(),
		"peer_type":         httpStart.GetPeerType(),
		"request_id":        httpStart.GetRequestId(),
		"remote_addr":       httpStart.GetRemoteAddress(),
		"timestamp":         httpStart.GetTimestamp(),
		"uri":               httpStart.GetUri(),
		"user_agent":        httpStart.GetUserAgent(),
	}

	return Event{
		Fields: fields,
		Msg:    "",
		Type:   msg.GetEventType().String(),
	}
}

func Heartbeat(msg *events.Envelope) Event {
	heartbeat := msg.GetHeartbeat()

	var avail uint64

	if heartbeat.GetSentCount() > 0 {
		avail = heartbeat.GetReceivedCount() / heartbeat.GetSentCount()
	}

	fields := logrus.Fields{
		"ctl_msg_id":     heartbeat.GetControlMessageIdentifier(),
		"error_count":    heartbeat.GetErrorCount(),
		"origin":         msg.GetOrigin(),
		"received_count": heartbeat.GetReceivedCount(),
		"sent_count":     heartbeat.GetSentCount(),
		"availability":   avail,
	}

	return Event{
		Fields: fields,
		Msg:    "",
		Type:   msg.GetEventType().String(),
	}
}

func (d *APMFirehoseNozzle) handleMessage(envelope *events.Envelope) {
	if envelope.GetEventType() == events.Envelope_CounterEvent && envelope.CounterEvent.GetName() == "TruncatingBuffer.DroppedMessages" && envelope.GetOrigin() == "doppler" {
		log.Printf("We've intercepted an upstream message which indicates that the nozzle or the TrafficController is not keeping up. Please try scaling up the nozzle.")
		d.client.AlertSlowConsumerError()
	}
}