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
|
package beanstalk
import (
"strings"
"time"
"github.com/beanstalkd/go-beanstalk"
"github.com/spiral/errors"
"github.com/spiral/roadrunner/v2/pkg/events"
priorityqueue "github.com/spiral/roadrunner/v2/pkg/priority_queue"
"github.com/spiral/roadrunner/v2/plugins/config"
"github.com/spiral/roadrunner/v2/plugins/jobs/job"
"github.com/spiral/roadrunner/v2/plugins/jobs/pipeline"
"github.com/spiral/roadrunner/v2/plugins/logger"
)
type JobConsumer struct {
log logger.Logger
eh events.Handler
pq priorityqueue.Queue
// beanstalk
conn *beanstalk.Conn
tout time.Duration
// tube name
tName string
}
func NewBeanstalkConsumer(configKey string, log logger.Logger, cfg config.Configurer, e events.Handler, pq priorityqueue.Queue) (*JobConsumer, error) {
const op = errors.Op("new_beanstalk_consumer")
// PARSE CONFIGURATION -------
var pipeCfg Config
var globalCfg GlobalCfg
err := cfg.UnmarshalKey(configKey, &pipeCfg)
if err != nil {
return nil, errors.E(op, err)
}
pipeCfg.InitDefault()
err = cfg.UnmarshalKey(pluginName, &globalCfg)
if err != nil {
return nil, errors.E(op, err)
}
globalCfg.InitDefault()
// initialize job consumer
jc := &JobConsumer{
pq: pq,
log: log,
eh: e,
tout: globalCfg.Timeout,
}
// PARSE CONFIGURATION -------
dsn := strings.Split(globalCfg.Addr, "://")
if len(dsn) != 2 {
return nil, errors.E(op, errors.Errorf("invalid socket DSN (tcp://localhost:11300, unix://beanstalk.sock), provided: %s", globalCfg.Addr))
}
jc.conn, err = beanstalk.DialTimeout(dsn[0], dsn[1], jc.tout)
if err != nil {
return nil, err
}
// start redial listener
go jc.redial()
return jc, nil
}
func FromPipeline(pipe *pipeline.Pipeline, log logger.Logger, cfg config.Configurer, e events.Handler, pq priorityqueue.Queue) (*JobConsumer, error) {
const op = errors.Op("new_beanstalk_consumer")
const (
tube string = "tube"
)
// PARSE CONFIGURATION -------
var globalCfg GlobalCfg
err := cfg.UnmarshalKey(pluginName, &globalCfg)
if err != nil {
return nil, errors.E(op, err)
}
globalCfg.InitDefault()
// initialize job consumer
jc := &JobConsumer{
pq: pq,
log: log,
eh: e,
tout: globalCfg.Timeout,
tName: pipe.String(tube, ""),
}
// PARSE CONFIGURATION -------
dsn := strings.Split(globalCfg.Addr, "://")
if len(dsn) != 2 {
return nil, errors.E(op, errors.Errorf("invalid socket DSN (tcp://localhost:11300, unix://beanstalk.sock), provided: %s", globalCfg.Addr))
}
jc.conn, err = beanstalk.DialTimeout(dsn[0], dsn[1], jc.tout)
if err != nil {
return nil, err
}
// start redial listener
go jc.redial()
return jc, nil
}
func (j *JobConsumer) Push(job *job.Job) error {
panic("implement me")
}
func (j *JobConsumer) Register(pipeline *pipeline.Pipeline) error {
panic("implement me")
}
func (j *JobConsumer) Run(pipeline *pipeline.Pipeline) error {
panic("implement me")
}
func (j *JobConsumer) Stop() error {
panic("implement me")
}
func (j *JobConsumer) Pause(pipeline string) {
panic("implement me")
}
func (j *JobConsumer) Resume(pipeline string) {
panic("implement me")
}
|
