Skip to content

Commit edbce2c

Browse files
committed
Merged in feature/SCFF-36 (pull request #11)
SCFF-36 added loop for POP while queue is not empty and batchsize is not reached
2 parents 3f77cc0 + 87bc0c0 commit edbce2c

1 file changed

Lines changed: 13 additions & 13 deletions

File tree

sumoCFFirehose/sumoLogicAppender.go

Lines changed: 13 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -37,19 +37,20 @@ func (s *SumoLogicAppender) Start() {
3737
logging.Info.Println("Starting Appender Worker")
3838
for {
3939
time.Sleep(300 * time.Millisecond)
40-
if s.nozzleQueue.GetCount() != 0 { //if queue is not empty, AppendLogs
41-
s.AppendLogs()
40+
// while queue is not empty && s.eventsBatchSize not completed, queue.POP (appendLogs)
41+
for s.nozzleQueue.GetCount() != 0 && s.logEventsInCurrentBuffer <= s.eventsBatchSize {
42+
s.AppendLogs() //this method POP an event from queue
43+
timer = time.Now() //reset timer
44+
if s.logEventsInCurrentBuffer == s.eventsBatchSize { //if buffer is full, send logs to sumo
45+
logging.Trace.Println("Batch Size complete")
46+
break
47+
} else if time.Since(timer).Seconds() >= 10 { // else if timer is up, send existing logs to sumo
48+
logging.Trace.Println("Sending current batch of logs after timer exceeded limit")
49+
break
50+
}
4251
}
43-
if s.logEventsInCurrentBuffer >= s.eventsBatchSize { //if buffer is full, send logs to sumo
44-
logging.Trace.Println("Batch Size complete")
45-
s.SendToSumo(s.logStringToSend)
46-
47-
} else if time.Since(timer).Seconds() >= 10 { // else if timer is up, send existing logs to sumo
48-
logging.Trace.Println("Sending current batch of logs after timer exceeded limit")
49-
s.SendToSumo(s.logStringToSend)
50-
timer = time.Now() //reset timer
51-
}
52-
52+
s.SendToSumo(s.logStringToSend)
53+
timer = time.Now() //reset timer
5354
}
5455
}
5556

@@ -69,7 +70,6 @@ func StringBuilder(event *events.Event) string {
6970

7071
func (s *SumoLogicAppender) AppendLogs() {
7172
// the appender calls for the next message in the queue and parse it to a string
72-
//timer := time.NewTimer(60 * time.Second)
7373
s.logStringToSend.Write([]byte(StringBuilder(s.nozzleQueue.Pop())))
7474
s.logEventsInCurrentBuffer++
7575
}

0 commit comments

Comments
 (0)