-
Notifications
You must be signed in to change notification settings - Fork 1
/
item.go
161 lines (137 loc) · 3.88 KB
/
item.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
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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
package memoryjobs
import (
"context"
"sync"
"sync/atomic"
"time"
"github.com/goccy/go-json"
"github.com/roadrunner-server/api/v4/plugins/v3/jobs"
"github.com/roadrunner-server/errors"
)
type Item struct {
// Job contains name of job broker (usually PHP class).
Job string `json:"job"`
// Ident is unique identifier of the job, should be provided from outside
Ident string `json:"id"`
// Payload is string data (usually JSON) passed to Job broker.
Payload []byte `json:"payload"`
// Headers with key-values pairs
headers map[string][]string
// Options contains set of PipelineOptions specific to job execution. Can be empty.
Options *Options `json:"options,omitempty"`
}
// Options carry information about how to handle given job.
type Options struct {
// Priority is job priority, default - 10
// pointer to distinguish 0 as a priority and nil as priority not set
Priority int64 `json:"priority"`
// Pipeline manually specified pipeline.
Pipeline string `json:"pipeline,omitempty"`
// Delay defines time duration to delay execution for. Defaults to none.
Delay int64 `json:"delay,omitempty"`
// private
stopped *uint64
requeueFn func(context.Context, *Item) error
cond *sync.Cond
msgInFlight *int64
delayed *int64
}
// DelayDuration returns delay duration in a form of time.Duration.
func (o *Options) DelayDuration() time.Duration {
return time.Second * time.Duration(o.Delay)
}
func (i *Item) ID() string {
return i.Ident
}
func (i *Item) GroupID() string {
return i.Options.Pipeline
}
func (i *Item) Priority() int64 {
return i.Options.Priority
}
// Body packs job payload into binary payload.
func (i *Item) Body() []byte {
return i.Payload
}
// Context packs job context (job, id) into binary payload.
func (i *Item) Context() ([]byte, error) {
ctx, err := json.Marshal(
struct {
ID string `json:"id"`
Job string `json:"job"`
Driver string `json:"driver"`
Headers map[string][]string `json:"headers"`
Pipeline string `json:"pipeline"`
}{
ID: i.Ident,
Job: i.Job,
Driver: pluginName,
Headers: i.headers,
Pipeline: i.Options.Pipeline,
},
)
if err != nil {
return nil, err
}
return ctx, nil
}
func (i *Item) Headers() map[string][]string {
return i.headers
}
func (i *Item) Ack() error {
// reduce number of the all active jobs
i.atomicallyReduceCount()
if atomic.LoadUint64(i.Options.stopped) == 1 {
return errors.Str("failed to acknowledge the JOB, the pipeline is probably stopped")
}
return nil
}
func (i *Item) Nack() error {
i.atomicallyReduceCount()
if atomic.LoadUint64(i.Options.stopped) == 1 {
return errors.Str("failed to acknowledge the JOB, the pipeline is probably stopped")
}
return nil
}
func (i *Item) Requeue(headers map[string][]string, delay int64) error {
i.atomicallyReduceCount()
if atomic.LoadUint64(i.Options.stopped) == 1 {
return errors.Str("failed to acknowledge the JOB, the pipeline is probably stopped")
}
// overwrite the delay
i.Options.Delay = delay
i.headers = headers
err := i.Options.requeueFn(context.Background(), i)
if err != nil {
return err
}
return nil
}
// Respond for the in-memory is no-op
func (i *Item) Respond([]byte, string) error {
return nil
}
// atomicallyReduceCount reduces counter of active or delayed jobs
func (i *Item) atomicallyReduceCount() {
// reduce number of the all active jobs
atomic.AddInt64(i.Options.msgInFlight, ^int64(0))
// pass 1 job
i.Options.cond.Signal()
// if job was delayed, reduce number of the delayed jobs
if i.Options.Delay > 0 {
atomic.AddInt64(i.Options.delayed, ^int64(0))
}
}
func fromJob(job jobs.Message) *Item {
return &Item{
Job: job.Name(),
Ident: job.ID(),
Payload: job.Payload(),
headers: job.Headers(),
Options: &Options{
Priority: job.Priority(),
Pipeline: job.GroupID(),
Delay: job.Delay(),
},
}
}