We read every piece of feedback, and take your input very seriously.
To see all available qualifiers, see our documentation.
There was an error while loading. Please reload this page.
2 parents 9bcba1e + 3bfec51 commit 9b86919Copy full SHA for 9b86919
buffer.go
@@ -67,13 +67,12 @@ func (buffer *Buffer) Close() error {
67
}
68
69
func (buffer *Buffer) consume() {
70
+ count := 0
71
items := make([]interface{}, buffer.options.Size)
72
+ mustFlush := false
73
ticker, stopTicker := newTicker(buffer.options.FlushInterval)
74
- count := 0
75
isOpen := true
- mustFlush := false
76
-
77
for isOpen {
78
select {
79
case item := <-buffer.dataCh:
@@ -92,7 +91,9 @@ func (buffer *Buffer) consume() {
92
91
if mustFlush {
93
stopTicker()
94
buffer.options.Flusher.Write(items[:count])
+
95
count = 0
96
+ items = make([]interface{}, buffer.options.Size)
97
mustFlush = false
98
ticker, stopTicker = newTicker(buffer.options.FlushInterval)
99
0 commit comments