diff --git a/packages/Amqp/tests/Integration/AmqpStreamChannelTest.php b/packages/Amqp/tests/Integration/AmqpStreamChannelTest.php index ec155e4f7..374a860c2 100644 --- a/packages/Amqp/tests/Integration/AmqpStreamChannelTest.php +++ b/packages/Amqp/tests/Integration/AmqpStreamChannelTest.php @@ -1125,7 +1125,7 @@ public function test_commit_interval_working_correctly_with_message_limit(): voi $consumerId = $channelName; // Run consumer with message limit - should commit after each message - $ecotoneLite->run($channelName, ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 5)); + $ecotoneLite->run($channelName, ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 5, maxExecutionTimeInMilliseconds: 5000)); // Verify all messages were consumed $orders = $ecotoneLite->getQueryBus()->sendWithRouting('order.getOrders'); diff --git a/packages/Dbal/tests/Integration/ORMTest.php b/packages/Dbal/tests/Integration/ORMTest.php index 5e4dfda5a..100de228d 100644 --- a/packages/Dbal/tests/Integration/ORMTest.php +++ b/packages/Dbal/tests/Integration/ORMTest.php @@ -330,7 +330,7 @@ public function test_throwing_exception_when_setting_up_doctrine_orm_using_non_o $this->expectException(InvalidArgumentException::class); $ecotoneLite->sendCommandWithRoutingKey('multipleInternalCommands', [['personId' => 99, 'personName' => 'Johny', 'exception' => false]]); - $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 2, failAtError: true)); + $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 2, maxExecutionTimeInMilliseconds: 5000, failAtError: true)); } private function bootstrapEcotone(array $namespaces = ['Test\Ecotone\Dbal\Fixture\ORM\Person']): FlowTestSupport diff --git a/packages/Dbal/tests/Integration/Transaction/DbalTransactionAsynchronousEndpointTest.php b/packages/Dbal/tests/Integration/Transaction/DbalTransactionAsynchronousEndpointTest.php index dca5bae79..30786e223 100644 --- a/packages/Dbal/tests/Integration/Transaction/DbalTransactionAsynchronousEndpointTest.php +++ b/packages/Dbal/tests/Integration/Transaction/DbalTransactionAsynchronousEndpointTest.php @@ -65,7 +65,7 @@ public function test_turning_on_transactions_for_polling_consumer() ['personId' => 101, 'personName' => 'Johny', 'exception' => true], ]); - $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 2, failAtError: false)); + $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 2, maxExecutionTimeInMilliseconds: 5000, failAtError: false)); /** Should be rolled back */ $aggregateCommitted = true; @@ -120,7 +120,7 @@ public function test_reconnecting_on_lost_connection_during_commit() // Run with enough message handling attempts to process both commands // The second command will encounter a connection failure but should recover - $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 5, failAtError: false)); + $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 5, maxExecutionTimeInMilliseconds: 10000, failAtError: false)); // Verify that despite the connection failure, at least one aggregate was successfully created try { @@ -233,7 +233,7 @@ public function test_turning_on_transactions_for_polling_consumer_with_tenant_co ['personId' => 100, 'personName' => 'Johny', 'exception' => false], ['personId' => 101, 'personName' => 'Johny', 'exception' => true], ], metadata: ['tenant' => 'tenant_a']); - $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 2, failAtError: false)); + $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 2, maxExecutionTimeInMilliseconds: 5000, failAtError: false)); /** Should be rolled back */ $aggregateCommitted = true; @@ -257,7 +257,7 @@ public function test_turning_on_transactions_for_polling_consumer_with_tenant_co ['personId' => 100, 'personName' => 'Johny', 'exception' => false], ['personId' => 101, 'personName' => 'Johny', 'exception' => false], ], metadata: ['tenant' => 'tenant_a']); - $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 2, failAtError: false)); + $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 2, maxExecutionTimeInMilliseconds: 5000, failAtError: false)); $this->assertNotNull($ecotoneLite->sendQueryWithRouting('person.getName', metadata: ['aggregate.id' => 100, 'tenant' => 'tenant_a'])); $this->assertNotNull($ecotoneLite->sendQueryWithRouting('person.getName', metadata: ['aggregate.id' => 101, 'tenant' => 'tenant_a'])); @@ -288,7 +288,7 @@ public function test_turning_on_transactions_for_polling_consumer_with_document_ ['personId' => 101, 'personName' => 'Johny', 'exception' => true], ]); - $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 2, failAtError: false)); + $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 2, maxExecutionTimeInMilliseconds: 5000, failAtError: false)); /** Should be rolled back */ $aggregateCommitted = true; @@ -346,7 +346,7 @@ public function test_turning_on_transactions_for_polling_consumer_with_multiple_ ['personId' => 101, 'personName' => 'Johny', 'exception' => true], ], metadata: ['tenant' => 'tenant_a']); - $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 3, failAtError: false)); + $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 3, maxExecutionTimeInMilliseconds: 10000, failAtError: false)); /** Not created yet, as processed first two messages */ $aggregateCommitted = true; @@ -370,7 +370,7 @@ public function test_turning_on_transactions_for_polling_consumer_with_multiple_ ['personId' => 100, 'personName' => 'Johny', 'exception' => false], ['personId' => 101, 'personName' => 'Johny', 'exception' => false], ], metadata: ['tenant' => 'tenant_b']); - $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 4, failAtError: false)); + $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 4, maxExecutionTimeInMilliseconds: 10000, failAtError: false)); /** Saved in tenant b */ $this->assertNotNull($ecotoneLite->sendQueryWithRouting('person.getName', metadata: ['aggregate.id' => 100, 'tenant' => 'tenant_b'])); diff --git a/packages/Kafka/src/Configuration/KafkaAdmin.php b/packages/Kafka/src/Configuration/KafkaAdmin.php index 427379210..fb453d87c 100644 --- a/packages/Kafka/src/Configuration/KafkaAdmin.php +++ b/packages/Kafka/src/Configuration/KafkaAdmin.php @@ -7,6 +7,7 @@ use Ecotone\Kafka\Attribute\KafkaConsumer as KafkaConsumerAttribute; use Ecotone\Kafka\Outbound\KafkaDeliveryTracker; use Ecotone\Messaging\Config\ConfigurationException; +use Ecotone\Messaging\Handler\ExpressionEvaluationService; use Ecotone\Messaging\Handler\Logger\LoggingGateway; use Exception; use RdKafka\KafkaConsumer; @@ -50,6 +51,7 @@ public function __construct( private array $kafkaBrokerConfigurations, private array $topicReferenceMapping, private LoggingGateway $loggingGateway, + private ExpressionEvaluationService $expressionEvaluationService, ) { } @@ -89,7 +91,7 @@ public function getConsumer(string $endpointId, string $channelName): KafkaConsu $this->setLoggerCallbacks($conf, $endpointId); $consumer = new KafkaConsumer($conf); - $topics = $this->getMappedTopicNames($kafkaConsumerConfig->getTopics()); + $topics = $this->getMappedTopicNames($this->resolveExpressions($kafkaConsumerConfig->getTopics())); $consumer->subscribe($topics); $this->initializedConsumers[$endpointId] = $consumer; @@ -158,6 +160,25 @@ public function getTopicForProducer(string $referenceName): ProducerTopic ); } + /** + * @param string[] $topics + * @return string[] + */ + private function resolveExpressions(array $topics): array + { + $resolved = []; + foreach ($topics as $topic) { + if (str_contains($topic, '(')) { + $result = $this->expressionEvaluationService->evaluateWithContext($topic, []); + $resolved = array_merge($resolved, is_array($result) ? $result : [$result]); + } else { + $resolved[] = $topic; + } + } + + return $resolved; + } + private function getMappedTopicNames(string|array $topicName): string|array { if (is_array($topicName)) { diff --git a/packages/Kafka/src/Configuration/KafkaModule.php b/packages/Kafka/src/Configuration/KafkaModule.php index 4cc9b884c..71ca6d04f 100644 --- a/packages/Kafka/src/Configuration/KafkaModule.php +++ b/packages/Kafka/src/Configuration/KafkaModule.php @@ -27,6 +27,7 @@ use Ecotone\Messaging\Handler\Gateway\ParameterToMessageConverter\GatewayHeadersBuilder; use Ecotone\Messaging\Handler\Gateway\ParameterToMessageConverter\GatewayHeaderValueBuilder; use Ecotone\Messaging\Handler\Gateway\ParameterToMessageConverter\GatewayPayloadBuilder; +use Ecotone\Messaging\Handler\ExpressionEvaluationService; use Ecotone\Messaging\Handler\InterfaceToCallRegistry; use Ecotone\Messaging\Handler\Logger\LoggingGateway; use Ecotone\Messaging\MessageHeaders; @@ -164,6 +165,7 @@ public function prepare(Configuration $messagingConfiguration, array $extensionO $kafkaBrokerConfigurations, $topicReferenceMapping, Reference::to(LoggingGateway::class), + Reference::to(ExpressionEvaluationService::REFERENCE), ]) ); } diff --git a/packages/Kafka/tests/Attribute/KafkaConsumerUnsupportedTopicsSyntaxTest.php b/packages/Kafka/tests/Attribute/KafkaConsumerUnsupportedTopicsSyntaxTest.php new file mode 100644 index 000000000..0e8db5282 --- /dev/null +++ b/packages/Kafka/tests/Attribute/KafkaConsumerUnsupportedTopicsSyntaxTest.php @@ -0,0 +1,41 @@ +expectException(TypeError::class); + + new KafkaConsumer( + endpointId: 'orders', + topics: new Expression("reference('config').getOrderTopics()"), + ); + } + + public function test_a_plain_percent_placeholder_is_used_literally_not_resolved(): void + { + $consumer = new KafkaConsumer( + endpointId: 'orders', + topics: '%app.multi_tenancy.kafka.orders.topics%', + ); + + $this->assertSame( + ['%app.multi_tenancy.kafka.orders.topics%'], + $consumer->getTopics(), + "Dynamic topics use an expression string containing '(' (e.g. parameter('...') or reference('...')->method()), not Symfony's %...% container-parameter placeholder syntax.", + ); + } +} diff --git a/packages/Kafka/tests/Fixture/DynamicTopics/DynamicTopicsKafkaConsumer.php b/packages/Kafka/tests/Fixture/DynamicTopics/DynamicTopicsKafkaConsumer.php new file mode 100644 index 000000000..ee7206bf6 --- /dev/null +++ b/packages/Kafka/tests/Fixture/DynamicTopics/DynamicTopicsKafkaConsumer.php @@ -0,0 +1,34 @@ +messages[] = $payload; + } + + /** + * @return string[] + */ + #[QueryHandler('dynamicTopicsConsumer.getMessages')] + public function getMessages(): array + { + return $this->messages; + } +} diff --git a/packages/Kafka/tests/Fixture/DynamicTopics/MixedTopicsKafkaConsumer.php b/packages/Kafka/tests/Fixture/DynamicTopics/MixedTopicsKafkaConsumer.php new file mode 100644 index 000000000..c3a083218 --- /dev/null +++ b/packages/Kafka/tests/Fixture/DynamicTopics/MixedTopicsKafkaConsumer.php @@ -0,0 +1,34 @@ +messages[] = $payload; + } + + /** + * @return string[] + */ + #[QueryHandler('mixedTopicsConsumer.getMessages')] + public function getMessages(): array + { + return $this->messages; + } +} diff --git a/packages/Kafka/tests/Fixture/DynamicTopics/ParameterTopicsKafkaConsumer.php b/packages/Kafka/tests/Fixture/DynamicTopics/ParameterTopicsKafkaConsumer.php new file mode 100644 index 000000000..972bc9e3c --- /dev/null +++ b/packages/Kafka/tests/Fixture/DynamicTopics/ParameterTopicsKafkaConsumer.php @@ -0,0 +1,34 @@ +messages[] = $payload; + } + + /** + * @return string[] + */ + #[QueryHandler('parameterTopicsConsumer.getMessages')] + public function getMessages(): array + { + return $this->messages; + } +} diff --git a/packages/Kafka/tests/Fixture/DynamicTopics/TopicsProvider.php b/packages/Kafka/tests/Fixture/DynamicTopics/TopicsProvider.php new file mode 100644 index 000000000..b3d1677cd --- /dev/null +++ b/packages/Kafka/tests/Fixture/DynamicTopics/TopicsProvider.php @@ -0,0 +1,19 @@ +toRfc4122(); + + $ecotoneLite = EcotoneLite::bootstrapFlowTesting( + [DynamicTopicsKafkaConsumer::class], + [ + new DynamicTopicsKafkaConsumer(), + 'topicsProvider' => new TopicsProvider(), + KafkaBrokerConfiguration::class => ConnectionTestCase::getConnection(), + ], + ServiceConfiguration::createWithDefaults() + ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::KAFKA_PACKAGE])) + ->withExtensionObjects([ + TopicConfiguration::createWithReferenceName('dynamicOrdersTopic', $topicName), + ]), + licenceKey: LicenceTesting::VALID_LICENCE, + ); + + $this->publishToTopic($topicName, 'order-placed'); + + $ecotoneLite->run('dynamicTopicsConsumer', ExecutionPollingMetadata::createWithTestingSetup( + amountOfMessagesToHandle: 1, + maxExecutionTimeInMilliseconds: 30000, + )); + + $this->assertSame( + ['order-placed'], + $ecotoneLite->sendQueryWithRouting('dynamicTopicsConsumer.getMessages'), + 'Expression-provided topic reference name should have been resolved and subscribed to.', + ); + } + + public function test_mixed_literal_and_expression_topics_are_both_subscribed(): void + { + $literalTopicName = 'literal_orders_' . Uuid::v7()->toRfc4122(); + $dynamicTopicName = 'dynamic_orders_' . Uuid::v7()->toRfc4122(); + + $ecotoneLite = EcotoneLite::bootstrapFlowTesting( + [MixedTopicsKafkaConsumer::class], + [ + new MixedTopicsKafkaConsumer(), + 'topicsProvider' => new TopicsProvider(), + KafkaBrokerConfiguration::class => ConnectionTestCase::getConnection(), + ], + ServiceConfiguration::createWithDefaults() + ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::KAFKA_PACKAGE])) + ->withExtensionObjects([ + TopicConfiguration::createWithReferenceName('literalOrdersTopic', $literalTopicName), + TopicConfiguration::createWithReferenceName('dynamicOrdersTopic', $dynamicTopicName), + ]), + licenceKey: LicenceTesting::VALID_LICENCE, + ); + + $this->publishToTopic($literalTopicName, 'literal-order-placed'); + $this->publishToTopic($dynamicTopicName, 'dynamic-order-placed'); + + $ecotoneLite->run('mixedTopicsConsumer', ExecutionPollingMetadata::createWithTestingSetup( + amountOfMessagesToHandle: 2, + maxExecutionTimeInMilliseconds: 30000, + )); + + $messages = $ecotoneLite->sendQueryWithRouting('mixedTopicsConsumer.getMessages'); + sort($messages); + + $this->assertSame( + ['dynamic-order-placed', 'literal-order-placed'], + $messages, + 'Both the literal topic and the expression-resolved topic should have been subscribed to.', + ); + } + + public function test_parameter_function_resolves_topic_via_configuration_variable_service(): void + { + $topicName = 'dynamic_orders_' . Uuid::v7()->toRfc4122(); + + $ecotoneLite = EcotoneLite::bootstrapFlowTesting( + [ParameterTopicsKafkaConsumer::class], + [ + new ParameterTopicsKafkaConsumer(), + KafkaBrokerConfiguration::class => ConnectionTestCase::getConnection(), + ], + ServiceConfiguration::createWithDefaults() + ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::KAFKA_PACKAGE])) + ->withExtensionObjects([ + TopicConfiguration::createWithReferenceName('dynamicOrdersTopic', $topicName), + ]), + configurationVariables: ['ordersTopicReferenceName' => 'dynamicOrdersTopic'], + licenceKey: LicenceTesting::VALID_LICENCE, + ); + + $this->publishToTopic($topicName, 'order-placed-via-parameter'); + + $ecotoneLite->run('parameterTopicsConsumer', ExecutionPollingMetadata::createWithTestingSetup( + amountOfMessagesToHandle: 1, + maxExecutionTimeInMilliseconds: 30000, + )); + + $this->assertSame( + ['order-placed-via-parameter'], + $ecotoneLite->sendQueryWithRouting('parameterTopicsConsumer.getMessages'), + "parameter('ordersTopicReferenceName') should resolve via ConfigurationVariableService to the real topic reference name.", + ); + } + + private function publishToTopic(string $topic, string $payload): void + { + $brokerList = ConnectionTestCase::getConnection()->getBootstrapServers()[0]; + + $conf = new Conf(); + $conf->set('metadata.broker.list', $brokerList); + $conf->set('socket.timeout.ms', '50'); + $producer = new Producer($conf); + + $kafkaTopic = $producer->newTopic($topic); + $kafkaTopic->produce(RD_KAFKA_PARTITION_UA, 0, $payload); + $producer->poll(0); + + for ($i = 0; $i < 50 && $producer->getOutQLen() > 0; $i++) { + $producer->poll(50); + } + } +} diff --git a/packages/PdoEventSourcing/tests/Integration/AsynchronousEventDrivenProjectionTest.php b/packages/PdoEventSourcing/tests/Integration/AsynchronousEventDrivenProjectionTest.php index 994a23395..189213faf 100644 --- a/packages/PdoEventSourcing/tests/Integration/AsynchronousEventDrivenProjectionTest.php +++ b/packages/PdoEventSourcing/tests/Integration/AsynchronousEventDrivenProjectionTest.php @@ -89,8 +89,8 @@ classesToResolve: [ $ecotone->run(InProgressTicketList::PROJECTION_CHANNEL); $finishTime = microtime(true); - // around ~300 ms as default testing setup is 100ms (however connection and set up might take longer) - self::assertLessThan(300, ($finishTime - $currentTime) * 1000); + // well below the default 1s polling timeout, proving the run does not wait for it (CI runners can be slow) + self::assertLessThan(1000, ($finishTime - $currentTime) * 1000); self::assertEquals([['ticket_id' => '123', 'ticket_type' => 'alert']], $ecotone->sendQueryWithRouting('getInProgressTickets')); } diff --git a/packages/Tempest/src/MessagingSystemInitializer.php b/packages/Tempest/src/MessagingSystemInitializer.php index 068e9a0fe..23cd89a64 100644 --- a/packages/Tempest/src/MessagingSystemInitializer.php +++ b/packages/Tempest/src/MessagingSystemInitializer.php @@ -161,7 +161,7 @@ private function prepareFromCache( if ($useProductionCache && $cacheDirectory) { $ecotoneContainer = EcotoneSymfonyContainerFactory::loadCachedWithDefaults( new ServiceCacheConfiguration($cacheDirectory, true), - new TempestConfigurationVariableService(), + new TempestConfigurationVariableService($container), $externalContainer, ); if ($ecotoneContainer !== null) { @@ -183,11 +183,11 @@ private function prepareFromCache( $ecotoneContainer = EcotoneSymfonyContainerFactory::bootstrap( $serviceCacheConfiguration, - new TempestConfigurationVariableService(), + new TempestConfigurationVariableService($container), $externalContainer, fn () => MessagingSystemConfiguration::prepareWithAnnotationFinder( $annotationFinder, - new TempestConfigurationVariableService(), + new TempestConfigurationVariableService($container), $applicationConfiguration, enableTestPackage: $enableTesting, ), diff --git a/packages/Tempest/src/TempestConfigurationVariableService.php b/packages/Tempest/src/TempestConfigurationVariableService.php index dce93e94c..d8711e1e4 100644 --- a/packages/Tempest/src/TempestConfigurationVariableService.php +++ b/packages/Tempest/src/TempestConfigurationVariableService.php @@ -5,6 +5,7 @@ namespace Ecotone\Tempest; use Ecotone\Messaging\ConfigurationVariableService; +use Tempest\Container\Container; use function Tempest\env; @@ -13,13 +14,29 @@ */ final class TempestConfigurationVariableService implements ConfigurationVariableService { + public function __construct(private Container $container) + { + } + public function getByName(string $name): mixed { + if (str_contains($name, '::')) { + [$class, $property] = explode('::', $name, 2); + + return $this->container->get($class)->{$property}; + } + return env($name); } public function hasName(string $name): bool { + if (str_contains($name, '::')) { + [$class, $property] = explode('::', $name, 2); + + return class_exists($class) && property_exists($class, $property); + } + return getenv($name) !== false; } } diff --git a/packages/Tempest/tests/Fixture/Config/OrdersMultiTenancyConfig.php b/packages/Tempest/tests/Fixture/Config/OrdersMultiTenancyConfig.php new file mode 100644 index 000000000..6d7b84828 --- /dev/null +++ b/packages/Tempest/tests/Fixture/Config/OrdersMultiTenancyConfig.php @@ -0,0 +1,13 @@ +assertTrue($service->hasName('ECOTONE_TEST_VAR')); $this->assertSame('test-value', $service->getByName('ECOTONE_TEST_VAR')); @@ -25,9 +27,27 @@ public function test_reads_env_variables(): void public function test_returns_false_for_missing_env_variable(): void { - $service = new TempestConfigurationVariableService(); + $service = new TempestConfigurationVariableService(new GenericContainer()); $this->assertFalse($service->hasName('ECOTONE_NONEXISTENT_VAR_XYZ')); $this->assertNull($service->getByName('ECOTONE_NONEXISTENT_VAR_XYZ')); } + + public function test_reads_property_from_a_config_class_registered_in_the_container(): void + { + $container = new GenericContainer(); + $container->singleton(OrdersMultiTenancyConfig::class, new OrdersMultiTenancyConfig('dynamicOrdersTopic')); + + $service = new TempestConfigurationVariableService($container); + + $this->assertTrue($service->hasName(OrdersMultiTenancyConfig::class . '::topicReferenceName')); + $this->assertSame('dynamicOrdersTopic', $service->getByName(OrdersMultiTenancyConfig::class . '::topicReferenceName')); + } + + public function test_returns_false_for_a_config_class_property_that_does_not_exist(): void + { + $service = new TempestConfigurationVariableService(new GenericContainer()); + + $this->assertFalse($service->hasName(OrdersMultiTenancyConfig::class . '::missingProperty')); + } }