Merged in feature/SCFF-29 (pull request #7)

Feature/SCFF-29

Author: sklava

eventQueue/eventQueue.go

@@ -33,6 +33,10 @@ func (q *Queue) GetNode() []*Node {
 	return q.nodes
 }
 
+func (n *Queue) GetCount() int {
+	return n.count
+}
+
 func (n *Node) GetNodeEvent() Event {
 	return n.event
 }

eventRouting/eventrouting.go

@@ -75,14 +75,9 @@ 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.Pop() != nil { //if the queue is not empty
-				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.Msg)
-				fmt.Println("sendig event TO queue")
-				e.queue.Push(eventQueue.NewNode(*event))
-			}
+			//Push the event to the queue
+			fmt.Println("pushing event to queue")
+			e.queue.Push(eventQueue.NewNode(*event))
 			e.selectedEventsCount[eventType.String()]++
 
 		}
@@ -150,8 +145,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,10 @@ func main() {
 
 	fmt.Println("this is the sumo endpoint")
 	fmt.Println(sumoEndpoint)
-	loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender(*sumoEndpoint, 1000)
+
+	//Creating queue
+	queue := eventQueue.NewQueue(make([]*eventQueue.Node, 100))
+	loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender(*sumoEndpoint, 1000, *queue)
 
 	fmt.Printf("Starting firehose-to-sumo %s \n", version)
 
@@ -64,9 +67,6 @@ func main() {
 		cachingClient = caching.NewCachingEmpty()
 	}
 
-	//Creating queue
-	queue := eventQueue.NewQueue(make([]*eventQueue.Node, 100))
-
 	//Creating Events
 	events := eventRouting.NewEventRouting(cachingClient, *loggingClientSumo, *queue)
 	err := events.SetupEventRouting(*wantedEvents)

sumoCFFirehose/sumoLogicAppender.go

@@ -7,20 +7,22 @@ import (
 	"net/http"
 	"time"
 
-	. "bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/events"
+	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/eventQueue"
 )
 
 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,
 	}
 }
 
@@ -41,7 +43,10 @@ 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
+	fmt.Println("i'm in appendLogs")
+	event := s.nozzleQueue.Pop().GetNodeEvent()
 	/*
 		if event == nil {
 			return
@@ -55,7 +60,8 @@ func (s *SumoLogicAppender) AppendLogs(event Event) {
 		return
 	}
 
-	Message := /*strconv.Itoa(fields["timestamp"]) + */ "\t" + event.Fields["message_type"].(string) + "\t" + event.Msg
+	Message := time.Unix(0, event.Fields["timestamp"].(int64)*int64(time.Nanosecond)).String() + "\t" + event.Fields["message_type"].(string) + "\t" + event.Msg + "\n"
+	fmt.Println(Message)
 	s.SendToSumo(Message)
 }