SCFF-30 remove the queue from Appender

Author: nmillalaf

eventQueue/eventQueue_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestQueueFIFO(t *testing.T) {
-
+	assert := assert.New(t)
 	node1 := Node{
 		Event: Event{
 			Fields: map[string]interface{}{
@@ -45,7 +45,7 @@ func TestQueueFIFO(t *testing.T) {
 	queue.Push(&node2)
 	queue.Push(&node3)
 
-	assert.Equal(t, queue.Pop().Event.Msg, "index [01]", "")
-	assert.Equal(t, queue.Pop().Event.Msg, "index [02]", "")
-	assert.Equal(t, queue.Pop().Event.Msg, "index [03]", "")
+	assert.Equal(queue.Pop().Event.Msg, "index [01]", "")
+	assert.Equal(queue.Pop().Event.Msg, "index [02]", "")
+	assert.Equal(queue.Pop().Event.Msg, "index [03]", "")
 }

sumoCFFirehose/sumoLogicAppender.go

@@ -45,29 +45,28 @@ func (s *SumoLogicAppender) Connect() bool {
 	return success
 }
 
-func StringBuilder(queue eventQueue.Queue) string {
+func StringBuilder(node *eventQueue.Node) string {
 	buf := new(bytes.Buffer)
-	for queue.GetCount() > 0 { //Pop eventsBatch from queue
-		event := queue.Pop().GetNodeEvent()
-		if event.Fields["message_type"] == nil {
-			return ""
-		}
-		if event.Fields["message_type"] == "" {
-			return ""
-		}
-		message := time.Unix(0, event.Fields["timestamp"].(int64)*int64(time.Nanosecond)).String() + "\t" + event.Fields["message_type"].(string) + "\t" + event.Msg + "\n"
-		buf.WriteString(message)
+	if node.Event.Fields["message_type"] == nil {
+		return ""
 	}
+	if node.Event.Fields["message_type"] == "" {
+		return ""
+	}
+	message := time.Unix(0, node.Event.Fields["timestamp"].(int64)*int64(time.Nanosecond)).String() + "\t" + node.Event.Fields["message_type"].(string) + "\t" + node.Event.Msg + "\n"
+	buf.WriteString(message)
 
 	return buf.String()
 }
 
-func (s *SumoLogicAppender) AppendLogs(queue eventQueue.Queue) {
+func (s *SumoLogicAppender) AppendLogs() {
 	// the appender calls for the next message in the queue and parse it to a string
-	if queue.GetCount() > s.eventsBatchSize { //whent the batch limit is met, call stringBuilder
-		logMessage := StringBuilder(queue)
-		s.SendToSumo(logMessage)
+	logMessage := ""
+	if s.nozzleQueue.GetCount() > s.eventsBatchSize { //when the batch limit is met, call stringBuilder
+		logMessage = logMessage + StringBuilder(s.nozzleQueue.Pop())
+
 	}
+	s.SendToSumo(logMessage)
 
 }
 

sumoCFFirehose/sumoLogicAppender_test.go

@@ -48,7 +48,10 @@ func testAppenderStringBuilder(t *testing.T) {
 	queue.Push(&node2)
 	queue.Push(&node3)
 
-	finalString := StringBuilder(queue)
+	finalString := ""
+	for queue.GetCount() > 0 {
+		finalString = finalString + StringBuilder(queue.Pop())
+	}
 	assert.Equal(t, finalString, "2016-12-12 16:02:41.828366387 -0300 CLST"+"\t"+"OUT"+"\t"+"index [01]"+"\n"+
 		"2016-12-12 16:02:42.844737993 -0300 CLST"+"\t"+"OUT"+"\t"+"index [02]"+"\n"+
 		"2016-12-12 16:02:43.862436654 -0300 CLST"+"\t"+"OUT"+"\t"+"index [03]"+"\n", "")