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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
package jobs
import (
"github.com/spiral/errors"
"github.com/spiral/roadrunner/v2/plugins/jobs/job"
"github.com/spiral/roadrunner/v2/plugins/logger"
jobsv1beta "github.com/spiral/roadrunner/v2/proto/jobs/v1beta"
)
type rpc struct {
log logger.Logger
p *Plugin
}
/*
List of the RPC methods:
1. Push - single job push
2. PushBatch - push job batch
3. Reset - managed by the Resetter plugin
4. Pause - pauses set of pipelines
5. Resume - resumes set of pipelines
6. Workers - managed by the Informer plugin.
7. Stat - jobs statistic
*/
func (r *rpc) Push(j *jobsv1beta.PushRequest, _ *jobsv1beta.Empty) error {
const op = errors.Op("jobs_rpc_push")
// convert transport entity into domain
// how we can do this quickly
if j.GetJob().GetId() == "" {
return errors.E(op, errors.Str("empty ID field not allowed"))
}
err := r.p.Push(r.from(j.GetJob()))
if err != nil {
return errors.E(op, err)
}
return nil
}
func (r *rpc) PushBatch(j *jobsv1beta.PushBatchRequest, _ *jobsv1beta.Empty) error {
const op = errors.Op("jobs_rpc_push")
l := len(j.GetJobs())
batch := make([]*job.Job, l)
for i := 0; i < l; i++ {
// convert transport entity into domain
// how we can do this quickly
batch[i] = r.from(j.GetJobs()[i])
}
err := r.p.PushBatch(batch)
if err != nil {
return errors.E(op, err)
}
return nil
}
func (r *rpc) Pause(req *jobsv1beta.MaintenanceRequest, _ *jobsv1beta.Empty) error {
pipelines := make([]string, len(req.GetPipelines()))
for i := 0; i < len(pipelines); i++ {
pipelines[i] = req.GetPipelines()[i]
}
r.p.Pause(pipelines)
return nil
}
func (r *rpc) Resume(req *jobsv1beta.MaintenanceRequest, _ *jobsv1beta.Empty) error {
pipelines := make([]string, len(req.GetPipelines()))
for i := 0; i < len(pipelines); i++ {
pipelines[i] = req.GetPipelines()[i]
}
r.p.Resume(pipelines)
return nil
}
func (r *rpc) List(_ *jobsv1beta.Empty, resp *jobsv1beta.List) error {
resp.Pipelines = r.p.List()
return nil
}
// from converts from transport entity to domain
func (r *rpc) from(j *jobsv1beta.Job) *job.Job {
headers := map[string][]string{}
for k, v := range j.GetHeaders() {
headers[k] = v.GetValue()
}
jb := &job.Job{
Job: j.GetJob(),
Headers: headers,
Ident: j.GetId(),
Payload: j.GetPayload(),
Options: &job.Options{
Priority: j.GetOptions().GetPriority(),
Pipeline: j.GetOptions().GetPipeline(),
Delay: j.GetOptions().GetDelay(),
Attempts: j.GetOptions().GetAttempts(),
RetryDelay: j.GetOptions().GetRetryDelay(),
Timeout: j.GetOptions().GetTimeout(),
},
}
return jb
}
|
