feat: resolve KafkaConsumer topics from expressions and config parameters (fixes #693)

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');

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

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']));

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)) {

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),
             ])
         );
     }

packages/Kafka/tests/Attribute/KafkaConsumerUnsupportedTopicsSyntaxTest.php

@@ -0,0 +1,41 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Kafka\Attribute;
+
+use Ecotone\Kafka\Attribute\KafkaConsumer;
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\ExpressionLanguage\Expression;
+use TypeError;
+
+/**
+ * licence Enterprise
+ * @internal
+ */
+final class KafkaConsumerUnsupportedTopicsSyntaxTest extends TestCase
+{
+    public function test_a_symfony_expression_object_is_not_accepted_as_topics(): void
+    {
+        $this->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.",
+        );
+    }
+}

packages/Kafka/tests/Fixture/DynamicTopics/DynamicTopicsKafkaConsumer.php

@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Kafka\Fixture\DynamicTopics;
+
+use Ecotone\Kafka\Attribute\KafkaConsumer;
+use Ecotone\Modelling\Attribute\QueryHandler;
+
+/**
+ * licence Enterprise
+ */
+final class DynamicTopicsKafkaConsumer
+{
+    /**
+     * @var string[]
+     */
+    private array $messages = [];
+
+    #[KafkaConsumer('dynamicTopicsConsumer', topics: "reference('topicsProvider').getTopics()")]
+    public function handle(string $payload): void
+    {
+        $this->messages[] = $payload;
+    }
+
+    /**
+     * @return string[]
+     */
+    #[QueryHandler('dynamicTopicsConsumer.getMessages')]
+    public function getMessages(): array
+    {
+        return $this->messages;
+    }
+}

packages/Kafka/tests/Fixture/DynamicTopics/MixedTopicsKafkaConsumer.php

@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Kafka\Fixture\DynamicTopics;
+
+use Ecotone\Kafka\Attribute\KafkaConsumer;
+use Ecotone\Modelling\Attribute\QueryHandler;
+
+/**
+ * licence Enterprise
+ */
+final class MixedTopicsKafkaConsumer
+{
+    /**
+     * @var string[]
+     */
+    private array $messages = [];
+
+    #[KafkaConsumer('mixedTopicsConsumer', topics: ['literalOrdersTopic', "reference('topicsProvider').getTopics()"])]
+    public function handle(string $payload): void
+    {
+        $this->messages[] = $payload;
+    }
+
+    /**
+     * @return string[]
+     */
+    #[QueryHandler('mixedTopicsConsumer.getMessages')]
+    public function getMessages(): array
+    {
+        return $this->messages;
+    }
+}

packages/Kafka/tests/Fixture/DynamicTopics/ParameterTopicsKafkaConsumer.php

@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Kafka\Fixture\DynamicTopics;
+
+use Ecotone\Kafka\Attribute\KafkaConsumer;
+use Ecotone\Modelling\Attribute\QueryHandler;
+
+/**
+ * licence Enterprise
+ */
+final class ParameterTopicsKafkaConsumer
+{
+    /**
+     * @var string[]
+     */
+    private array $messages = [];
+
+    #[KafkaConsumer('parameterTopicsConsumer', topics: "parameter('ordersTopicReferenceName')")]
+    public function handle(string $payload): void
+    {
+        $this->messages[] = $payload;
+    }
+
+    /**
+     * @return string[]
+     */
+    #[QueryHandler('parameterTopicsConsumer.getMessages')]
+    public function getMessages(): array
+    {
+        return $this->messages;
+    }
+}

packages/Kafka/tests/Fixture/DynamicTopics/TopicsProvider.php

@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Kafka\Fixture\DynamicTopics;
+
+/**
+ * licence Enterprise
+ */
+final class TopicsProvider
+{
+    /**
+     * @return string[]
+     */
+    public function getTopics(): array
+    {
+        return ['dynamicOrdersTopic'];
+    }
+}