-
Notifications
You must be signed in to change notification settings - Fork 14
/
tailBatcher.go
41 lines (34 loc) · 880 Bytes
/
tailBatcher.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
package batchers
import (
"rare/pkg/followreader"
"rare/pkg/logger"
"sync"
)
// TailFilesToChan tails a set of files to an input batcher that can be consumed by extractor
// unlike a normal file batcher, this will attempt to tail all files at once
func TailFilesToChan(filenames <-chan string, batchSize int, reopen, poll bool) *Batcher {
out := newBatcher(128)
go func() {
var wg sync.WaitGroup
for filename := range filenames {
wg.Add(1)
go func(filename string) {
defer func() {
wg.Done()
out.stopFileReading(filename)
}()
out.startFileReading(filename)
r, err := followreader.New(filename, reopen, poll)
if err != nil {
logger.Print("Unable to open file: ", err)
out.incErrors()
return
}
out.syncReaderToBatcher(filename, r, batchSize)
}(filename)
}
wg.Wait()
out.close()
}()
return out
}