objectManager->get(MessageBusInterface::class); $messageBus->dispatch(new TestMessage('Hallo Welt!')); $sendersContainer = $this->objectManager->get(TransportsContainer::class); /** @var InMemoryTransport $transport1 */ $transport1 = $sendersContainer->get('test-in-memory-1'); /** @var InMemoryTransport $transport2 */ $transport2 = $sendersContainer->get('test-in-memory-2'); /** @var DoctrineTransport $transport3 */ $transport3 = $sendersContainer->get('test-doctrine'); $this->assertInstanceOf(InMemoryTransport::class, $transport1); $this->assertInstanceOf(InMemoryTransport::class, $transport2); $this->assertInstanceOf(DoctrineTransport::class, $transport3); $this->assertCount(1, $transport1->getSent()); $this->assertCount(1, $transport2->getSent()); $this->assertCount(1, $transport3->all()); $this->assertCount(0, $transport1->getAcknowledged()); $this->assertCount(0, $transport2->getAcknowledged()); $this->assertCount(1, $transport3->all()); $eventDispatcher = $this->objectManager->get('DigiComp.FlowSymfonyBridge.Messenger:EventDispatcher'); $eventDispatcher->addSubscriber(new StopWorkerOnMessageLimitListener(1)); foreach (['test-in-memory-1', 'test-in-memory-2', 'test-doctrine'] as $transportId) { $worker = new Worker([$transportId => $sendersContainer->get($transportId)], $messageBus, $eventDispatcher); $worker->run(); } // TODO: Check for success on all workers - doctrine does not seem to get executed $this->assertCount(1, $transport1->getAcknowledged()); $this->assertCount(1, $transport2->getAcknowledged()); $this->assertCount(0, $transport3->all()); } /** * @test */ public function itRetriesFailingMessages() { $messageBus = $this->objectManager->get(MessageBusInterface::class); $messageBus->dispatch(new FailingMessage()); $sendersContainer = $this->objectManager->get(TransportsContainer::class); /** @var DoctrineTransport $transport1 */ $transport1 = $sendersContainer->get('test-retry-doctrine'); /** @var DoctrineTransport $failedTransport */ $failedTransport = $sendersContainer->get('test-failed-doctrine'); $this->assertCount(1, $transport1->all()); $this->assertCount(0, $failedTransport->all()); /** @var EventDispatcherInterface $eventDispatcher */ $eventDispatcher = $this->objectManager->get('DigiComp.FlowSymfonyBridge.Messenger:EventDispatcher'); $eventDispatcher->addSubscriber(new StopWorkerOnMessageLimitListener(1)); $worker = new Worker( ['test-retry-doctrine' => $sendersContainer->get('test-retry-doctrine')], $messageBus, $eventDispatcher ); $worker->run(); $this->assertCount(1, $transport1->all()); $this->assertCount(0, $failedTransport->all()); $worker = new Worker( ['test-retry-doctrine' => $sendersContainer->get('test-retry-doctrine')], $messageBus, $eventDispatcher ); $worker->run(); $this->assertCount(0, $transport1->all()); $this->assertCount(1, $failedTransport->all()); } }