|
1
|
|
|
<?php |
|
2
|
|
|
|
|
3
|
|
|
namespace Bernard; |
|
4
|
|
|
|
|
5
|
|
|
use Symfony\Component\EventDispatcher\EventDispatcherInterface; |
|
6
|
|
|
use Bernard\Event\EnvelopeEvent; |
|
7
|
|
|
use Bernard\Event\PingEvent; |
|
8
|
|
|
use Bernard\Event\RejectEnvelopeEvent; |
|
9
|
|
|
|
|
10
|
|
|
/** |
|
11
|
|
|
* @package Consumer |
|
12
|
|
|
*/ |
|
13
|
|
|
class Consumer |
|
14
|
|
|
{ |
|
15
|
|
|
protected $router; |
|
16
|
|
|
protected $dispatcher; |
|
17
|
|
|
protected $shutdown = false; |
|
18
|
|
|
protected $pause = false; |
|
19
|
|
|
protected $configured = false; |
|
20
|
|
|
protected $options = [ |
|
21
|
|
|
'max-runtime' => PHP_INT_MAX, |
|
22
|
|
|
'max-messages' => null, |
|
23
|
|
|
'stop-when-empty' => false, |
|
24
|
|
|
'stop-on-error' => false, |
|
25
|
|
|
]; |
|
26
|
|
|
|
|
27
|
|
|
/** |
|
28
|
|
|
* @param Router $router |
|
29
|
|
|
* @param EventDispatcherInterface $dispatcher |
|
30
|
|
|
*/ |
|
31
|
|
|
public function __construct(Router $router, EventDispatcherInterface $dispatcher) |
|
32
|
|
|
{ |
|
33
|
|
|
$this->router = $router; |
|
34
|
|
|
$this->dispatcher = $dispatcher; |
|
35
|
|
|
} |
|
36
|
|
|
|
|
37
|
|
|
/** |
|
38
|
|
|
* Starts an infinite loop calling Consumer::tick(); |
|
39
|
|
|
* |
|
40
|
|
|
* @param Queue $queue |
|
41
|
|
|
* @param array $options |
|
42
|
|
|
*/ |
|
43
|
|
|
public function consume(Queue $queue, array $options = []) |
|
44
|
|
|
{ |
|
45
|
|
|
declare (ticks = 1); |
|
46
|
|
|
|
|
47
|
|
|
$this->bind(); |
|
48
|
|
|
|
|
49
|
|
|
while ($this->tick($queue, $options)) { |
|
50
|
|
|
// NO op |
|
51
|
|
|
} |
|
52
|
|
|
} |
|
53
|
|
|
|
|
54
|
|
|
/** |
|
55
|
|
|
* Returns true do indicate it should be run again or false to indicate |
|
56
|
|
|
* it should not be run again. |
|
57
|
|
|
* |
|
58
|
|
|
* @param Queue $queue |
|
59
|
|
|
* @param array $options |
|
60
|
|
|
* |
|
61
|
|
|
* @return bool |
|
62
|
|
|
*/ |
|
63
|
|
|
public function tick(Queue $queue, array $options = []) |
|
64
|
|
|
{ |
|
65
|
|
|
$this->configure($options); |
|
66
|
|
|
|
|
67
|
|
|
if ($this->shutdown) { |
|
68
|
|
|
return false; |
|
69
|
|
|
} |
|
70
|
|
|
|
|
71
|
|
|
if (microtime(true) > $this->options['max-runtime']) { |
|
72
|
|
|
return false; |
|
73
|
|
|
} |
|
74
|
|
|
|
|
75
|
|
|
if ($this->pause) { |
|
76
|
|
|
return true; |
|
77
|
|
|
} |
|
78
|
|
|
|
|
79
|
|
|
$this->dispatcher->dispatch(BernardEvents::PING, new PingEvent($queue)); |
|
80
|
|
|
|
|
81
|
|
|
if (!$envelope = $queue->dequeue()) { |
|
82
|
|
|
return !$this->options['stop-when-empty']; |
|
83
|
|
|
} |
|
84
|
|
|
|
|
85
|
|
|
$this->invoke($envelope, $queue); |
|
86
|
|
|
|
|
87
|
|
|
if (null === $this->options['max-messages']) { |
|
88
|
|
|
return true; |
|
89
|
|
|
} |
|
90
|
|
|
|
|
91
|
|
|
return (boolean) --$this->options['max-messages']; |
|
92
|
|
|
} |
|
93
|
|
|
|
|
94
|
|
|
/** |
|
95
|
|
|
* Mark Consumer as shutdown |
|
96
|
|
|
*/ |
|
97
|
|
|
public function shutdown() |
|
98
|
|
|
{ |
|
99
|
|
|
$this->shutdown = true; |
|
100
|
|
|
} |
|
101
|
|
|
|
|
102
|
|
|
/** |
|
103
|
|
|
* Pause consuming |
|
104
|
|
|
*/ |
|
105
|
|
|
public function pause() |
|
106
|
|
|
{ |
|
107
|
|
|
$this->pause = true; |
|
108
|
|
|
} |
|
109
|
|
|
|
|
110
|
|
|
/** |
|
111
|
|
|
* Resume consuming |
|
112
|
|
|
*/ |
|
113
|
|
|
public function resume() |
|
114
|
|
|
{ |
|
115
|
|
|
$this->pause = false; |
|
116
|
|
|
} |
|
117
|
|
|
|
|
118
|
|
|
/** |
|
119
|
|
|
* Until there is a real extension point to doing invoked stuff, this can be used |
|
120
|
|
|
* by wrapping the invoke method. |
|
121
|
|
|
* |
|
122
|
|
|
* @param Envelope $envelope |
|
123
|
|
|
* @param Queue $queue |
|
124
|
|
|
* |
|
125
|
|
|
* @throws \Exception |
|
126
|
|
|
* @throws \Throwable |
|
127
|
|
|
*/ |
|
128
|
|
|
public function invoke(Envelope $envelope, Queue $queue) |
|
129
|
|
|
{ |
|
130
|
|
|
try { |
|
131
|
|
|
$this->dispatcher->dispatch(BernardEvents::INVOKE, new EnvelopeEvent($envelope, $queue)); |
|
132
|
|
|
|
|
133
|
|
|
// for 5.3 support where a function name is not a callable |
|
134
|
|
|
call_user_func($this->router->map($envelope), $envelope->getMessage()); |
|
135
|
|
|
|
|
136
|
|
|
// We successfully processed the message. |
|
137
|
|
|
$queue->acknowledge($envelope); |
|
138
|
|
|
|
|
139
|
|
|
$this->dispatcher->dispatch(BernardEvents::ACKNOWLEDGE, new EnvelopeEvent($envelope, $queue)); |
|
140
|
|
|
} catch (\Throwable $error) { |
|
|
|
|
|
|
141
|
|
|
$this->rejectDispatch($error, $envelope, $queue); |
|
142
|
|
|
} catch (\Exception $exception) { |
|
143
|
|
|
$this->rejectDispatch($exception, $envelope, $queue); |
|
144
|
|
|
} |
|
145
|
|
|
} |
|
146
|
|
|
|
|
147
|
|
|
/** |
|
148
|
|
|
* @param array $options |
|
149
|
|
|
*/ |
|
150
|
|
|
protected function configure(array $options) |
|
151
|
|
|
{ |
|
152
|
|
|
if ($this->configured) { |
|
153
|
|
|
return $this->options; |
|
154
|
|
|
} |
|
155
|
|
|
|
|
156
|
|
|
$this->options = array_filter($options) + $this->options; |
|
157
|
|
|
$this->options['max-runtime'] += microtime(true); |
|
158
|
|
|
$this->configured = true; |
|
159
|
|
|
} |
|
160
|
|
|
|
|
161
|
|
|
/** |
|
162
|
|
|
* Setup signal handlers for unix signals. |
|
163
|
|
|
* |
|
164
|
|
|
* If the process control extension does not exist (e.g. on Windows), ignore the signal handlers. |
|
165
|
|
|
* The difference is that when terminating the consumer, running processes will not stop gracefully |
|
166
|
|
|
* and will terminate immediately. |
|
167
|
|
|
*/ |
|
168
|
|
|
protected function bind() |
|
169
|
|
|
{ |
|
170
|
|
|
if (function_exists('pcntl_signal')) { |
|
171
|
|
|
pcntl_signal(SIGTERM, [$this, 'shutdown']); |
|
172
|
|
|
pcntl_signal(SIGINT, [$this, 'shutdown']); |
|
|
|
|
|
|
173
|
|
|
pcntl_signal(SIGQUIT, [$this, 'shutdown']); |
|
174
|
|
|
pcntl_signal(SIGUSR2, [$this, 'pause']); |
|
175
|
|
|
pcntl_signal(SIGCONT, [$this, 'resume']); |
|
176
|
|
|
} |
|
177
|
|
|
} |
|
178
|
|
|
|
|
179
|
|
|
/** |
|
180
|
|
|
* @param \Throwable|\Exception $exception note that the type-hint is missing due to PHP 5.x compat |
|
181
|
|
|
* |
|
182
|
|
|
* @param Envelope $envelope |
|
183
|
|
|
* @param Queue $queue |
|
184
|
|
|
* |
|
185
|
|
|
* @throws \Exception |
|
186
|
|
|
* @throws \Throwable |
|
187
|
|
|
*/ |
|
188
|
|
|
private function rejectDispatch($exception, Envelope $envelope, Queue $queue) |
|
189
|
|
|
{ |
|
190
|
|
|
// Make sure the exception is not interfering. |
|
191
|
|
|
// Previously failing jobs handling have been moved to a middleware. |
|
192
|
|
|
// |
|
193
|
|
|
// Emit an event to let others log that exception |
|
194
|
|
|
$this->dispatcher->dispatch(BernardEvents::REJECT, new RejectEnvelopeEvent($envelope, $queue, $exception)); |
|
195
|
|
|
|
|
196
|
|
|
if ($this->options['stop-on-error']) { |
|
197
|
|
|
throw $exception; |
|
198
|
|
|
} |
|
199
|
|
|
} |
|
200
|
|
|
} |
|
201
|
|
|
|
bernardphp /
bernard
Loading history...
Scrutinizer analyzes your
composer.json/composer.lockfile if available to determine the classes, and functions that are defined by your dependencies.It seems like the listed class was neither found in your dependencies, nor was it found in the analyzed files in your repository. If you are using some other form of dependency management, you might want to disable this analysis.