t3n / jobqueue.rabbitmq Goto Github PK
View Code? Open in Web Editor NEWRabbitMQ backend implementation for Flowpack.JobQueue.Common
RabbitMQ backend implementation for Flowpack.JobQueue.Common
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:
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).
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.