-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.go
91 lines (78 loc) · 1.68 KB
/
main.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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
package main
import log "github.com/Sirupsen/logrus"
var (
JobQueue chan Job
pages chan Page
processed chan Job
lineEnqueued chan bool
lineProcessed chan Job
allDone chan bool
source Source
service *Service
)
type Page struct {
count int
page int
last bool
}
type Job struct {
id int
key string
page int
}
func init() {
logger(err_log, "errors.log")
logger(info_log, "info.log")
options = getOptions()
}
func main() {
service = NewService()
if options.Reader == "file" {
source = NewFileSource(options.FilePath)
lineEnqueued = make(chan bool, options.Queue)
lineProcessed = make(chan Job, options.Queue)
} else {
source = NewS3Source()
pages = make(chan Page, 1000)
processed = make(chan Job, options.Queue)
}
JobQueue = make(chan Job, options.Queue)
allDone = make(chan bool, 1)
dispatcher := NewDispatcher(options.Workers)
dispatcher.Run()
source.Run()
}
// watches each page for added and processed objects, determines when it's done and logs finished
func watch() {
lines := 0
done := false
p := make(map[int]int) //page:count
for {
select {
case page := <-pages:
p[page.page] = page.count
done = page.last
case job := <-processed:
p[job.page]--
if p[job.page] == 0 {
info_log.WithFields(log.Fields{"Key": job.key, "Page": job.page, "Done": true}).
Info("Page ", job.page, " finished all jobs")
delete(p, job.page)
}
if done && len(p) == 0 {
allDone <- true
return
}
case <-lineEnqueued:
lines++
case job := <-lineProcessed:
lines--
info_log.WithFields(log.Fields{"Key": job.key}).
Info("Object copied")
if lines == 0 {
allDone <- true
return
}
}
}
}