Message Queue, Job Queue, Broadcasting, WebSockets packages for PHP, Symfony, Laravel, Magento. DEVELOPMENT REPOSITORY - provided by Forma-Pro

I am totally confused with documentation that is constructed really poor.

I understand that we have topics to connect messages to processors, but there is no information about connecting this to transport queue. There is also no info about ability of configuration more clients (Clients or topics?). Naming in config is also confusing, like 'routerQueue' vs. 'defaultQueue'.

Wasted lot of time to research it on my own...

When an error occurs in RdKafkaProducer, it is silently ignored - or, more accurately, since it occurs outside of the main thread of php/php-fpm, it is not reported back.

A simple test can be done to see this issue: trying to send a message to a non-existent Kafka server will result in nothing (see point 2. though)

This results in two things:

1. Process has no way of knowing, that a message that was supposed to be delivered will be not. Producer returns immediately without waiting for message to be acknowledged.
2. Due to how it is handled in arnaud-lb/php-rdkafka, process that was supposed to send the message will be locked and retry the operation for a long time (from my testing it seems around 5 mins, wasn't able to find a configuration option to change it to something else). In my case with default configuration of php-fpm docker image it resulted in fpm pool becoming locked after 5 requests being made, since thats the default configuration for max spawned children of php-fpm.
   While this particular part is not really possible to fix inside enqueue, it's important since the message might actually be delivered later on.

I'd like to ask for opinion regarding how RdKafkaProducer should handle this situation. IMO it is worthwile to add a configuration option to make sending messages synchronous for this particular Producer - or at least wait a specified amount of time for message to be potentially acknowledged. This can be done by introducing this code at the end of send method:

$topic = $this->producer->newTopic($destination->getTopicName(), $destination->getConf());
$topic->produce($partition, 0 /* must be 0 */, $payload, $key);
$start = microtime(true);
while ($this->producer->getOutQLen() > 0) {
     $this->producer->poll(1);

     if (microtime(true) - $start > 10) {
           throw new \RuntimeException("Message sending failed");
     }
}

This has a side effect of actually calling dr_msg_cb & error_cb callbacks, which are otherwise ignored (or at least that's what my testing indicated).

Thoughts?

Closes: https://github.com/php-enqueue/enqueue-dev/issues/1097

To detail my scenario, I'm running a long-lived PHP process (laravel queue worker) that listens on one connection, but can occasionally publish to another connection. The connection used for listening doesn't suffer from any issues because enqueue successfully sets up heartbeats, anticipating the fact that consumers tend to be long lived. For producers however, there might be the assumption that they are transient (web requests) and as such, don't require connections to live any longer than it takes for the request thread to finish all its work.

@makasim - My apologies if I've missed something or if this seems like a shoehorn fix / neglects any other concerns. After doing some digging around however I am starting to wonder if this might actually be a desirable approach.

What I've noticed in the stomp-php library is that their implementation of Client::isConnected has the potential to report false positives for the state of the connection and that without getting into a more convoluted fix, can only be "tested" by attempting to use it.

With the code fix in this PR, failures to write are jogged with one last-ditch attempt to connect. My long lived mixed consumer+producer processes are able to send messages and reestablish connections on-demand with no ceremony or boilerplate.

I'd like to add the caveat that this is not my preferred strategy and would favour something more explicit, but just based on the fact that StompProducer isn't set up for configuration and the aforementioned false-positives from stomp-php, this may do well for all scenarios.

The method \Enqueue\Redis\RedisProducer::setPriority currently throws an logic exception for the reason, that this isn't implemented. This exception happens on consuming data from the queue.

But it is possible to send messages having a priority.

Therefor, the whole queue went invalid. It's no longer possible to process messages if this happens.

If the Redis producer doesn't care about priority, this information should be simply ignored. Throwing an exception isn't necessary and useful at all.

Suppose we have two application that have installed php-enqueue. When I try to send a message from one to other application I have to run a send action twice. The steps what I did to reproduce the problem:

1. Run ./app/console enqueue:consume --setup-broker -vvv command for every application
2. Run sending action for the first time and look at console of application that was used to send a message. It returned a description of sent message, but there is nothing in console of application that must receive the message and send the answer. I havent got reply message, but got hanging request in my browser or console. After timeout browser or console writes "Rpc call timeout is reached without receiving a reply message"
3. Run the same action second time and get nothing in first application console (where I got a message description at the first time), but get message description in second application console (where I can see tha message was recieved and answer was sent). And eventualy I get result in thr browser or console. Important! I this step, after sending request second time I get message, that was sent in previous step,

Next I'll show all configurations, producers and processors of both applications. I used Enqueue Symfony bundle for my applications with next configs of both applications

enqueue:
    transport:
        default: "amqp"
        amqp:
            driver: ext
            host: "%env(RABBIT_HOST)%"
            port: "%env(RABBIT_PORT)%"
            user: "%env(RABBIT_USER)%"
            pass: "%env(RABBIT_PASSWORD)%"
            vhost: "%env(RABBIT_DEFAULT_VHOST)%"
            receive_method: basic_consume
    client:
        app_name: "message-bus"

Producer in sending application is

<?php
namespace AcmeBundle\Service\Producer;

use Enqueue\Client\Message;

class RequestProducer {
    /** @var Producer */
    protected $producer;

    /**
     * RequestProducer constructor.
     * @param ProducerInterface $producer
     */
    public function __construct(ProducerInterface $producer)
    {
        $this->producer = $producer;
    }

    /**
     * Sent request and get response from a service
     *
     * @param $content
     * @return object
     */
    public function send($content)
    {
        $message = new Message($content);
        $response = $this->producer->sendCommand('currentAppName', $message, true)->receive(5000)->getBody();

        return $response;
    }
}

Processor in receiving application

<?php
namespace AcmeBundle\Service\Processor;

use Interop\Queue\PsrProcessor;
use Interop\Queue\PsrMessage;
use Interop\Queue\PsrContext;
use Enqueue\Consumption\Result;
use Enqueue\Client\CommandSubscriberInterface;

class RequestProcessor implements PsrProcessor, CommandSubscriberInterface
{
    /**
     * {@inheritdoc}
     */
    public function process(PsrMessage $message, PsrContext $context)
    {
        return Result::reply(
            $context->createMessage(
                'This is an answer'
            )
        );
    }

    /**
     * {@inheritdoc}
     */
    public static function getSubscribedCommand()
    {
        return [
            'processorName' => 'currentAppName',
            'queueName' =>  'currentAppName',
            'queueNameHardcoded' => true,
            'exclusive' => true,
        ];
    }
}

When I put both processor and producer in one application they work perfect without wasting time and all request get response. If I put they in different applications I'll get situation described above

This will allow to Symfony users that have autoconfigure/autowiring enabled to avoid defining explicitly the processor service and its tag. Now, the symfony container does it automatically.

Hello, i'm getting a lot of errors with not found object (after creating object)

error:

[info] Message received from the queue: fos_elastica_doctrine_orm_sync_index_with_object_change
[debug] Headers: array (
)
[debug] Properties: array (
)
[debug] Payload: '{"action":"insert","model_class":"AppBundle\\\\Entity\\\\Order","model_id":"id","id":1744210,"index_name":"app","type_name":"order"}'
[debug] [ExclusiveCommandExtension] This is a exclusive command queue and client's properties are not set. Setting them
[info] Message processed: enqueue.ack
[info] The object "AppBundle\Entity\Order" with id "1744210" could not be found.

also, when i'm trying update entity, i'm getting not latest object, but previous. Seems like it caches somewhere

maybe we need to have delay? Or i've missed something in configuration?

My consumer always fails to commit the offset:

%4|1560006005.731|COMMITFAIL|rdkafka#consumer-20| [thrd:main]: Offset commit (manual) failed for 1/1 partition(s): Broker: Unknown member: commands[0]@1(Broker: Unknown member)

Here is my Config:

        consumer:
            dsn: "..."
            global:
                ### Make sure this is unique for each application / consumer group and does not change
                ### Otherwise, Kafka won't be able to track your last offset and will always start according to
                ### `auto.offset.reset` setting.
                ### See Kafka documentation regarding `group.id` property if you want to know more
                group.id: 'backend'
                metadata.broker.list: '...'
                security.protocol: sasl_ssl
                ssl.ca.location: '%kernel.project_dir%/config/kafka/ca.pem'
                sasl.username: '...'
                sasl.password: '...'
                sasl.mechanisms: "SCRAM-SHA-256"
                max.poll.interval.ms: '25000'
                enable.auto.commit: 'false'
                enable.auto.offset.store: 'false'
            topic:
                auto.offset.reset: beginning
            ### Commit async is true by default since version 0.9.9.
            ### It is suggested to set it to true in earlier versions since otherwise consumers become extremely slow,
            ### waiting for offset to be stored on Kafka before continuing.
            commit_async: true

Any ideas, why this is happening?

Tested with both heroku Kafka & CloudKarafka.

In a nutshell

• I was able to setup the queue on AWS and send messages to it
• there is no way I can get the broker to eat these messages

Constraints

• has to work with multiple environments (staging, production) which have separate SQS queues and ES indices
• the AWS IAM user identify it's using has minimal SQS permission

Populating the queue

Setting up the AWS AIM and SQS

• create a 'acme-testing' user account in AWS AIM
• create 'acme_fos_elastica_populate' and 'acme_fos_elastica_populate_reply_queue' SQS queues

Setting up the EventListener

    /**
     * Specifies AWS SQS queue name.
     *
     * @param PrePersistEvent $event
     */
    public function prePersist(PrePersistEvent $event)
    {
        $options = $event->getOptions();

        $options['max_per_page'] = 1000;
        $options['populate_queue'] = $this->queueName;
        $options['populate_reply_queue'] = $this->replyQueueName;

        $event->setOptions($options);
    }

And then I can see options passed to Enqueue\ElasticaBundle\Persister\QueuePagerPersister

$event = new PrePersistEvent($pager, $objectPersister, $options);
$this->dispatcher->dispatch(Events::PRE_PERSIST, $event);
$pager = $event->getPager();
$options = $event->getOptions();

And ultimately, in AWS web-gui the queue becomes populated with messages. This part works great.

Digesting the messages

Here is where problems begin. The official docs say that I should run

$ ./bin/console enqueue:consume --setup-broker -vvv 

But this ends badly

$ ./bin/console enqueue:consume --setup-broker -vvv 
[debug] Set context's logger "Symfony\Component\Console\Logger\ConsoleLogger"
[debug] [SqsDriver] Declare router queue: enqueue_dot_app_dot_default

In WrappedHttpHandler.php line 191:
                                                                                                                                                                                                         
  [Aws\Sqs\Exception\SqsException]                                                                                                                                                                       
  Error executing "CreateQueue" on "https://sqs.eu-west-1.amazonaws.com"; AWS HTTP error: Client error: `POST https://sqs.eu-west-1.amazonaws.com` resulted in a `403 Forbidden` response:               
  <?xml version="1.0"?><ErrorResponse xmlns="http://queue.amazonaws.com/doc/2012-11-05/"><Error><Type>Sender</Type><Code>A (truncated...)                                                                
   AccessDenied (client): Access to the resource https://sqs.eu-west-1.amazonaws.com/ is denied. - <?xml version="1.0"?><ErrorResponse xmlns="http://queue.amazonaws.com/doc/2012-11-05/"><Error><Type>  
  Sender</Type><Code>AccessDenied</Code><Message>Access to the resource https://sqs.eu-west-1.amazonaws.com/ is denied.</Message><Detail/></Error><RequestId>c5b0ce94-072d-5a5d-980f-9af4b19c28de</Requ  
  estId></ErrorResponse>                                                                                                                                                                                                                                                                                                                                                                                          

What is enqueue_dot_app_dot_default? Who knows! :D Never mentioned anywhere

So let's workaround by giving 'acme-testing' IAM user a 'AmazonSQSFullAccess' permission. This moves us past this roadblock but not much further

$ ./bin/console enqueue:consume --setup-broker -vvv 
[debug] Set context's logger "Symfony\Component\Console\Logger\ConsoleLogger"
[debug] [SqsDriver] Declare router queue: enqueue_dot_app_dot_default
[debug] [SqsDriver] Declare processor queue: enqueue_dot_app_dot_default
[debug] [SqsDriver] Declare processor queue: fos_elastica_populate
[debug] [SqsDriver] Declare processor queue: fos_elastica_doctrine_orm_sync_index_with_object_change
[info] Start consuming

What are fos_elastica_populate and fos_elastica_doctrine_orm_sync_index_with_object_change? Not clear where these are configured or how i can replace them with pre-created SQS queues

I've started using the Google Pub/Sub queue integration now, and it doesn't seem to work quite right.

It would seem that the message gets sent to a destination topic that is not subscribed to in the consumer. I added some extra logging below.

The weird thing is that there are times where it picks up the message, but a long time later (like 10-15mins). This seems to occur randomly. Most of the time I can start the consumer, and send a message with the producer and it will never arrive.

This is my config.yml file.

enqueue:
  transport:
    default:
      alias: gps
    redis: 
      host: redis
      vendor: predis
    gps:
      projectId: ...
      keyFilePath: ...
  client: 
    traceable_producer: true

And the produce/consume commands with extra logging I added.

$ ./bin/console enqueue:produce testtopic testmessage -vvv
02:36:35 DEBUG     [event] Notified event "console.command" to listener "Symfony\Component\HttpKernel\EventListener\DebugHandlersListener::configure". ["event" => "console.command","listener" => "Symfony\Component\HttpKernel\EventListener\DebugHandlersListener::configure"]
02:36:35 DEBUG     [event] Notified event "console.command" to listener "Symfony\Component\HttpKernel\EventListener\DumpListener::configure". ["event" => "console.command","listener" => "Symfony\Component\HttpKernel\EventListener\DumpListener::configure"]
02:36:35 DEBUG     [event] Notified event "console.command" to listener "Symfony\Bridge\Monolog\Handler\ConsoleHandler::onCommand". ["event" => "console.command","listener" => "Symfony\Bridge\Monolog\Handler\ConsoleHandler::onCommand"]
02:36:35 DEBUG     [event] Notified event "console.command" to listener "Symfony\Bridge\Monolog\Handler\ConsoleHandler::onCommand". ["event" => "console.command","listener" => "Symfony\Bridge\Monolog\Handler\ConsoleHandler::onCommand"]
/app/vendor/enqueue/gps/GpsProducer.php:37:
string(35) "Sending message to: enqueue.default" <-- value of `$destination->getTopicName()`
Message is sent

$ ./bin/console enqueue:consume --setup-broker -vvv
02:40:31 DEBUG     [event] Notified event "console.command" to listener "Symfony\Component\HttpKernel\EventListener\DebugHandlersListener::configure". ["event" => "console.command","listener" => "Symfony\Component\HttpKernel\EventListener\DebugHandlersListener::configure"]
02:40:31 DEBUG     [event] Notified event "console.command" to listener "Symfony\Component\HttpKernel\EventListener\DumpListener::configure". ["event" => "console.command","listener" => "Symfony\Component\HttpKernel\EventListener\DumpListener::configure"]
02:40:31 DEBUG     [event] Notified event "console.command" to listener "Symfony\Bridge\Monolog\Handler\ConsoleHandler::onCommand". ["event" => "console.command","listener" => "Symfony\Bridge\Monolog\Handler\ConsoleHandler::onCommand"]
02:40:31 DEBUG     [event] Notified event "console.command" to listener "Symfony\Bridge\Monolog\Handler\ConsoleHandler::onCommand". ["event" => "console.command","listener" => "Symfony\Bridge\Monolog\Handler\ConsoleHandler::onCommand"]
[debug] Set context's logger Symfony\Component\Console\Logger\ConsoleLogger
[debug] [GpsDriver] Subscribe router topic to queue: enqueue.default -> enqueue.app.default
[debug] [GpsDriver] Subscribe processor topic to queue: enqueue.app.default -> enqueue.app.default
[info] Start consuming
/app/vendor/google/cloud-pubsub/Subscription.php:407:
string(92) "Pulling from GPS subscription: projects/.../subscriptions/enqueue.app.default"
/app/vendor/google/cloud-pubsub/Subscription.php:407:
string(92) "Pulling from GPS subscription: projects/.../subscriptions/enqueue.app.default"
/app/vendor/google/cloud-pubsub/Subscription.php:407:
string(92) "Pulling from GPS subscription: projects/.../subscriptions/enqueue.app.default"

So it seems like it sends to enqueue.default and pulls from enqueue.app.default.

After I switched to using Redis instead of GPS, everything worked as expected. I started a consumer, send a message, and the consumer picked it up a split second later.

Updates #987.

So, I spent a bit of time on this Symfony 5 compatibility, and there was quite a lot to do actually:

• I had to upgrade PHPStan to 0.12 and PHPUnit to 7.5 so that it can work. Upgrading PHPUnit forced me to fix some deprecations like using non namespaced classes or setExpectedException.
• I got rid of the full dependency to https://github.com/voryx/Thruway, requiring only https://github.com/thruway/client instead. The client is enough for this lib, voryx/thruway is actually only needed for dev and testing, so I moved it to the thruway docker container.
• I had to create alternate versions of the classes directly extending or type-hinting the EventDispatcher and the Event class from Symfony. Symfony completely changed those between 4.3 and 5.0. They provide a proxy to help with migrating from 4.3+ to 5.0, but the proxy is not enough and does not exist for prior versions. So it’s a bit messy but basically, the old Event class is supported for Symfony < 4.3, the new Event class is supported for Symfony >= 5.0, and the versions in between support both classes, allowing people to migrate their code.
• Commands return values were sometimes null, which is not allowed anymore.
• replaced deprecated kernel.root_dir with kernel.project_dir.
• removed framework.templating option in config.yml, as it does not exist anymore.
• adapted the MessageQueueCollector, since the DataCollector interface has changed.
• adapted the RunCommandProcessor to use Process construction from array instead of string, which is deprecated.

Quorum queues are modern queues based on Raft consensus algorithm. It is available as of RabbitMQ 3.8.0. Classic mirrored queues are deprecated and will be removed in the future. This type of queues should be declared explicitly by setting argument x-queue-type to quorum (the default is classic)

I have already add a separate RabbitmqQuorumDriver in my project. And can make a PR to this repo, if maintainers are interested to support this feature.

see https://www.rabbitmq.com/quorum-queues.html#policy-support

Since Redis 6.0 released on 2020 a float value is allowed to define the timeout for brpop commands:

https://github.com/redis/redis/commit/eca0187370c14aa2c126fe07e5310e44c2780a95

https://redis.io/commands/brpop/

Starting with Redis version 6.0.0: timeout is interpreted as a double instead of an integer.

However, I can't see an easy way to extend the library to offer this support. In the past I've used Symfony decorators to extend the context to provide a different serializer for example, but since all the properties are private it is a little hard to override any internal implementation.

I ended up creating a custom package which I'm basically copying most of the existing code, but altering the parts needed to offer this support, specially both PhpRedis and PRedis clients to allow passing a float timeout on brpop().

Note: The predis library already offers support to pass a float timeout to the Redis engine, however phpredis still does not. I recently submitted an issue about it: https://github.com/phpredis/phpredis/issues/2157. As a workaround I am using rawCommand().

Here's the implementation that I have for receive():

public function receive(int $timeout = 0): ?Message
{
    $timeout = (float) ($timeout / 1000);

    if ($timeout <= 0) {
        while (true) {
            if ($message = $this->receive(600000)) {
                return $message;
            }
        }
    }

    return $this->receiveMessage([$this->queue], $timeout, $this->redeliveryDelay);
}

PhpRedis:

public function brpop(array $keys, float $timeout): ?RedisResult
{
    try {
        if ($result = $this->redis->rawCommand('brpop', \implode(' ', $keys), $timeout)) {
            return new RedisResult($result[0], $result[1]);
        }

        return null;
    } catch (\RedisException $e) {
        throw new ServerException('brpop command has failed', 0, $e);
    }
}

PRedis:

public function brpop(array $keys, float $timeout): ?RedisResult
{
    try {
        if ($result = $this->redis->brpop($keys, $timeout)) {
            return new RedisResult($result[0], $result[1]);
        }

        return null;
    } catch (PRedisServerException $e) {
        throw new ServerException('brpop command has failed', 0, $e);
    }
}

  1x: The "Symfony\Component\Console\Command\Command::$defaultName" property is considered final. You should not override it in "Enqueue\Symfony\Consumption\ConfigurableConsumeCommand".

  1x: The "Symfony\Component\Console\Command\Command::$defaultName" property is considered final. You should not override it in "Enqueue\Symfony\Client\ConsumeCommand".

  1x: The "Symfony\Component\Console\Command\Command::$defaultName" property is considered final. You should not override it in "Enqueue\Symfony\Client\ProduceCommand".

  1x: The "Symfony\Component\Console\Command\Command::$defaultName" property is considered final. You should not override it in "Enqueue\Symfony\Client\SetupBrokerCommand".

  1x: The "Symfony\Component\Console\Command\Command::$defaultName" property is considered final. You should not override it in "Enqueue\Symfony\Client\RoutesCommand".

  1x: Since symfony/console 6.1: Relying on the static property "$defaultName" for setting a command name is deprecated. Add the "Symfony\Component\Console\Attribute\AsCommand" attribute to the "Enqueue\Symfony\Consumption\ConfigurableConsumeCommand" class instead.

  1x: Since symfony/console 6.1: Relying on the static property "$defaultName" for setting a command name is deprecated. Add the "Symfony\Component\Console\Attribute\AsCommand" attribute to the "Enqueue\Symfony\Client\ConsumeCommand" class instead.

  1x: Since symfony/console 6.1: Relying on the static property "$defaultName" for setting a command name is deprecated. Add the "Symfony\Component\Console\Attribute\AsCommand" attribute to the "Enqueue\Symfony\Client\ProduceCommand" class instead.

  1x: Since symfony/console 6.1: Relying on the static property "$defaultName" for setting a command name is deprecated. Add the "Symfony\Component\Console\Attribute\AsCommand" attribute to the "Enqueue\Symfony\Client\SetupBrokerCommand" class instead.

  1x: Since symfony/console 6.1: Relying on the static property "$defaultName" for setting a command name is deprecated. Add the "Symfony\Component\Console\Attribute\AsCommand" attribute to the "Enqueue\Symfony\Client\RoutesCommand" class instead.

I have a process which is generating lots (measured in the 1000s) of events to be sent via enque. This obviusly causes a delay for any other events sent afterwards by default.

I have tried to set the priority of the bulk events to MessagePriority::VERY_LOW, but it has no effect - even when other events are excplicity set to a higher priority.

Digging through the enqueue code, I can see that all events are first sent to the router, via GenericDriver::sendToRouter(). This clones the event (via createTransportMessage()) but does not set the priority.

Once the router handes any events, they are pushed via GenericDriver::GenericDriver(). This also clones the event (via the same createTransportMessage()), but also includes a block of code to set the priority - aswell as delay and expiry.

Since eveything must go to the router first, and no prority is set in GenericDriver::sendToRouter(), my bulk events are always blocking any handling of higher priority events sent afterwards.

I am currently using the DBAL driver.