README

Symfony 插件，用于简化异步消息的路由。

为什么需要这个插件？

它允许您：

自动创建 MessageProducer 服务。
在 yaml 或 xml 配置中为每个服务定义路由。
为每个服务定义 AsynchronousMessageProducerBridge，您可以使用任何基础设施来实现。

快速开始

1) 通过 composer 安装此库

composer require tomcizek/symfony-prooph-asynchronous-router

2) 在您的内核中注册这些插件（独立，但可能在 config/bundles.php 中）

return [
    Prooph\Bundle\ServiceBus\ProophServiceBusBundle::class => ['all' => true],
    TomCizek\AsynchronousRouter\ProophAsynchronousRouterBundle::class => ['all' => true],
];

3) 实现您自己的 TomCizek\AsynchronousRouter\AsynchronousMessageProducerBridge

RabbitMq 的最简单工作示例实现可能如下所示（基于 php-amqplib 基础设施） php-amqplib

<?php declare(strict_types = 1);

namespace YourApp\Infrastructure\Rabbit;

use DateTime;
use OldSound\RabbitMqBundle\RabbitMq\Producer;
use Prooph\Common\Messaging\Message;
use Prooph\Common\Messaging\MessageConverter;
use Prooph\Common\Messaging\MessageDataAssertion;
use TomCizek\AsynchronousRouter\AsynchronousMessageProducerBridge;

final class RabbitAsynchronousMessageProducerBridge implements AsynchronousMessageProducerBridge
{
	/** @var Producer */
	private $producer;

	/** @var MessageConverter */
	private $messageConverter;

	public function __construct(Producer $producer, MessageConverter $messageConverter)
	{
		$this->producer = $producer;
		$this->messageConverter = $messageConverter;
	}

	public function publishWithRoutingKey(Message $message, string $routingKey): void
	{
		$stringMessage = $this->convertMessageToString($message);
		$this->producer->publish($stringMessage, $routingKey);
	}

	private function convertMessageToString(Message $message): string
	{
		$messageData = $this->messageConverter->convertToArray($message);
		MessageDataAssertion::assert($messageData);
		$messageData['created_at'] = $message->createdAt()->format(DateTime::ATOM);

		return json_encode($messageData);
	}
}

4) 在您的 symfony *.yml 配置中设置 prooph 组件的配置！

prooph_asynchronous_router:
    producers:
        # this key defines service id of created producer, this will create 'prooph_asynchronous_router.firstProducer'
        firstProducer:
            # bridge refers to service with your implementation of AsynchronousMessageProducerBridge
            bridge: 'firstAsynchronousMessageProducerBridge'
            # Routes are defined that key is message name and value is routing key. 
            # AsynchronousMessageProducerBridge will then recieve message instance and routing key string 
            # in publishWithRoutingKey method. 
            routes:
                App\Namespace\FirstMessage: first_routing_key

# this is standard service-bus-symfony-bundle configuration
prooph_service_bus:
    event_buses:
        first_event_bus:
            router:
                # here you refer to automatically created service you defined few lines above.
                async_switch: 'prooph_asynchronous_router.firstProducer'
services:
    # here you register your implementation of AsynchronousMessageProducerBridge to which we refer above.
    firstAsynchronousMessageProducerBridge:
        class: TomCizek\AsynchronousRouter\Tests\DependencyInjection\Fixture\Model\FirstAsynchronousMessageProducerBridge

另一个配置示例

贡献

请随意分支并扩展现有功能或添加新功能，并发送带有您更改的 pull request！为了建立一致的代码质量，请为所有更改提供单元测试，并可能更新文档。

tomcizek / symfony-prooph-asynchronous-router

维护者

详情