Use Messenger to read queued message not sent with Messenger

Question

I'm trying to read a queued message (in RabbitMQ) that wasn't send with Symfony Messenger. It seems that Messenger adds some headers, like

headers: 
    type: App\Message\Transaction

but when reading external messages, this header does not exist.

So, is there a way to tell Messenger that every message in queue A must be consider as a message type Transaction ?

What I have today is :

framework:
    messenger:
        transports:
            # Uncomment the following line to enable a transport named "amqp"
            amqp:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    exchange:
                        name: messages
                        type: direct
                    queue:
                        name: queue_messages

        routing:
            # Route your messages to the transports
             'App\Message\Transaction': amqp

and what I would like to add is something like:

        routing:
            # Route your messages to the transports
             amqp: 'App\Message\Transaction'

That's not really a solution, but if you have control on message creation, you can had the header `type` with FQCN as value. I would appreciate a better solution like describe in your post too. — Erwan Haquet, Apr 09 '19 at 07:23
I wasn't clear enough. The thing is, I don't have control on the message creation, I only know what will be send in this queue. — Kevin Pires, Apr 09 '19 at 07:36
Agree, it's not clear and actually not a viable solution. Will ask on symfony slack if such a configuration or a factory can be implemented, i come back to you. — Erwan Haquet, Apr 09 '19 at 10:02

Erwan Haquet · Accepted Answer · 2019-04-10T07:56:40.643

Ryan Weaver replied to a similar question on symfony's slack:

You will need a custom serializer for messenger if the messages do not originate from messenger :)

1) You create a custom serialize (implements SerializerInterface from Messenger) and configure it under the messenger config

2) Somehow in that serializer, you take JSON and turn it into some "message" object you have in your code. How you do that is up to you - you need to somehow be able to look at your JSON and figure out which message class it should be mapped to. You could then create that object manually and populate the data, or use Symfony's serializer. Wrap this in an Envelope before returning it

3) Because your serializer is now returning a "message" object if some sort, Messenger uses its normal logic to find the handler(s) for that Message and execute them

I did a quick implementation for my own needs, up to you to fit with your business logic :

1 - Create a Serializer wich implement the SerializerInterface :


   // I keeped the default serializer, and just override his decode method.

   /**
     * {@inheritdoc}
     */
    public function decode(array $encodedEnvelope): Envelope
    {
        if (empty($encodedEnvelope['body']) || empty($encodedEnvelope['headers'])) {
            throw new InvalidArgumentException('Encoded envelope should have at least a "body" and some "headers".');
        }

        if (empty($encodedEnvelope['headers']['action'])) {
            throw new InvalidArgumentException('Encoded envelope does not have an "action" header.');
        }

        // Call a factory to return the Message Class associate with the action
        if (!$messageClass = $this->messageFactory->getMessageClass($encodedEnvelope['headers']['action'])) {
            throw new InvalidArgumentException(sprintf('"%s" is not a valid action.', $encodedEnvelope['headers']['action']));
        }

        // ... keep the default Serializer logic

        return new Envelope($message, ...$stamps);
    }

2 - Retrieve the right Message using a factory :

class MessageFactory
{
    /**
     * @param string $action
     * @return string|null
     */
    public function getMessageClass(string $action)
    {
        switch($action){
            case ActionConstants::POST_MESSAGE :
                return PostMessage::class ;
            default:
                return null;
        }
    }
}

3) Configure your new custom serializer for messenger :

framework:
  messenger:
    serializer: 'app.my_custom_serializer'

I'll try to go a bit further and find a way to "connect" a queue directly, will let you know.

Use Messenger to read queued message not sent with Messenger

1 Answers1