Skip to content

Commit ec6d36b

Browse files
adm-bomeadm-bome
authored
Fix laravel job attempts, when removing DLX (vyuldashev#304)
* Feature: Alter the delay-strategy from config. - manage the delay-strategy from config. - add backoff-strategy on delayed messages. - add prioritization of delayed messages. * Correct uneccesary git deletions (for readability) * moved: delay config to the end of the file * Fixed the updating of attempts when pushing the job back to the queue * Fixed losing messages when no delay was set * Cleanup fork * Cleanup fork * Code style cleanup * Code style cleanup * Ditched the xDeathCount because the value is not a controlled value. The value can mess things up. The attempts() method was designed to return the current attempt value, not the value of the previous run. Else the worker will try 1 try to long. * Fixed Tests The attempts() method on the Job is/was interpreted wrong. the method mus return the previous_attempt + 1 = current attempt. * Added prioritization when messages put back onto the queue when relaesed. Previously this was handled by rabbitmq with a DLX. [removed] * Fixed Style issues * - Added exchange publishing. - Added delayed messages with a higher priority when put back to the queue. - Added ability to reroute failed messages - Fixed when to acknowledge or decline messages. - Fixed attempts() when multiple delay values are used. - Fixed death worker loop, when queues are missing. - Fixed issue vyuldashev#305 - Fixed issue vyuldashev#302 - Fixed issue vyuldashev#301 - Fixed issue vyuldashev#299 * - Fixed style issues (not showingup in test:style) * - Altered the config structure * - Updated the readMe * - Fixed style * - Fixed Tests (bump) * - Altered the default priority levels when using this feature * - Fixed test for getQueueMaxPriority()
1 parent 6a7e861 commit ec6d36b

File tree

9 files changed

+739
-74
lines changed

9 files changed

+739
-74
lines changed

‎README.md‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,94 @@ Add connection to `config/queue.php`:
7373
],
7474
```
7575

76+
### Optional Config
77+
78+
Optionally add queue options to the config of a connection.
79+
Every queue created for this connection, get's the properties.
80+
81+
When you want to prioritize messages when they were delayed, then this is possible by adding extra options.
82+
- When max-priority is omitted, the max priority is set with 2 when used.
83+
84+
```php
85+
'connections' => [
86+
// ...
87+
88+
'rabbitmq' => [
89+
// ...
90+
91+
'options' => [
92+
'queue' => [
93+
// ...
94+
95+
'prioritize_delayed_messages' => false,
96+
'queue_max_priority' => 10,
97+
],
98+
],
99+
],
100+
101+
// ...
102+
],
103+
```
104+
105+
When you want to publish messages against an exchange with routing-key's, then this is possible by adding extra options.
106+
- When the exchange is omitted, RabbitMQ will use the `amq.direct` exchange for the routing-key
107+
- When routing-key is omitted the routing-key by default is the `queue` name.
108+
- When using `%s` in the routing-key the queue_name will be substituted.
109+
110+
> Note: when using exchange with routing-key, u probably create your queues with bindings yourself.
111+
112+
```php
113+
'connections' => [
114+
// ...
115+
116+
'rabbitmq' => [
117+
// ...
118+
119+
'options' => [
120+
'queue' => [
121+
// ...
122+
123+
'exchange' => 'application-x',
124+
'exchange_type' => 'topic',
125+
'exchange_routing_key' => '',
126+
],
127+
],
128+
],
129+
130+
// ...
131+
],
132+
```
133+
134+
In Laravel failed jobs are stored into the database. But maybe you want to instruct some other process to also do something with the message.
135+
When you want to instruct RabbitMQ to reroute failed messages to a exchange or a specific queue, then this is possible by adding extra options.
136+
- When the exchange is omitted, RabbitMQ will use the `amq.direct` exchange for the routing-key
137+
- When routing-key is omitted, the routing-key by default the `queue` name is substituted with `'.failed'`.
138+
- When using `%s` in the routing-key the queue_name will be substituted.
139+
140+
> Note: When using failed_job exchange with routing-key, u probably need to create your exchange/queue with bindings yourself.
141+
142+
```php
143+
'connections' => [
144+
// ...
145+
146+
'rabbitmq' => [
147+
// ...
148+
149+
'options' => [
150+
'queue' => [
151+
// ...
152+
153+
'reroute_failed' => true,
154+
'failed_exchange' => 'failed-exchange',
155+
'failed_routing_key' => 'application-x.%s',
156+
],
157+
],
158+
],
159+
160+
// ...
161+
],
162+
```
163+
76164
## Laravel Usage
77165

78166
Once you completed the configuration you can use Laravel Queue API. If you used other queue drivers you do not need to change anything else. If you do not know how to use Queue API, please refer to the official Laravel documentation: http://laravel.com/docs/queues

‎src/Console/ExchangeDeclareCommand.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,6 @@ public function handle(RabbitMQConnector $connector): void
4040
(bool) $this->option('auto-delete')
4141
);
4242

43-
$this->warn('Exchange declared successfully.');
43+
$this->info('Exchange declared successfully.');
4444
}
4545
}

‎src/Queue/Connectors/RabbitMQConnector.php‎

Lines changed: 24 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -38,12 +38,13 @@ public function __construct(Dispatcher $dispatcher)
3838
*/
3939
public function connect(array $config): Queue
4040
{
41-
$connection = $this->createConnection($config);
41+
$connection = $this->createConnection(Arr::except($config, 'options.queue'));
4242

4343
$queue = $this->createQueue(
4444
Arr::get($config, 'worker', 'default'),
4545
$connection,
46-
$config['queue']
46+
$config['queue'],
47+
Arr::get($config, 'options.queue', [])
4748
);
4849

4950
if (! $queue instanceof RabbitMQQueue) {
@@ -71,29 +72,42 @@ protected function createConnection(array $config): AbstractConnection
7172
/** @var AbstractConnection $connection */
7273
$connection = Arr::get($config, 'connection', AMQPLazyConnection::class);
7374

74-
$hosts = Arr::shuffle(Arr::get($config, 'hosts', []));
75-
7675
// manually disable heartbeat so long-running tasks will not fail
77-
$config['options']['heartbeat'] = 0;
76+
Arr::set($config, 'options.heartbeat', 0);
7877

7978
return $connection::create_connection(
80-
$hosts,
79+
Arr::shuffle(Arr::get($config, 'hosts', [])),
8180
$this->filter(Arr::get($config, 'options', []))
8281
);
8382
}
8483

85-
protected function createQueue(string $worker, AbstractConnection $connection, string $queue)
84+
/**
85+
* Create a queue for the worker.
86+
*
87+
* @param string $worker
88+
* @param AbstractConnection $connection
89+
* @param string $queue
90+
* @param array $options
91+
* @return HorizonRabbitMQQueue|RabbitMQQueue|Queue
92+
*/
93+
protected function createQueue(string $worker, AbstractConnection $connection, string $queue, array $options = [])
8694
{
8795
switch ($worker) {
8896
case 'default':
89-
return new RabbitMQQueue($connection, $queue);
97+
return new RabbitMQQueue($connection, $queue, $options);
9098
case 'horizon':
91-
return new HorizonRabbitMQQueue($connection, $queue);
99+
return new HorizonRabbitMQQueue($connection, $queue, $options);
92100
default:
93-
return new $worker($connection, $queue);
101+
return new $worker($connection, $queue, $options);
94102
}
95103
}
96104

105+
/**
106+
* Recursively filter only null values.
107+
*
108+
* @param array $array
109+
* @return array
110+
*/
97111
private function filter(array $array): array
98112
{
99113
foreach ($array as $index => &$value) {

‎src/Queue/Jobs/RabbitMQJob.php‎

Lines changed: 24 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -75,15 +75,25 @@ public function attempts(): int
7575
$headers = Arr::get($this->message->get_properties(), 'application_headers');
7676

7777
if (! $headers) {
78-
return 0;
78+
return 1;
7979
}
8080

8181
$data = $headers->getNativeData();
8282

8383
$laravelAttempts = (int) Arr::get($data, 'laravel.attempts', 0);
8484
$xDeathCount = (int) Arr::get($headers->getNativeData(), 'x-death.0.count', 0);
8585

86-
return $laravelAttempts + $xDeathCount;
86+
return ($laravelAttempts) + 1;
87+
}
88+
89+
public function fail($e = null): void
90+
{
91+
parent::fail($e);
92+
93+
// We must tel rabbitMQ this Job is failed
94+
// The message must be rejected when the Job marked as failed, in case rabbitMQ wants to do some extra magic.
95+
// like: Death lettering the message to an other exchange/routing-key.
96+
$this->rabbitmq->reject($this);
8797
}
8898

8999
/**
@@ -95,7 +105,11 @@ public function delete(): void
95105
{
96106
parent::delete();
97107

98-
$this->rabbitmq->ack($this);
108+
// When delete is called and the Job was not failed, the message must be acknowledged.
109+
// This is because this is a controlled call by a developer. So the message was handled correct.
110+
if (! $this->failed) {
111+
$this->rabbitmq->ack($this);
112+
}
99113

100114
// required for Laravel Horizon
101115
if ($this->rabbitmq instanceof HorizonRabbitMQQueue) {
@@ -108,17 +122,15 @@ public function delete(): void
108122
*/
109123
public function release($delay = 0): void
110124
{
111-
parent::release($delay);
112-
113-
if ($delay > 0) {
114-
$this->rabbitmq->ack($this);
125+
parent::release();
115126

116-
$this->rabbitmq->laterRaw($delay, $this->message->body, $this->queue, $this->attempts());
127+
// Always create a new message when this Job is released
128+
$this->rabbitmq->laterRaw($delay, $this->message->body, $this->queue, $this->attempts());
117129

118-
return;
119-
}
120-
121-
$this->rabbitmq->reject($this);
130+
// Releasing a Job means the message was failed to process.
131+
// Because this Job is always recreated and pushed as new message, this Job is correctly handled.
132+
// We must tell rabbitMQ this fact.
133+
$this->rabbitmq->ack($this);
122134
}
123135

124136
/**

0 commit comments

Comments
 (0)