logCollationWorker will take log lines over the given channel, and buffer them until either the buffer is full, or the flushTimeout is exceeded. This is to reduce the number of writes to the log files, in order to batch them up as larger collated chunks, whilst maintaining a low-level of latency wit
(loggerClosed chan struct{}, collateBuffer chan string, flushChan chan struct{}, collateBufferWg *sync.WaitGroup, logger *log.Logger, maxBufferSize int, collateFlushTimeout time.Duration)
| 62 | // logCollationWorker will take log lines over the given channel, and buffer them until either the buffer is full, or the flushTimeout is exceeded. |
| 63 | // This is to reduce the number of writes to the log files, in order to batch them up as larger collated chunks, whilst maintaining a low-level of latency with the flush timeout. |
| 64 | func logCollationWorker(loggerClosed chan struct{}, collateBuffer chan string, flushChan chan struct{}, collateBufferWg *sync.WaitGroup, logger *log.Logger, maxBufferSize int, collateFlushTimeout time.Duration) { |
| 65 | // The initial duration of the timeout timer doesn't matter, |
| 66 | // because we reset it whenever we buffer a log without flushing it. |
| 67 | t := time.NewTimer(math.MaxInt64) |
| 68 | logBuffer := make([]string, 0, maxBufferSize) |
| 69 | |
| 70 | for { |
| 71 | select { |
| 72 | case l := <-collateBuffer: |
| 73 | logBuffer = append(logBuffer, l) |
| 74 | collateBufferWg.Done() |
| 75 | if len(logBuffer) >= maxBufferSize { |
| 76 | // Flush if the buffer is full after this log |
| 77 | logger.Print(strings.Join(logBuffer, "\n")) |
| 78 | logBuffer = logBuffer[:0] |
| 79 | } else { |
| 80 | // Start the timeout timer to flush this partial buffer. |
| 81 | // Note: We don't need to care about stopping the timer as per Go docs, |
| 82 | // because we're not bothered about a double-firing of the timer, |
| 83 | // since we check if there's anything to flush first. |
| 84 | _ = t.Reset(collateFlushTimeout) |
| 85 | } |
| 86 | case <-flushChan: |
| 87 | if len(logBuffer) > 0 { |
| 88 | // We've sent an explicit "flush now" signal, and want to use a wait group to signal when we've actually performed the flush. |
| 89 | logger.Print(strings.Join(logBuffer, "\n")) |
| 90 | logBuffer = logBuffer[:0] |
| 91 | } |
| 92 | flushLogBuffersWaitGroup.Done() |
| 93 | case <-t.C: |
| 94 | if len(logBuffer) > 0 { |
| 95 | // We've timed out waiting for more logs to be put into the buffer, so flush it now. |
| 96 | logger.Print(strings.Join(logBuffer, "\n")) |
| 97 | logBuffer = logBuffer[:0] |
| 98 | } |
| 99 | case <-loggerClosed: |
| 100 | return |
| 101 | } |
| 102 | } |
| 103 | } |
no test coverage detected