SCFF-31 start method for the appender

Author: nmillalaf

sumoCFFirehose/sumoLogicAppender.go

@@ -11,20 +11,24 @@ import (
 )
 
 type SumoLogicAppender struct {
-	url               string
-	connectionTimeout int //10000
-	httpClient        http.Client
-	nozzleQueue       eventQueue.Queue
-	eventsBatchSize   int
+	url                      string
+	connectionTimeout        int //10000
+	httpClient               http.Client
+	nozzleQueue              eventQueue.Queue
+	eventsBatchSize          int
+	logEventsInCurrentBuffer int
+	logStringToSend          string
 }
 
 func NewSumoLogicAppender(urlValue string, connectionTimeoutValue int, nozzleQueue eventQueue.Queue, eventsBatchSize int) *SumoLogicAppender {
 	return &SumoLogicAppender{
-		url:               urlValue,
-		connectionTimeout: connectionTimeoutValue,
-		httpClient:        http.Client{Timeout: time.Duration(connectionTimeoutValue * int(time.Millisecond))},
-		nozzleQueue:       nozzleQueue,
-		eventsBatchSize:   eventsBatchSize,
+		url:                      urlValue,
+		connectionTimeout:        connectionTimeoutValue,
+		httpClient:               http.Client{Timeout: time.Duration(connectionTimeoutValue * int(time.Millisecond))},
+		nozzleQueue:              nozzleQueue,
+		eventsBatchSize:          eventsBatchSize,
+		logEventsInCurrentBuffer: 0,
+		logStringToSend:          "",
 	}
 }
 
@@ -46,7 +50,25 @@ func (s *SumoLogicAppender) Connect() bool {
 }
 
 func (s *SumoLogicAppender) Start() {
-	s.AppendLogs()
+	timer := time.NewTimer(60 * time.Second)
+	fmt.Println("Starting Appender Worker")
+	s.logStringToSend = ""
+	for {
+		time.Sleep(300 * time.Millisecond)
+		if s.nozzleQueue.GetCount() > 0 {
+			s.AppendLogs()
+		}
+		if s.eventsBatchSize >= s.logEventsInCurrentBuffer {
+			s.SendToSumo(s.logStringToSend)
+			s.logEventsInCurrentBuffer = 0 // reset counter
+			s.logStringToSend = ""         //reset String
+		} else if (<-timer.C).Second() == 0 {
+			s.SendToSumo(s.logStringToSend)
+			s.logEventsInCurrentBuffer = 0 // reset counter
+			s.logStringToSend = ""         //reset String
+		}
+
+	}
 }
 
 func StringBuilder(node *eventQueue.Node) string {
@@ -66,17 +88,8 @@ func StringBuilder(node *eventQueue.Node) string {
 func (s *SumoLogicAppender) AppendLogs() {
 	// the appender calls for the next message in the queue and parse it to a string
 	//timer := time.NewTimer(60 * time.Second)
-	logMessage := ""
-	stringBuilderCalls := 0
-	for s.nozzleQueue.GetCount() > s.eventsBatchSize { //when the batch limit is met, call stringBuilder
-		logMessage = logMessage + StringBuilder(s.nozzleQueue.Pop())
-		stringBuilderCalls++
-		if s.eventsBatchSize == stringBuilderCalls {
-			s.SendToSumo(logMessage)
-			stringBuilderCalls = 0
-			logMessage = ""
-		}
-	}
+	s.logStringToSend = s.logStringToSend + StringBuilder(s.nozzleQueue.Pop())
+	s.logEventsInCurrentBuffer++
 
 }