t3n / jobqueue.rabbitmq Goto Github PK

Since RabbitMQ 3.9 there is a new Queue Type called Streams: https://www.rabbitmq.com/streams.html

Adding support for Streams is actually quite simple, since the Protocol remains AMQP. This package already works with Streams with one exception: The consumer does not track it's offset in a stream.

This has two downsides:

• A new consumer that connects to an existing stream will not start from the beginning, but will only receive new messages
• A consumer that already consumed a stream will not continue at the same offset after a restart

RabbitMQ Streams support server-side offset tracking: https://blog.rabbitmq.com/posts/2021/09/rabbitmq-streams-offset-tracking/

Unfortunately I did not find a way to save the offset using php-amqplib, as the IOStream is not exposed to the outside. Since the blog post for offset recommends to also store the offset client-side we can simply create a persistent flow cache to store the offset of our consumers.

It should be possible to configure the RabbitQueue to use a DLX https://www.rabbitmq.com/dlx.html instead of the current requeueing mechanism.

Proposal: Add a configuration option to enable usage of a DLX.

If a DLX is in use, the release function could look like this:

public function release(string $messageId, array $options = []): void
{
    $this->channel->basic_nack($messageId);
}

This way, the message would not have to be requeued using the ReleaseService. Once that works, the x-death message header can be used to track the retry count of the message header.

Background:
I publish the jobs to a fanout exchange so multiple workers receive the same message. When a message fails, it is published to that exchange again, leading to duplicate messages in all worker queues.

Dependabot encountered the following error when parsing your .dependabot/config.yml:

The property '#/update_configs/0' did not contain a required property of 'directory'

Please update the config file to conform with Dependabot's specification using our docs and online validator.

The job:work command of the Flowpack.JobQueue.Common package supports the parameter --exit-after which allows to set a time limit after which the command exists.

If during that time no message is published to the RabbitMQ queue, an AMQPTimeoutException is thrown, setting the exit code of the job:work process to 1.

Expected behavior:
If no message is found in time, the RabbitQueue should return null as stated in the PHPDoc of the QueueInterface (See here).