[FrameworkBundle] Added kafka transport to services

symfony · andythorne · Jul 21, 2023 · Jul 21, 2023 · Jul 21, 2023 · Jul 21, 2023
commit a44b50f37205e001243920b1083e66b9e09495d2
diff --git a/src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php b/src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php
@@ -2151,6 +2151,10 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
             $container->getDefinition('messenger.transport.beanstalkd.factory')->addTag('messenger.transport_factory');
         }
 
+        if (ContainerBuilder::willBeAvailable('symfony/kafka-messenger', MessengerBridge\Kafka\Transport\KafkaTransportFactory::class, ['symfony/framework-bundle', 'symfony/messenger'])) {
+            $container->getDefinition('messenger.transport.kafka.factory')->addTag('messenger.transport_factory');
+        }
+
         if (!class_exists(StopWorkerOnSignalsListener::class)) {
             $container->removeDefinition('messenger.listener.stop_worker_signals_listener');
         } elseif ($config['stop_worker_on_signals']) {
@@ -2213,6 +2217,7 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
             $container->removeDefinition('messenger.transport.redis.factory');
             $container->removeDefinition('messenger.transport.sqs.factory');
             $container->removeDefinition('messenger.transport.beanstalkd.factory');
+            $container->removeDefinition('messenger.transport.kafka.factory');
             $container->removeAlias(SerializerInterface::class);
         } else {
             $container->getDefinition('messenger.transport.symfony_serializer')

diff --git a/src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.php b/src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.php
@@ -147,6 +147,13 @@
 
         ->set('messenger.transport.beanstalkd.factory', BeanstalkdTransportFactory::class)
 
+        ->set('messenger.transport.kafka.factory', KafkaTransportFactory::class)
+            ->args([
+                service('logger')->ignoreOnInvalid(),
+                service(KafkaFactory::class)->ignoreOnInvalid(),
+            ])
+            ->tag('monolog.logger', ['channel' => 'messenger'])
+
         // retry
         ->set('messenger.retry_strategy_locator', ServiceLocator::class)
             ->args([

diff --git a/src/Symfony/Bundle/FrameworkBundle/Tests/DependencyInjection/FrameworkExtensionTestCase.php b/src/Symfony/Bundle/FrameworkBundle/Tests/DependencyInjection/FrameworkExtensionTestCase.php
@@ -59,6 +59,7 @@
 use Symfony\Component\Messenger\Bridge\AmazonSqs\Transport\AmazonSqsTransportFactory;
 use Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpTransportFactory;
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdTransportFactory;
+use Symfony\Component\Messenger\Bridge\Kafka\Transport\KafkaTransportFactory;
 use Symfony\Component\Messenger\Bridge\Redis\Transport\RedisTransportFactory;
 use Symfony\Component\Messenger\Transport\TransportFactory;
 use Symfony\Component\Notifier\ChatterInterface;
@@ -841,6 +842,10 @@ public function testMessenger()
             $expectedFactories[] = 'messenger.transport.beanstalkd.factory';
         }
 
+        if (class_exists(KafkaTransportFactory::class)) {
+            $expectedFactories[] = 'messenger.transport.kafka.factory';
+        }
+
         $this->assertTrue($container->hasDefinition('messenger.receiver_locator'));
         $this->assertTrue($container->hasDefinition('console.command.messenger_consume_messages'));
         $this->assertTrue($container->hasAlias('messenger.default_bus'));

diff --git a/src/Symfony/Component/Messenger/Bridge/Kafka/Tests/Transport/KafkaTransportFactoryTest.php b/src/Symfony/Component/Messenger/Bridge/Kafka/Tests/Transport/KafkaTransportFactoryTest.php
@@ -51,8 +51,6 @@ public function testSupports(): void
     {
         self::assertTrue($this->factory->supports('kafka://', []));
         self::assertTrue($this->factory->supports('kafka://localhost:9092', []));
-        self::assertTrue($this->factory->supports('kafka+ssl://', []));
-        self::assertTrue($this->factory->supports('kafka+ssl://localhost:9092', []));
         self::assertFalse($this->factory->supports('plaintext://localhost:9092', []));
         self::assertFalse($this->factory->supports('kafka', []));
     }