the appender calls for the next Event in queue and parse it to string

Author: nmillalaf

eventRouting/eventrouting.go

@@ -75,15 +75,8 @@ func (e *EventRouting) RouteEvent(msg *events.Envelope) {
 		if ignored, hasIgnoredField := event.Fields["cf_ignored_app"]; ignored == true && hasIgnoredField {
 			e.selectedEventsCount["ignored_app_message"]++
 		} else {
-			if e.queue.GetCount() == 10 { //if the queue has 10 elements, send to the appender (10 elements to wait to send to the appender?)
-				fmt.Println("sendig event from queue to appender")
-				e.sLAppender.AppendLogs(e.queue.Pop().GetNodeEvent()) // send to appender event from queue
-			} else { //if the queue is empty, send the event to queue
-				/*fmt.Println(event.Fields["timestamp"])
-				fmt.Println("sendig event TO queue")*/
-				e.queue.Push(eventQueue.NewNode(*event))
-			}
-
+			//Push the event to the queue
+			e.queue.Push(eventQueue.NewNode(*event))
 			e.selectedEventsCount[eventType.String()]++
 
 		}
@@ -151,8 +144,10 @@ func (e *EventRouting) LogEventTotals(logTotalsTime time.Duration) {
 			startTime = time.Now()
 			event, lastCount := e.getEventTotals(totalElapsedTime, elapsedTime, count)
 			count = lastCount
-			//*e.log.ShipEvents(event.Fields, event.Msg)
-			e.sLAppender.AppendLogs(*event)
+
+			//Push the event to the queue
+			e.queue.Push(eventQueue.NewNode(*event))
+			//e.sLAppender.AppendLogs(*event)
 		}
 	}()
 }

main.go

@@ -40,7 +40,9 @@ func main() {
 
 	fmt.Println("this is the sumo endpoint")
 	fmt.Println(sumoEndpoint)
-	loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender(*sumoEndpoint, 1000)
+	//Creating queue
+	nozzleQueue := eventQueue.NewQueue(make([]*eventQueue.Node, 100))
+	loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender(*sumoEndpoint, 1000, *nozzleQueue)
 
 	fmt.Printf("Starting firehose-to-sumo %s \n", version)
 
@@ -64,11 +66,8 @@ func main() {
 		cachingClient = caching.NewCachingEmpty()
 	}
 
-	//Creating queue
-	queue := eventQueue.NewQueue(make([]*eventQueue.Node, 100))
-
 	//Creating Events
-	events := eventRouting.NewEventRouting(cachingClient, *loggingClientSumo, *queue)
+	events := eventRouting.NewEventRouting(cachingClient, *loggingClientSumo, *nozzleQueue)
 	err := events.SetupEventRouting(*wantedEvents)
 	if err != nil {
 		log.Fatal("Error setting up event routing: ", err)

sumoCFFirehose/sumoLogicAppender.go

@@ -7,21 +7,23 @@ import (
 	"net/http"
 	"time"
 
-	//"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/eventQueue"
-	. "bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/events"
+	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/eventQueue"
+	//"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/events"
 )
 
 type SumoLogicAppender struct {
 	url               string
 	connectionTimeout int //10000
 	httpClient        http.Client
+	nozzleQueue       eventQueue.Queue
 }
 
-func NewSumoLogicAppender(urlValue string, connectionTimeoutValue int) *SumoLogicAppender {
+func NewSumoLogicAppender(urlValue string, connectionTimeoutValue int, nozzleQueue eventQueue.Queue) *SumoLogicAppender {
 	return &SumoLogicAppender{
 		url:               urlValue,
 		connectionTimeout: connectionTimeoutValue,
 		httpClient:        http.Client{Timeout: time.Duration(connectionTimeoutValue * int(time.Millisecond))},
+		nozzleQueue:       nozzleQueue,
 	}
 }
 
@@ -42,7 +44,9 @@ func (s *SumoLogicAppender) Connect() bool {
 	return success
 }
 
-func (s *SumoLogicAppender) AppendLogs(event Event) {
+func (s *SumoLogicAppender) AppendLogs() {
+	// the appender calls for the next message in the queue and parse it to a string
+	event := s.nozzleQueue.Pop().GetNodeEvent()
 	/*
 		if event == nil {
 			return
@@ -56,7 +60,8 @@ func (s *SumoLogicAppender) AppendLogs(event Event) {
 		return
 	}
 
-	Message := /*strconv.Itoa(fields["timestamp"]) + */ "\t" + event.Fields["message_type"].(string) + "\t" + event.Msg + "\n"
+	Message := time.Unix(int64(event.Fields["timestamp"].(int64)), 0).String() + "\t" + event.Fields["message_type"].(string) + "\t" + event.Msg + "\n"
+	fmt.Println(Message)
 	s.SendToSumo(Message)
 }