Asynchronous extendable Gateways

packages/Ecotone/src/Lite/Test/FlowTestSupport.php

@@ -20,7 +20,7 @@
 use Ecotone\Messaging\Support\MessageBuilder;
 use Ecotone\Modelling\AggregateMessage;
 use Ecotone\Modelling\CommandBus;
-use Ecotone\Modelling\Config\BusModule;
+use Ecotone\Modelling\Config\MessageBusChannel;
 use Ecotone\Modelling\Config\ModellingHandlerModule;
 use Ecotone\Modelling\Event;
 use Ecotone\Modelling\EventBus;
@@ -303,9 +303,9 @@ public function getRecordedCommandsWithRouting(): array
     {
         $commandWithRouting = [];
         foreach ($this->getRecordedCommandHeaders() as $commandHeaders) {
-            if ($commandHeaders->containsKey(BusModule::COMMAND_CHANNEL_NAME_BY_NAME)) {
+            if ($commandHeaders->containsKey(MessageBusChannel::COMMAND_CHANNEL_NAME_BY_NAME)) {
                 $command = [
-                    $commandHeaders->get(BusModule::COMMAND_CHANNEL_NAME_BY_NAME),
+                    $commandHeaders->get(MessageBusChannel::COMMAND_CHANNEL_NAME_BY_NAME),
                 ];
 
                 if ($commandHeaders->containsKey('aggregate.id')) {

packages/Ecotone/src/Messaging/Config/Annotation/ModuleConfiguration/GatewayModule.php

@@ -5,6 +5,7 @@
 namespace Ecotone\Messaging\Config\Annotation\ModuleConfiguration;
 
 use Ecotone\AnnotationFinder\AnnotationFinder;
+use Ecotone\Messaging\Attribute\Asynchronous;
 use Ecotone\Messaging\Attribute\MessageGateway;
 use Ecotone\Messaging\Attribute\ModuleAnnotation;
 use Ecotone\Messaging\Attribute\Parameter\Header;
@@ -24,6 +25,8 @@
 use Ecotone\Messaging\Handler\Gateway\ParameterToMessageConverter\GatewayPayloadExpressionBuilder;
 use Ecotone\Messaging\Handler\InterfaceParameter;
 use Ecotone\Messaging\Handler\InterfaceToCallRegistry;
+use Ecotone\Messaging\Handler\TypeDescriptor;
+use Ecotone\Messaging\Support\LicensingException;
 
 #[ModuleAnnotation]
 /**
@@ -123,6 +126,12 @@ public function canHandle($extensionObject): bool
     public function prepare(Configuration $messagingConfiguration, array $extensionObjects, ModuleReferenceSearchService $moduleReferenceSearchService, InterfaceToCallRegistry $interfaceToCallRegistry): void
     {
         foreach ($this->gatewayBuilders as $gatewayBuilder) {
+            /** @var Asynchronous[] $asynchronous */
+            $asynchronous = $interfaceToCallRegistry->getFor($gatewayBuilder->getInterfaceName(), $gatewayBuilder->getRelatedMethodName())->getAnnotationsByImportanceOrder(TypeDescriptor::create(Asynchronous::class));
+            if ($asynchronous && !$messagingConfiguration->isRunningForEnterpriseLicence()) {
+                throw LicensingException::create("Gateway {$gatewayBuilder->getInterfaceName()}::{$gatewayBuilder->getRelatedMethodName()} is marked as asynchronous. This functionality is available as part of Ecotone Enterprise.");
+            }
+
             $messagingConfiguration->registerGatewayBuilder($gatewayBuilder);
         }
     }

packages/Ecotone/src/Messaging/Config/MessagingSystemConfiguration.php

@@ -61,7 +61,7 @@
 use Ecotone\Messaging\NullableMessageChannel;
 use Ecotone\Messaging\PollableChannel;
 use Ecotone\Messaging\Support\Assert;
-use Ecotone\Modelling\Config\BusModule;
+use Ecotone\Modelling\Config\MessageBusChannel;
 use Exception;
 
 use function is_a;
@@ -410,10 +410,10 @@ private function configureAsynchronousEndpoints(): void
                     $this->registerMessageHandler(
                         UninterruptibleServiceActivator::create(
                             HeaderEnricher::create([
-                                BusModule::COMMAND_CHANNEL_NAME_BY_NAME => null,
-                                BusModule::COMMAND_CHANNEL_NAME_BY_OBJECT => null,
-                                BusModule::EVENT_CHANNEL_NAME_BY_OBJECT => null,
-                                BusModule::EVENT_CHANNEL_NAME_BY_NAME => null,
+                                MessageBusChannel::COMMAND_CHANNEL_NAME_BY_NAME => null,
+                                MessageBusChannel::COMMAND_CHANNEL_NAME_BY_OBJECT => null,
+                                MessageBusChannel::EVENT_CHANNEL_NAME_BY_OBJECT => null,
+                                MessageBusChannel::EVENT_CHANNEL_NAME_BY_NAME => null,
                                 MessageHeaders::ROUTING_SLIP => implode(',', $consequentialChannels),
                             ]),
                             'transform',

packages/Ecotone/src/Messaging/Config/ModuleClassList.php

@@ -44,12 +44,15 @@
 use Ecotone\Messaging\Config\Annotation\ModuleConfiguration\TransformerModule;
 use Ecotone\Messaging\Handler\Logger\Config\LoggingModule;
 use Ecotone\Messaging\Handler\Logger\Config\MessageHandlerLogger;
-use Ecotone\Modelling\Config\BusModule;
+use Ecotone\Modelling\CommandBus;
+use Ecotone\Modelling\Config\MessageBusChannel;
 use Ecotone\Modelling\Config\BusRoutingModule;
 use Ecotone\Modelling\Config\DistributedGatewayModule;
 use Ecotone\Modelling\Config\InstantRetry\InstantRetryModule;
 use Ecotone\Modelling\Config\ModellingHandlerModule;
+use Ecotone\Modelling\EventBus;
 use Ecotone\Modelling\MessageHandling\MetadataPropagator\MessageHeadersPropagatorInterceptor;
+use Ecotone\Modelling\QueryBus;
 use Ecotone\OpenTelemetry\Configuration\OpenTelemetryModule;
 use Ecotone\Redis\Configuration\RedisMessageConsumerModule;
 use Ecotone\Redis\Configuration\RedisMessagePublisherModule;
@@ -66,7 +69,6 @@ class ModuleClassList
         DistributedGatewayModule::class,
         ModellingHandlerModule::class,
         BusRoutingModule::class,
-        BusModule::class,
         MethodInterceptorModule::class,
         MessagingCommandsModule::class,
         EndpointHeadersInterceptorModule::class,
@@ -92,6 +94,9 @@ class ModuleClassList
         /** Attribute based configurations */
         MessageHeadersPropagatorInterceptor::class,
         MessageHandlerLogger::class,
+        CommandBus::class,
+        QueryBus::class,
+        EventBus::class,
     ];
 
     public const ASYNCHRONOUS_MODULE = [

packages/Ecotone/src/Messaging/Handler/Gateway/GatewayInternalProcessor.php

@@ -27,11 +27,15 @@
  */
 class GatewayInternalProcessor implements MessageProcessor, AroundInterceptable
 {
+    /**
+     * @param string[] $routingSlipChannels
+     */
     public function __construct(
         private string $interfaceToCallName,
         private ?Type $returnType,
         private bool $returnTypeAllowsNull,
         private MessageChannel $requestChannel,
+        private array $routingSlipChannels,
         private ?PollableChannel $replyChannel,
         private int $replyMilliSecondsTimeout
     ) {
@@ -47,6 +51,8 @@ public function process(Message $requestMessage): ?Message
     {
         //      Gateway can be called inside service activator. So it means, we need to preserve reply channel in order to reply with it
         $previousReplyChannel = $requestMessage->getHeaders()->containsKey(MessageHeaders::REPLY_CHANNEL) ? $requestMessage->getHeaders()->getReplyChannel() : null;
+        $previousRoutingSlip = $requestMessage->getHeaders()->containsKey(MessageHeaders::ROUTING_SLIP) ? $requestMessage->getHeaders()->get(MessageHeaders::ROUTING_SLIP) : null;
+
         $canReturnValue = $this->returnType?->isVoid() === false;
 
         $requestMessage = MessageBuilder::fromMessage($requestMessage);
@@ -58,7 +64,9 @@ public function process(Message $requestMessage): ?Message
             $requestMessage = $requestMessage->removeHeader(MessageHeaders::REPLY_CHANNEL);
         }
 
-        $requestMessage = $requestMessage->build();
+        $requestMessage = $requestMessage
+            ->setRoutingSlip($this->routingSlipChannels)
+            ->build();
 
         $this->requestChannel->send($requestMessage);
 
@@ -78,6 +86,7 @@ public function process(Message $requestMessage): ?Message
         if ($replyMessage instanceof Message && $previousReplyChannel) {
             $replyMessage = MessageBuilder::fromMessage($replyMessage)
                 ->setReplyChannel($previousReplyChannel)
+                ->setHeader(MessageHeaders::ROUTING_SLIP, $previousRoutingSlip)
                 ->build();
         }
 

packages/Ecotone/src/Messaging/Handler/Gateway/GatewayInternalProcessorBuilder.php

@@ -16,22 +16,30 @@
 
 class GatewayInternalProcessorBuilder implements InterceptedMessageProcessorBuilder
 {
+    /**
+     * @param string[] $asynchronousChannels
+     */
     public function __construct(
         private InterfaceToCallReference $interfaceToCallReference,
-        private string $requestChannelName,
-        private ?string $replyChannelName,
-        private int $replyMilliSecondsTimeout,
+        private string                   $requestChannelName,
+        private array                    $asynchronousChannels,
+        private ?string                  $replyChannelName,
+        private int                      $replyMilliSecondsTimeout,
     ) {
     }
 
     public function compile(MessagingContainerBuilder $builder, array $aroundInterceptors = []): Definition|Reference
     {
+        $routingSlipChannels = $this->asynchronousChannels;
+        $routingSlipChannels[] = $this->requestChannelName;
+
         $interfaceToCall = $builder->getInterfaceToCall($this->interfaceToCallReference);
         $gatewayInternalProcessor = new Definition(GatewayInternalProcessor::class, [
             $interfaceToCall->toString(),
             $interfaceToCall->getReturnType(),
             $interfaceToCall->canItReturnNull(),
-            new ChannelReference($this->requestChannelName),
+            new ChannelReference(array_shift($routingSlipChannels)),
+            $routingSlipChannels,
             $this->replyChannelName ? new ChannelReference($this->replyChannelName) : null,
             $this->replyMilliSecondsTimeout,
         ]);

packages/Ecotone/src/Messaging/Handler/Gateway/GatewayProxyBuilder.php

@@ -4,6 +4,7 @@
 
 namespace Ecotone\Messaging\Handler\Gateway;
 
+use Ecotone\Messaging\Attribute\Asynchronous;
 use Ecotone\Messaging\Config\Container\AttributeDefinition;
 use Ecotone\Messaging\Config\Container\ChannelReference;
 use Ecotone\Messaging\Config\Container\CompilableBuilder;
@@ -27,6 +28,7 @@
 use Ecotone\Messaging\Handler\Logger\LoggingGateway;
 use Ecotone\Messaging\Handler\Processor\ChainedMessageProcessorBuilder;
 use Ecotone\Messaging\Handler\Processor\MethodInvoker\AroundInterceptorBuilder;
+use Ecotone\Messaging\Handler\TypeDescriptor;
 use Ecotone\Messaging\MessageHeaders;
 use Ecotone\Messaging\MessagingException;
 use Ecotone\Messaging\PollableChannel;
@@ -316,7 +318,12 @@ public function compile(MessagingContainerBuilder $builder): Definition
         }
 
         foreach ($this->methodArgumentConverters as $messageConverterBuilder) {
-            $methodArgumentConverters[] = $messageConverterBuilder->compile($builder, $interfaceToCall);
+            /** This need to evaluated first, as other Header Converters may replace given Message Header */
+            if ($messageConverterBuilder instanceof GatewayHeadersBuilder) {
+                array_unshift($methodArgumentConverters, $messageConverterBuilder->compile($builder, $interfaceToCall));
+            } else {
+                $methodArgumentConverters[] = $messageConverterBuilder->compile($builder, $interfaceToCall);
+            }
         }
 
         $messageConverters = [];
@@ -380,10 +387,15 @@ private function compileGatewayInternalProcessor(MessagingContainerBuilder $buil
             $aroundInterceptors,
         );
 
+        /** @var Asynchronous[] $asynchronous */
+        $asynchronous = $interfaceToCall->getAnnotationsByImportanceOrder(TypeDescriptor::create(Asynchronous::class));
+        $channelNames = $asynchronous ? $asynchronous[0]->getChannelName() : [];
+
         return ChainedMessageProcessorBuilder::create()
             ->chainInterceptedProcessor(new GatewayInternalProcessorBuilder(
                 $interfaceToCallReference,
                 $this->requestChannelName,
+                $channelNames,
                 $this->replyChannelName,
                 $this->replyMilliSecondsTimeout
             ))

packages/Ecotone/src/Messaging/MessageHeaders.php

@@ -6,7 +6,7 @@
 use Ecotone\Messaging\Gateway\MessagingEntrypoint;
 use Ecotone\Messaging\Handler\TypeDescriptor;
 use Ecotone\Modelling\AggregateMessage;
-use Ecotone\Modelling\Config\BusModule;
+use Ecotone\Modelling\Config\MessageBusChannel;
 use Ecotone\Modelling\DistributionEntrypoint;
 
 use function json_encode;
@@ -265,12 +265,12 @@ public static function unsetDistributionKeys(array $metadata): array
     public static function unsetBusKeys(array $metadata): array
     {
         unset(
-            $metadata[BusModule::COMMAND_CHANNEL_NAME_BY_NAME],
-            $metadata[BusModule::COMMAND_CHANNEL_NAME_BY_OBJECT],
-            $metadata[BusModule::EVENT_CHANNEL_NAME_BY_NAME],
-            $metadata[BusModule::EVENT_CHANNEL_NAME_BY_OBJECT],
-            $metadata[BusModule::QUERY_CHANNEL_NAME_BY_NAME],
-            $metadata[BusModule::QUERY_CHANNEL_NAME_BY_OBJECT]
+            $metadata[MessageBusChannel::COMMAND_CHANNEL_NAME_BY_NAME],
+            $metadata[MessageBusChannel::COMMAND_CHANNEL_NAME_BY_OBJECT],
+            $metadata[MessageBusChannel::EVENT_CHANNEL_NAME_BY_NAME],
+            $metadata[MessageBusChannel::EVENT_CHANNEL_NAME_BY_OBJECT],
+            $metadata[MessageBusChannel::QUERY_CHANNEL_NAME_BY_NAME],
+            $metadata[MessageBusChannel::QUERY_CHANNEL_NAME_BY_OBJECT]
         );
 
         return $metadata;

packages/Ecotone/src/Messaging/Support/MessageBuilder.php

@@ -112,6 +112,17 @@ public function setHeader(string $headerName, $headerValue): self
         return $this;
     }
 
+    public function setRoutingSlip(array $routingSlip): self
+    {
+        if (!$routingSlip) {
+            return $this;
+        }
+
+        $this->setHeader(MessageHeaders::ROUTING_SLIP, implode(',', $routingSlip));
+
+        return $this;
+    }
+
     /**
      * @param MediaType $mediaType
      * @return MessageBuilder

packages/Ecotone/src/Modelling/CommandBus.php

@@ -4,14 +4,30 @@
 
 namespace Ecotone\Modelling;
 
+use Ecotone\Messaging\Attribute\MessageGateway;
+use Ecotone\Messaging\Attribute\Parameter\Header;
+use Ecotone\Messaging\Attribute\Parameter\Headers;
+use Ecotone\Messaging\Attribute\Parameter\Payload;
 use Ecotone\Messaging\Conversion\MediaType;
+use Ecotone\Messaging\MessageHeaders;
+use Ecotone\Modelling\Config\MessageBusChannel;
 
 /**
  * licence Apache-2.0
  */
 interface CommandBus
 {
-    public function send(object $command, array $metadata = []): mixed;
+    #[MessageGateway(MessageBusChannel::COMMAND_CHANNEL_NAME_BY_OBJECT)]
+    public function send(
+        #[Payload] object $command,
+        #[Headers] array  $metadata = []
+    ): mixed;
 
-    public function sendWithRouting(string $routingKey, mixed $command = [], string $commandMediaType = MediaType::APPLICATION_X_PHP, array $metadata = []): mixed;
+    #[MessageGateway(MessageBusChannel::COMMAND_CHANNEL_NAME_BY_NAME)]
+    public function sendWithRouting(
+        #[Header(MessageBusChannel::COMMAND_CHANNEL_NAME_BY_NAME)] string $routingKey,
+        #[Payload] mixed                                                  $command = [],
+        #[Header(MessageHeaders::CONTENT_TYPE)] string                    $commandMediaType = MediaType::APPLICATION_X_PHP,
+        #[Headers] array                                                  $metadata = []
+    ): mixed;
 }