The PR adds provider\publisher that can work with any queue interop compatible transport.

TODO:

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

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

On #61, @odolbeau pointed that message properties were not properly set by the provider when using amqplib library.

It appeared that the existing code was just setting additional headers and skipping all other properties.

Rationale on implementation choices

One way to solve that could have been this elegant (based on my very own and objective opinion :p) one-liner:

$properties = union_merge($envelope->get_properties(), $envelope->delivery_info);

Problem then comes when a client tries to access a key that has no value, as it raises an exception.

The new implementation is less elegant but at least avoids this problem. I didn't delve too much into the C implementation for the pecl package, but does a function like $envelope->getContentType() return something or raises an exception when the content-type is not set at message publication?

The second set of properties, delivery_info is tested using isset rather than has because, according to the doc:

If you don't want to access the delivery_info array directly you can also use $msg->get('delivery_tag') but keep in mind that's slower than just accessing the array by key.

To maintain uniformity between providers, some property names have been converted to the names used by the PeclPackageMessageProvider. For completeness, here they are:

• is_redelivery (PeclPackageMessageProvider) / redelivered (PhpAmqpLibMessageProvider)
• exchange_name (PeclPackageMessageProvider) / exchange (PhpAmqpLibMessageProvider)
• headers (PeclPackageMessageProvider) / application_headers (PhpAmqpLibMessageProvider)

For the same reason, the following fields have been added to the PeclPackageMessageProvider:

• user_id
• cluster_id
• channel
• consumer_tag

Tests

No tests so far as this PR will probably change, but do you have preferences / suggestions on this?

Thought this was missing. This is still a WIP though, as I just want to gather some feedbacks first.

• [x] Add the RPC Server Processor
• [x] Add the RPC Client Processor
• [x] Add the README
• [x] Add some mofo tests

I'm not really sure on some points though (just trying to adapt from the official documentation and thought it could be useful here)

Context

In Go or Javascript I've notice that you can see how many consumers are connected, what are their names and with which arguments they have been popped

After searching a bit I understood that we can also achieve it in PHP and with swarrot, by using the consume method instead of the get method.

Proposition

So my suggestion would be to add the possibility to choose either the consume or the get method when consuming.

Attached code is just an ugly draft to test it. I'd like to know if someone see an issue with this idea before going further ?

| | Get | Consume | | --- | --- | --- | | Benefits | Non-Blocking | Long polling, Faster, show consumer in RMQ UI | | Drawbacks | Short polling, Treatments in SleepyInterface load the host | Blocking |

Exemple

Measures

Blackfire profiles

• Get : https://blackfire.io/profiles/d63a7eef-fe40-476b-8850-d190334b6362/graph
• Consume : https://blackfire.io/profiles/dd9bbb8f-f7e5-4aa7-83d9-bd036b74793d/graph

Consume seems slightly more performant (~10%) than get

vagrant@local:~/www/lafourchette-b2c-crm(master)$ time app/console swarrot:consume:customer_consumer --max-messages=1000 --method=get

real	0m7.405s
user	0m2.476s
sys	0m0.228s
vagrant@local:~/www/lafourchette-b2c-crm(master)$ time app/console swarrot:consume:customer_consumer --max-messages=1000 --method=consume

real	0m5.400s
user	0m1.952s
sys	0m0.172s
vagrant@local:~/www/lafourchette-b2c-crm(master)$ time app/console swarrot:consume:customer_consumer --max-messages=10000 --method=get

real	1m0.685s
user	0m20.997s
sys	0m1.448s
vagrant@local:~/www/lafourchette-b2c-crm(master)$ time app/console swarrot:consume:customer_consumer --max-messages=10000 --method=consume

real	0m53.869s
user	0m19.749s
sys	0m1.196s

In php 5.3 a lambda function is not aware of the context :

<?php
  error_reporting(E_ALL);
  class SignalHandlerProcessor
  {
      protected static $shouldExit = false;
      public function process() 
      {
          $greet = function () {
              SignalHandlerProcessor::$shouldExit = true;
          };
          $greet();
      }
}

$d = new SignalHandlerProcessor();
$d->process();

Will produce the error :

PHP Fatal error:  Cannot access protected property SignalHandlerProcessor::$shouldExit on line 17

I see only two fixes :

• update the compatibility to php 5.4 on the composer.json
• set shouldExit as public

| Q | A | ------------- | --- | Branch? | master | Bug fix? | no | New feature? | no | BC breaks? | no | Deprecations? | no | Tests pass? | yes

Currently we're restricted to using Swarrot\Broker\Message as our Message format. I would like to introduce Swarrot\Broker\MessageInterface to allow custom implementations of messages (and providers / publishers). Also added some basic functionality (setters, getProperty, setProperty) to the MessageInterface.

Updated docs (including some old references to MessageProviders).

Used https://github.com/queue-interop/queue-interop/blob/master/src/PsrMessage.php for inspiration.

I will share the provider I have written to use swarrot with amazon sqs.

This code is already in production on our platform.

As we are currently changing our queueing system, we didn't use a message publisher yet

Hi,

IMO it could be great to have such information in the log context What do you think? Also what do you think about rewriting all log message so they are invaraible, and moving variable(s) to the context?

Thank you

This is adds support for publisher confirms (see here) while retaining backwards compatibility with custom publishers.

If this is accepted I will also add support for publisher confirms in the symfony bundle.

In order to get rid of outdated dependencies (aws/aws-sdk-php & sentry/sentry) I'd like to deprecate some classes & remove them in 4.0.

@nicolasThal are you still using swarrot with the SQS message provider? @notFloran are you still using the SentryProcessor? @Taluu are you still using RPC processors?

Any concern @stof @lyrixx?

This plays better with logging systems: those will often group logs by message, showing how many time similar events happened.

Fixes #187

Not using placeholders for now because I don't know how widely used they are, tell me if you want me to.

I fixed 2 out 18 occurrences for now, but I like early feedback.

EDIT: 18/18

I can't use application's message identifier like this :

$publisher->publish(new Message("body", [], 34));
$message = $consumer->get();
$message->getId(); // 34

The Message identifier return the delivery_tag.

I can make a PR but I don't understand this behavior, so I'm asking first.

Hi,

I want to ask if there any reason other than removing the outdated dependencies, that the RPC processors were deprecated and removed?

I am thinking of using an RpcServer-like processor.

PS: I've seen the comments on this PR: https://github.com/swarrot/swarrot/pull/197/files

Thanks in advance, Val

In my current project, I want to retry the message for some errors (temporary failure of the SMTP server for instance), while some other errors should never be retried (malformed message which cannot be handled by my application at all for instance, and would just fail again when retrying). IT would be great to provide a way to achieve this (it might be done as a documentation page though if it requires too much custom logic to be added in existing processors)