The driver can work with any queue interop compatible transport such as:

• enqueue/amqp-ext
• enqueue/amqp-lib
• enqueue/amqp-bunny
• enqueue/pheanstalk
• enqueue/stomp
• enqueue/sqs
• enqueue/rdkafka
• enqueue/redis
• enqueue/gearman
• enqueue/dbal
• enqueue/fs
• enqueue/null.
• zogwarg/enqueue-elasticsearch

It also provide an extended driver version for amqp interop compatible transports. It supports some amqp specific features such as queue declaration and message couting.

TODO:

• [x] add tests
• [x] add doc

I welcome everyone who is interested in MQs to join the queue interop group

Added options array to queue create and message send/push so we can pass options to the specific methods of each driver.

I've added this to fix #205, but I will expand it usage of the options so you can use all options of each driver method (queue create params and message push params)

I will create a seperate PR to add the docs for this and one for the usage of all params (for each specifc driver)

TODO:

• [x] Check if all option definitions are ok
• [x] Fix travis-ci build config so we test against symfony 2.3 and 3.0 (options-resolver)
• [x] Add BC layer to each configure*Options method (to support 2.3 and 3.0)

Two years old, still one of my most linked article: http://verraes.net/2013/09/sensible-interfaces/

One of the points here (which I agree with) is that Default means nothing. Default what?

Although it is a default implementation, I think PlainMessage sounds better. WDYT?

i am having a weird issue with bernard, when i run the command to consume manually on the console it works fine, however when i tell supervisord to run things, supervisord still works ok, but the php process is stopped and does not continue/finish == no logs to debug.

here is my conf:

[program:algorithmOne]
command=/usr/bin/php index.php consume algorithm-one
process_name=%(program_name)s
numprocs=1
directory=/tmp
priority=999
autostart=true
autorestart=true
startretries=3
stopwaitsecs=180
user=root
stdout_logfile=/var/log/supervisord.log
stdout_logfile_maxbytes=25MB

any hints on what could be happenign or how to go around it?

Is it possible to somehow ping the consumers, or let the consumers notify when they are active? Would be very nice to assert that there are consumers running before dispatching work (if not fallback to "normal" approach).

Since we don't have UPSERT support in DBAL (and sadly will not be in it for a while), this PR fixes #251 by providing graceful handling of queue creation.

1. operations are executed in a transaction
2. only unique constraint exceptions are caught (very important! if the DB dies, that's not something we can ignore)
3. avoid causing duplicate key issues, when possible

Somehow my messages are not marked as acknowledged in RabbitMQ and it seems not to work for other drivers as well. When I create a new default message like this;

$this->get('bernard.producer')->produce(new DefaultMessage('SubmitMessage', [
    'id' => $entity->getId()
]));

It goes correctly to a queue called submit-message. So I run a consumer like this;

app/console bernard:consume submit-message --no-debug

And it processes the message correctly. But it's never marked as acknowledged because of this line. The $queue variable contains the string SubmitMessage but the queue's key is submit-message, so this isset will always fail. Am I doing something wrong or is this a bug?

We use the DoctrineDriver and every now and then we'll hit a dead lock.

I think the reason for this is that the DoctrineDriver creates a bernard_messages table where the id is autoincrement.

This locks the table every time an insert is executed.

I think we should use UUID's and get rid of the autoincrement ids. The only problem is that we need to sort based on date + microtime.

The current producer/consumer model is based on the assumption that messages are published with a specified envelope name that contributes to how they are routed; more specifically, that they are routed to a queue based on that envelope name and that a separate consumer process is maintained per queue.

This is a very limiting architecture in infrastructures like Heroku, where I may only want to operate a single worker process across all queues because limited traffic doesn't warrant a separate process per queue.

It is possible to work around this limitation by sending messages to a single queue, extending the router, and having it resolve the receiver differently based on some attribute other than the envelope name, such as a message argument.

However, this solution is just that: a work-around. It would be nice if this was something the Bernard architecture supported natively through a more interface-driven design with consumers.

Have been thinking about Middlewares and how they work. I like them since they are easy to create and the interface is simple, i dont like them when you have functionality that are shared between consuming and producing. Because of that i propose the following.

1. Use events where the middlewares are currently used consume, produce and exception.
2. Create a Strategy implementation for executing messages. which will be a layer between the Consumer and calling the method.
3. Remove the middleware notion.

This work make the BatchMiddleware look as following (mockup).

<?php

namespace Bernard\EventDispatcher;

use Bernard\Envelope;
use Bernard\Queue;
use Bernard\Batch\Storage;
use Bernard\EventDispatcher;

class BatchListener implements EventSubscriber
{
    protected $storage;

    public function __construct(Storage $storage)
    {
        $this->storage;
    }

    public function onProduce(Envelope $envelope, Queue $queue)
    {
        if (false == $batch = $event->getEnvelope()->getStamp('batch')) {
            return;
        }

        $this->storage->register($batch);
    }

    public function onConsume(Envelope $envelope, Queue $queue)
    {
        if (false == $batch = $event->getEnvelope()->getStamp('batch')) {
            return;
        }

        $this->storage->increment($batch, 'consumed');
    }

    public function onException(Envelope $envelope, Queue $queue, $exception)
    {
        if (false == $batch = $event->getEnvelope()->getStamp('batch')) {
            return;
        }

        $this->storage->increment($batch, 'failed');
    }

    public function subscribe(EventDispatcher $dispatcher)
    {
        $dispatcher->on('bernard.consume', [$this, 'onConsume']);
        $dispatcher->on('bernard.exception', [$this, 'onException']);
        $dispatcher->on('bernard.produce', [$this, 'onProduce']);
    }
}

The convenience for this is that it will only requirering creating a single class for adding stuff of both the producer and consumer side.

the Strategy is to provide a extension point for executing jobs in different ways suchs as threads or forks with Spork.

PR #216 has introduced a bug in the doctrine driver. It does not pop messages from a queue anymore and fails silently due to the try ... catch in DoctrineDriver::popMessage. After some debugging I've found the error in the select ... for update statement that was modified in the linked PR. Can you release a 0.12.4 with the fix please?

This fixes the deprecation notices reported in #425 by specifying the int return type for the count functions.

Initially I wanted to use the master branch but it is lacking the Predis driver that we use.

I figured this would be more inclusive than using #[\ReturnTypeWillChange] given we only need to bump to PHP7 or higher and not PHP8.

Since this removes support for PHP5 (about time I guess? :)) I would suggest to release this as 0.14.0.

I use bernard with PHP 8.1, and it works, however there are some deprecation issues:

PHP Deprecated:  Return type of Bernard\QueueFactory\PersistentFactory::count() should either be compatible with Countable::count(): int, or the #[\ReturnTypeWillChange] attribute should be used to temporarily suppress the notice in /webapps/vendor/bernard/bernard/src/Bernard/QueueFactory/PersistentFactory.php on line 68

Deprecated: Return type of Bernard\QueueFactory\PersistentFactory::count() should either be compatible with Countable::count(): int, or the #[\ReturnTypeWillChange] attribute should be used to temporarily suppress the notice in /webapps/vendor/bernard/bernard/src/Bernard/QueueFactory/PersistentFactory.php on line 68

PHP Deprecated:  Return type of Bernard\Queue\PersistentQueue::count() should either be compatible with Countable::count(): int, or the #[\ReturnTypeWillChange] attribute should be used to temporarily suppress the notice in /webapps/vendor/bernard/bernard/src/Bernard/Queue/PersistentQueue.php on line 49

Deprecated: Return type of Bernard\Queue\PersistentQueue::count() should either be compatible with Countable::count(): int, or the #[\ReturnTypeWillChange] attribute should be used to temporarily suppress the notice in /webapps/vendor/bernard/bernard/src/Bernard/Queue/PersistentQueue.php on line 49

Currently the receipt can be anything thanks to the mixed type hint. I wonder if that makes sense as mixed includes types, like resource, boolean and array.

Are those types potential receipts?

Object is something I could imagine as a receipt.

Hi,

I am new to this project and I am trying to figure out how message serialization works. I know the default place for questions is usually stack overflow, but currently it doesn't have a bernard tag and I had a feeling that I would need to wait a few months to get an answer there :)

Bernard installed from composer (currently v0.13) comes with Bernard\Serializer\SimpleSerializer class and suggests using symfony or JMS serializers instead. This doc says that currently bernard uses symfony serializer by default.

On the other hand I couldn't find symfony/serializer in composer.json. The Bernard\Serializer namespace is gone and Bernard/Serializer class seems to use json_encode / json_decode. To confuse things even more, Bernard/Serializer is an interface in the latest composer and a class in the latest master branch.

So I figured I should go read the docs again :) This doc says that if I plan to use a custom message class (which I do), then I must provide a normalizer for it. Except there is no such thing in the currently available composer version of bernard.

My questions are:

PHP native serialize() function should be able to handle simple classes like PlainMessage. Why does bernard need complex serializers from symfony or JMS even for the most basic message type?

It seems that 0.x branch (which is installed by default via composer) is many years behind the current master and current documentation is no longer usable for 0.x. Is it generally a good idea to use the current master instead?

When I follow the README and executecomposer require bernard/bernard the version of bernard installed is, bernard/bernard": "^0.13.0". I see Bernard\Router\ReceiverMapRouter exists in master, and the only way I am aware of getting the latest version is if I spell out in my composer file the following "bernard/bernard": "dev-master".