soyuka / symfony-messenger-redis
Redis adapater for the symfony/messenger component
Installs: 2 062
Dependents: 0
Suggesters: 0
Security: 0
Stars: 11
Watchers: 5
Forks: 5
Open Issues: 0
Requires
- php: >=7.1.0
- ext-redis: *
Requires (Dev)
- phpunit/phpunit: ^7.1@dev
- symfony/config: ^4.1@dev
- symfony/dependency-injection: ^4.1@dev
- symfony/messenger: ^4.0@dev
- symfony/property-access: ^4.1@dev
- symfony/serializer: ^4.1@dev
Suggests
- ext-igbinary: *
README
Hello! This bundle is deprecated as symfony has now a Redis transport. It's available since symfony 4.3.0-BETA1
This is an experimental Receiver/Sender on Redis for the symfony/messenger component.
Quick start
For now we're exposing a bundle which is pre-configuring the Messenger component with receivers and senders.
composer require symfony/messenger soyuka/symfony-messenger-redis
Add the bundle new Soyuka\RedisMessengerAdapter\Bundle\RedisMessengerAdapterBundle()
.
Also requires the redis extension.
Add the following configuration:
redis_messenger_adapter: messages: 'App\Message\Foo': 'foo_queue'
Add a message handler:
<?php namespace App\MessageHandler; use App\Message\Foo; final class FooHandler { public function __invoke(Foo $message) { } }
Tag it:
services: App\MessageHandler\FooHandler: tags: - { name: messenger.message_handler }
You're done!
Launch bin/console messenger:consume-messages redis_messenger.receiver.foo_queue
and dispatch messages from the bus:
<?php $bus->dispatch(new Foo());
Configuration reference
redis_messenger_adapter: redis: url: '127.0.0.1' port: 6379 serializer: !php/const \Redis::SERIALIZER_IGBINARY # default is \Redis::SERIALIZER_PHP messages: 'App\Message\Foo': 'foo_queue' 'App\Message\Bar': queue: 'bar_queue' ttl: 10000 blockingTimeout: 1000
Internals
Relevant discussion: https://twitter.com/jderusse/status/980768426116485122
The sender uses a List and uses RPUSH
(add value to the tail of the list).
The receiver uses BRPOPLPUSH
which reads the last element of the list and adds in to the head of another list (queue_processing
). If no elements are present it'll block the connection until a new element shows up or the timeout is reached. When timeout is reached it works like a "ping" of some sort (todo wait for 26632 to be merged and $handle(null)
).
On every iteration, we will check the queue_processing
list. For every items in this queue we have a corresponding key
in redis with a given ttl
. If the key
has expired, the item is LREM
(removed) from queue_processing
and put back in the origin queue to be processed again.
This workaround helps avoiding lost messages.
I started a RedisAdapter and may add it to symfony once messenger documentation and AMQP adapter are merged.
- symfony/symfony#26632 (AMQP adapter PR)
- symfony/symfony-docs#9437 (messenger documentation)
- https://github.com/symfony/symfony/tree/master/src/Symfony/Component/Messenger (messenger component code)