使用 Messenger 阅读未通过 Messenger 发送的排队消息

Question

我正在尝试阅读一条 queued 消息（在 RabbitMQ 中），该消息不是使用 Symfony Messenger 发送的。 Messenger 似乎添加了一些 header，例如

headers: 
    type: App\Message\Transaction

但是读取外部消息时，这个header不存在

那么，有没有办法告诉 Messenger queue A 中的每条消息都必须被视为消息类型 Transaction？

我今天的内容是：

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

我想补充的是：

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

Answer 1

Ryan Weaver 在 symfony 的 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

我根据自己的需要做了一个快速实现，由你来适应你的业务逻辑 :

1 - 创建一个 Serializer 来实现 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 - 使用工厂检索正确的 Message :

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) 为 Messenger 配置新的自定义序列化程序：

framework:
  messenger:
    serializer: 'app.my_custom_serializer'

我会尝试更进一步，找到直接 "connect" 队列的方法，会让您知道。

使用 Messenger 阅读未通过 Messenger 发送的排队消息

Use Messenger to read queued message not sent with Messenger

rabbitmq

symfony

symfony-messenger