diff --git a/DependencyInjection/BdfQueueMessengerExtension.php b/DependencyInjection/BdfQueueMessengerExtension.php index a059adb..a901591 100644 --- a/DependencyInjection/BdfQueueMessengerExtension.php +++ b/DependencyInjection/BdfQueueMessengerExtension.php @@ -12,7 +12,7 @@ */ class BdfQueueMessengerExtension extends Extension { - public function load(array $configs, ContainerBuilder $container) + public function load(array $configs, ContainerBuilder $container): void { $loader = new YamlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config')); $loader->load('queue_messenger.yaml'); diff --git a/Tests/Functional/BundleIntegrationTest.php b/Tests/Functional/BundleIntegrationTest.php new file mode 100644 index 0000000..32fe371 --- /dev/null +++ b/Tests/Functional/BundleIntegrationTest.php @@ -0,0 +1,149 @@ +get('bdf_queue.messenger_transport.factory'); + + $this->assertInstanceOf(QueueTransportFactory::class, $factory); + } + + public function testDefaultStampSerializerIsThePhpOne() + { + self::bootKernel(); + + $serializer = self::getContainer()->get('bdf_queue.messenger_transport.stamp_serializer'); + + $this->assertInstanceOf(PhpStampsSerializer::class, $serializer); + } + + public function testMessageBusHandlerIsRegistered() + { + self::bootKernel(); + + $handler = self::getContainer()->get(MessageBusHandler::class); + + $this->assertInstanceOf(MessageBusHandler::class, $handler); + } + + public function testMessengerBuildsAQueueTransportForBdfQueueDsn() + { + self::bootKernel(); + + // The transport is built by symfony messenger using the tagged factory + // of the bundle : getting a QueueTransport proves the factory is + // properly tagged as "messenger.transport_factory". + $transport = self::getContainer()->get('messenger.transport.bdf'); + + $this->assertInstanceOf(QueueTransport::class, $transport); + } + + public function testSendAndReceiveRoundTrip() + { + self::bootKernel(); + + /** @var QueueTransport $transport */ + $transport = self::getContainer()->get('messenger.transport.bdf'); + + $message = new TestMessage('hello'); + $transport->send(new Envelope($message)); + + $received = iterator_to_array($transport->get()); + + $this->assertCount(1, $received); + + /** @var Envelope $envelope */ + $envelope = $received[0]; + $this->assertEquals($message, $envelope->getMessage()); + $this->assertNotNull($envelope->last(BdfQueueReceivedStamp::class)); + + // Acknowledging must not throw and should leave the queue empty. + $transport->ack($envelope); + + $this->assertSame([], iterator_to_array($transport->get())); + } + + public function testDispatchThroughMessengerBusRoutesToTransportAndIsHandled() + { + self::bootKernel(); + + $container = self::getContainer(); + + /** @var MessageBusInterface $bus */ + $bus = $container->get('messenger.default_bus'); + /** @var TestMessageHandler $handler */ + $handler = $container->get(TestMessageHandler::class); + + $message = new TestMessage('via-bus'); + + // Dispatch through the bus : the routing sends the message to the bdf + // transport asynchronously, so it must NOT be handled in process yet. + $envelope = $bus->dispatch($message); + + $this->assertNotNull($envelope->last(SentStamp::class), 'The message should have been sent to a transport'); + $this->assertSame([], $handler->handled, 'An async message must not be handled during dispatch'); + + // Consume the transport with a real worker : it receives the message + // from the queue and dispatches it back to the bus for handling. + $worker = new Worker( + ['bdf' => $container->get('messenger.transport.bdf')], + $bus, + $this->workerEventDispatcher() + ); + $worker->run(['sleep' => 0]); + + $this->assertCount(1, $handler->handled); + $this->assertEquals($message, $handler->handled[0]); + } + + /** + * Event dispatcher stopping the worker as soon as one message is handled, + * with a wall-clock deadline as a safety net so the test can never hang. + */ + private function workerEventDispatcher(): EventDispatcher + { + $dispatcher = new EventDispatcher(); + $dispatcher->addSubscriber(new StopWorkerOnMessageLimitListener(1)); + + $deadline = microtime(true) + 5; + $dispatcher->addListener(WorkerRunningEvent::class, function (WorkerRunningEvent $event) use ($deadline) { + if (microtime(true) >= $deadline) { + $event->getWorker()->stop(); + } + }); + + return $dispatcher; + } +} diff --git a/Tests/Functional/Fixtures/DestinationManagerFactory.php b/Tests/Functional/Fixtures/DestinationManagerFactory.php new file mode 100644 index 0000000..b31873b --- /dev/null +++ b/Tests/Functional/Fixtures/DestinationManagerFactory.php @@ -0,0 +1,40 @@ + 'memory://localhost', + ]); + $connectionFactory->addDriverResolver('memory', function () { + return new MemoryConnection(); + }); + + // Cache the connections so a message sent by the transport and the + // consumer that reads it back share the same in-memory storage. + $connectionFactory = new CachedConnectionDriverFactory($connectionFactory); + + return new DestinationManager( + $connectionFactory, + new ConfigurationDestinationFactory([], new DsnDestinationFactory($connectionFactory)) + ); + } +} diff --git a/Tests/Functional/Fixtures/PublicServicesPass.php b/Tests/Functional/Fixtures/PublicServicesPass.php new file mode 100644 index 0000000..1d627f1 --- /dev/null +++ b/Tests/Functional/Fixtures/PublicServicesPass.php @@ -0,0 +1,37 @@ +ids = $ids; + } + + public function process(ContainerBuilder $container): void + { + foreach ($this->ids as $id) { + if ($container->hasDefinition($id)) { + $container->getDefinition($id)->setPublic(true); + } elseif ($container->hasAlias($id)) { + $container->getAlias($id)->setPublic(true); + } + } + } +} diff --git a/Tests/Functional/Fixtures/TestKernel.php b/Tests/Functional/Fixtures/TestKernel.php new file mode 100644 index 0000000..45c8b60 --- /dev/null +++ b/Tests/Functional/Fixtures/TestKernel.php @@ -0,0 +1,98 @@ +addCompilerPass( + new PublicServicesPass([ + MessageBusHandler::class, + 'bdf_queue.messenger_bus_handler', + 'messenger.default_bus', + ]), + PassConfig::TYPE_BEFORE_REMOVING + ); + } + + public function registerContainerConfiguration(LoaderInterface $loader): void + { + $loader->load(function (ContainerBuilder $container) { + $config = [ + 'test' => true, + 'secret' => 'test', + 'http_method_override' => false, + 'php_errors' => ['log' => true], + 'messenger' => [ + 'transports' => [ + 'bdf' => 'bdfqueue://memory-queue', + ], + 'routing' => [ + TestMessage::class => ['bdf'], + ], + ], + ]; + + // Introduced in Symfony 6.4, unknown (and thus rejected) on 5.4. + if (Kernel::VERSION_ID >= 60400) { + $config['handle_all_throwables'] = true; + } + + $container->loadFromExtension('framework', $config); + + // Service usually provided by the bdf-queue framework bundle. + $container + ->register('bdf_queue.destination_manager', DestinationManager::class) + ->setFactory([DestinationManagerFactory::class, 'create']) + ->setPublic(true); + + // Handler used to assert the dispatch -> transport -> worker flow. + $container + ->register(TestMessageHandler::class, TestMessageHandler::class) + ->addTag('messenger.message_handler') + ->setPublic(true); + }); + } + + public function getCacheDir(): string + { + return $this->buildDir().'/cache'; + } + + public function getBuildDir(): string + { + return $this->buildDir().'/build'; + } + + public function getLogDir(): string + { + return $this->buildDir().'/log'; + } + + private function buildDir(): string + { + return sys_get_temp_dir().'/bdf_queue_messenger_bundle_tests/'.$this->environment; + } +} diff --git a/Tests/Functional/Fixtures/TestMessage.php b/Tests/Functional/Fixtures/TestMessage.php new file mode 100644 index 0000000..f7191a6 --- /dev/null +++ b/Tests/Functional/Fixtures/TestMessage.php @@ -0,0 +1,19 @@ +content = $content; + } +} diff --git a/Tests/Functional/Fixtures/TestMessageHandler.php b/Tests/Functional/Fixtures/TestMessageHandler.php new file mode 100644 index 0000000..112bf38 --- /dev/null +++ b/Tests/Functional/Fixtures/TestMessageHandler.php @@ -0,0 +1,20 @@ + transport -> worker -> handler round trip. + */ +class TestMessageHandler +{ + /** + * @var TestMessage[] + */ + public $handled = []; + + public function __invoke(TestMessage $message): void + { + $this->handled[] = $message; + } +}