-
Notifications
You must be signed in to change notification settings - Fork 1
/
worker.go
83 lines (69 loc) · 1.99 KB
/
worker.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
package sqsjkr
import (
"sync/atomic"
"github.com/kayac/sqsjkr/throttle"
)
// Worker struct
type Worker struct {
sjkr SQSJkr
id int
jobs <-chan Job
stats *Stats
}
// SpawnWorker spawn worker
func SpawnWorker(sjkr SQSJkr, wid int, js <-chan Job, s *Stats) {
defer func(id int) {
logger.Infof("[worker_id:%d] terminated command worker.", wid)
}(wid)
worker := Worker{
sjkr: sjkr,
id: wid,
jobs: js,
stats: s,
}
logger.Infof("[worker_id:%d] spawn worker.", wid)
worker.ReceiveMessage()
}
// ReceiveMessage receive messages
func (w Worker) ReceiveMessage() {
// worker will be killed when errCnt is over 5.
for job := range w.jobs {
if err := w.sjkr.Throttler().Set(job.JobID()); err != nil {
if err == throttle.ErrDuplicatedMessage {
logger.Errorf("duplicated message id: %s", job.JobID())
continue
}
logger.Errorf("reason=%s ,job=%v", err.Error(), job)
}
if err := w.executeJob(job); err != nil {
logger.Errorf("[worker_id:%d] execute job failed %s", w.id, err.Error())
}
}
logger.Infof("[worker_id:%d] terminating", w.id)
return
}
func (w Worker) executeJob(job Job) error {
// busy worker number count up
w.stats.busy <- struct{}{}
// decrement busy worker number when to return
defer func() {
<-w.stats.busy
}()
// Execute job
logger.Infof("CMD event_id:%s command:%s", job.EventID(), job.Command())
output, err := job.Execute(w.sjkr.Locker())
if err != nil && output == nil {
atomic.AddInt64(&w.stats.Invocations.Failed, 1)
logger.Errorf("[event:%s] failed to invoke command, reason: %s, job: %s", job.EventID(), err.Error(), job.String())
return err
} else if err != nil {
atomic.AddInt64(&w.stats.Invocations.Errored, 1)
logger.Errorf("[event:%s] errored to invoke command, reason: %s, job: %s", job.EventID(), err.Error(), job.String())
logger.Errorf(string(output))
return err
} else {
atomic.AddInt64(&w.stats.Invocations.Succeeded, 1)
}
logger.Debugf("[event:%s] output:\n%s", job.EventID(), string(output))
return nil
}