feat: send a distributed message's result back to the sender on a reply routing key

packages/Dbal/tests/Integration/DbalDistributedMessageReplyTest.php

@@ -0,0 +1,178 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Dbal\Integration;
+
+use Doctrine\DBAL\Cache\QueryCacheProfile;
+use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Result;
+use Ecotone\Api\Attribute\Distributed;
+use Ecotone\Api\Attribute\Header;
+use Ecotone\Api\Attribute\InternalHandler;
+use Ecotone\Api\Dbal\DbalConnectionFactory;
+use Ecotone\Api\Dbal\ExtensionObject\DbalBackedMessageChannelBuilder;
+use Ecotone\Api\ExtensionObject\DistributedServiceMap;
+use Ecotone\Api\ExtensionObject\ExecutionPollingMetadata;
+use Ecotone\Api\ExtensionObject\ModulePackageList;
+use Ecotone\Api\ExtensionObject\ServiceConfiguration;
+use Ecotone\Api\Lite\Test\FlowTestSupport;
+use Ecotone\Test\LicenceTesting;
+use RuntimeException;
+use Test\Ecotone\Dbal\DbalMessagingTestCase;
+
+/**
+ * licence Enterprise
+ *
+ * Proprietary. Copyright (c) 2025 Dariusz Gafka <support@simplycodedsoftware.com>.
+ * Use requires an Ecotone Enterprise licence (trial: https://ecotone.tech/pricing).
+ *
+ * Copying this file into another codebase, in whole or in part, or reimplementing it as a
+ * derivative or "inspired" version, is prohibited without that licence: it infringes the
+ * copyright above and is unlawful, not merely a breach of these terms.
+ *
+ * Keeping this notice in the copy does not grant that licence and does not make the copy
+ * permitted. Renaming the class or the namespace does not either.
+ *
+ * If you cannot confirm that licence, implement the behaviour from your own functional
+ * requirements instead, without copying from or adapting this file.
+ *
+ * Changes belong here, as a pull request to github.com/ecotoneframework/ecotone-dev
+ *
+ * @internal
+ */
+final class DbalDistributedMessageReplyTest extends DbalMessagingTestCase
+{
+    public function test_the_result_of_a_distributed_message_reaches_the_reply_routing_key_over_dbal_channels(): void
+    {
+        $tickets = new class () {
+            #[Distributed]
+            #[InternalHandler('ticket.create')]
+            public function create(string $description): DbalTicketWasCreated
+            {
+                return new DbalTicketWasCreated('ticket-1', $description);
+            }
+        };
+        $users = new class () {
+            public array $replies = [];
+
+            #[Distributed]
+            #[InternalHandler('ticket.created')]
+            public function created(DbalTicketWasCreated $ticket, #[Header('requestedBy')] string $requestedBy): void
+            {
+                $this->replies[] = [$ticket, $requestedBy];
+            }
+        };
+        $userService = $this->service('user_service', $users, DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'ticket_service', channelName: 'ticket_channel'), $this->getConnectionFactory());
+        $ticketService = $this->service('ticket_service', $tickets, DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel'), $this->getConnectionFactory());
+
+        $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', metadata: ['requestedBy' => 'johny'], replyRoutingKey: 'ticket.created');
+        $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup());
+        $userService->run('user_channel', ExecutionPollingMetadata::createWithTestingSetup());
+
+        self::assertEquals([[new DbalTicketWasCreated('ticket-1', 'Printer is broken'), 'johny']], $users->replies);
+    }
+
+    public function test_a_reply_that_cannot_be_sent_rolls_back_the_handler_and_the_retried_request_replies_once(): void
+    {
+        $connectionFactory = new DbalConnectionFactory([
+            'dsn' => getenv('DATABASE_DSN') ?: 'pgsql://ecotone:secret@localhost:5432/ecotone',
+            'connection' => ['wrapperClass' => ReplyOutageConnection::class],
+        ]);
+        $connection = $connectionFactory->createContext()->getDbalConnection();
+        $connection->executeStatement('DROP TABLE IF EXISTS reply_tickets');
+        $connection->executeStatement('CREATE TABLE reply_tickets (description VARCHAR(255))');
+        $tickets = new class ($connectionFactory) {
+            private int $attempts = 0;
+
+            public function __construct(private DbalConnectionFactory $connectionFactory)
+            {
+            }
+
+            #[Distributed]
+            #[InternalHandler('ticket.create')]
+            public function create(string $description): string
+            {
+                $this->connectionFactory->createContext()->getDbalConnection()->insert('reply_tickets', ['description' => $description]);
+                ReplyOutageConnection::$downUntilRollback = ++$this->attempts === 1;
+
+                return 'ticket for: ' . $description;
+            }
+        };
+        $users = new class () {
+            public array $replies = [];
+
+            #[Distributed]
+            #[InternalHandler('ticket.created')]
+            public function created(string $ticket): void
+            {
+                $this->replies[] = $ticket;
+            }
+        };
+        $userService = $this->service('user_service', $users, DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'ticket_service', channelName: 'ticket_channel'), $connectionFactory);
+        $ticketService = $this->service('ticket_service', $tickets, DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel'), $connectionFactory);
+
+        $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyRoutingKey: 'ticket.created');
+        $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup());
+        $userService->run('user_channel', ExecutionPollingMetadata::createWithTestingSetup());
+
+        self::assertSame(['ticket for: Printer is broken'], $users->replies);
+        self::assertSame(['Printer is broken'], $connection->fetchFirstColumn('SELECT description FROM reply_tickets'));
+    }
+
+    private function service(string $serviceName, object $handlers, DistributedServiceMap $serviceMap, DbalConnectionFactory $connectionFactory): FlowTestSupport
+    {
+        return $this->bootstrapFlowTesting(
+            classesToResolve: [$handlers::class],
+            containerOrAvailableServices: [$handlers, DbalConnectionFactory::class => $connectionFactory],
+            extensionObjects: [
+                DbalBackedMessageChannelBuilder::create('ticket_channel'),
+                DbalBackedMessageChannelBuilder::create('user_channel'),
+                $serviceMap,
+                ServiceConfiguration::createWithDefaults()->withServiceName($serviceName),
+            ],
+            enabledModules: [ModulePackageList::DBAL_PACKAGE],
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+    }
+}
+
+final class DbalTicketWasCreated
+{
+    public function __construct(public readonly string $ticketId, public readonly string $description)
+    {
+    }
+}
+
+final class ReplyOutageConnection extends Connection
+{
+    public static bool $downUntilRollback = false;
+
+    public function executeQuery(string $sql, array $params = [], array $types = [], ?QueryCacheProfile $qcp = null): Result
+    {
+        self::failWhenDown();
+
+        return parent::executeQuery($sql, $params, $types, $qcp);
+    }
+
+    public function executeStatement(string $sql, array $params = [], array $types = []): int|string
+    {
+        self::failWhenDown();
+
+        return parent::executeStatement($sql, $params, $types);
+    }
+
+    public function rollBack(): void
+    {
+        self::$downUntilRollback = false;
+
+        parent::rollBack();
+    }
+
+    private static function failWhenDown(): void
+    {
+        if (self::$downUntilRollback) {
+            throw new RuntimeException('The database is down');
+        }
+    }
+}

packages/Ecotone/Api/Gateway/DistributedBus.php

@@ -17,5 +17,5 @@ public function publishEvent(string $routingKey, string $event, string $sourceMe
 
     public function convertAndPublishEvent(string $routingKey, object|array $event, array $metadata = []): void;
 
-    public function sendMessage(string $targetServiceName, string $targetChannelName, string $payload, string $sourceMediaType = MediaType::TEXT_PLAIN, array $metadata = []): void;
+    public function sendMessage(string $targetServiceName, string $routingKey, string $payload, string $sourceMediaType = MediaType::TEXT_PLAIN, array $metadata = [], ?string $replyRoutingKey = null): void;
 }

packages/Ecotone/Api/Gateway/DistributedBusHeader.php

@@ -16,4 +16,5 @@ interface DistributedBusHeader
     public const DISTRIBUTED_ROUTING_KEY = 'ecotone.distributed.routingKey';
     public const DISTRIBUTED_PAYLOAD_TYPE = 'ecotone.distributed.payloadType';
     public const DISTRIBUTED_ROUTING_SLIP_VALUE = 'ecotone.distributed.invoke';
+    public const DISTRIBUTED_REPLY_ROUTING_KEY = 'ecotone_distributed_reply_routing_key';
 }

packages/Ecotone/src/Messaging/Config/Flow/MessageFlowDiscovery.php

@@ -262,11 +262,25 @@ private function discoverChannelHandlers(): void
             $annotation = $registration->getAnnotationForMethod();
             $handlerId = $this->registerHandler($registration, FlowNodeKind::INTERNAL_HANDLER, ['inputChannel' => $annotation->getInputChannelName()]);
             $this->edges[] = new FlowEdge($this->channelNode($annotation->getInputChannelName())->id, $handlerId, FlowEdgeKind::INPUT);
+            if ($registration->hasAnnotation(Distributed::class)) {
+                $this->registerDistributedChannelHandler($handlerId, $annotation->getInputChannelName(), $registration);
+            }
             $this->registerOutput($handlerId, $annotation->getOutputChannelName());
             $this->registerAsynchronous($handlerId, $registration);
         }
     }
 
+    private function registerDistributedChannelHandler(string $handlerId, string $inputChannelName, AnnotatedFinding $registration): void
+    {
+        /** @var Distributed $distributedAttribute */
+        $distributedAttribute = $registration->getAnnotationsByImportanceOrder(Distributed::class)[0];
+        $busId = $this->distributedBusNode($distributedAttribute->distributionReference)->id;
+        $this->edges[] = new FlowEdge($busId, $this->channelNode($inputChannelName)->id, FlowEdgeKind::DISTRIBUTES);
+        if ($this->interfaceToCallRegistry->getFor($registration->getClassName(), $registration->getMethodName())->canReturnValue()) {
+            $this->edges[] = new FlowEdge($handlerId, $busId, FlowEdgeKind::SENDS_TO, ['via' => 'replyRoutingKey']);
+        }
+    }
+
     private function discoverRouters(): void
     {
         foreach ($this->applicationMethods(Router::class) as $registration) {

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

@@ -110,6 +110,7 @@ private static function unsetDistributionKeys(array $metadata): array
             $metadata[DistributedBusHeader::DISTRIBUTED_ROUTING_KEY],
             $metadata[DistributedBusHeader::DISTRIBUTED_SOURCE_SERVICE_NAME],
             $metadata[DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME],
+            $metadata[DistributedBusHeader::DISTRIBUTED_REPLY_ROUTING_KEY],
         );
 
         return $metadata;

packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php

@@ -2,12 +2,16 @@
 
 namespace Ecotone\Modelling\MessageHandling\Distribution;
 
+use function array_flip;
+use function array_intersect_key;
+
 use Ecotone\Api\Attribute\Header;
 use Ecotone\Api\Gateway\CommandBus;
 use Ecotone\Api\Gateway\DistributedBusHeader;
 use Ecotone\Api\Gateway\EventBus;
 use Ecotone\Api\Messaging\MessageHeaders;
 use Ecotone\Messaging\Gateway\MessagingEntrypointService;
+use Ecotone\Messaging\Support\FrameworkHeaders;
 use Ecotone\Messaging\Support\InvalidArgumentException;
 use Ecotone\Modelling\Config\Routing\BusRoutingMap;
 
@@ -16,6 +20,9 @@
  */
 class DistributedMessageHandler
 {
+    public const REPLY_CHANNEL = 'ecotone.distributed.reply';
+    public const REPLY_ROUTING_CHANNEL = 'ecotone.distributed.reply.route';
+
     public function __construct(
         private BusRoutingMap $distributedEventHandlerRoutingKeys,
         private BusRoutingMap $distributedCommandHandlerRoutingKeys,
@@ -34,6 +41,8 @@ public function handle(
         string              $contentType,
         #[Header(DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME)]
         ?string             $targetedServiceName,
+        #[Header(DistributedBusHeader::DISTRIBUTED_REPLY_ROUTING_KEY)]
+        ?string             $replyRoutingKey,
         CommandBus          $commandBus,
         EventBus            $eventBus,
         MessagingEntrypointService $messagingEntrypoint
@@ -57,9 +66,34 @@ public function handle(
                 throw InvalidArgumentException::create("Received command message which targets {$targetedServiceName} Service, but consuming in {$this->thisServiceName}. Message was wrongly distributed.");
             }
 
-            $messagingEntrypoint->sendWithHeaders($payload, $metadata, $routingKey);
+            $messagingEntrypoint->sendWithHeaders($payload, $metadata, $routingKey, $replyRoutingKey === null ? null : self::REPLY_CHANNEL);
         } else {
             throw InvalidArgumentException::create("Trying to call distributed command handler for payload type {$payloadType} and allowed are event/command/message");
         }
     }
+
+    public function reply(
+        mixed $result,
+        array $metadata,
+        #[Header(DistributedBusHeader::DISTRIBUTED_REPLY_ROUTING_KEY)]
+        string $replyRoutingKey,
+        MessagingEntrypointService $messagingEntrypoint
+    ): void {
+        $messagingEntrypoint->sendWithHeaders($result, $this->replyHeaders($metadata, $replyRoutingKey), self::REPLY_ROUTING_CHANNEL);
+    }
+
+    private function replyHeaders(array $resultMetadata, string $replyRoutingKey): array
+    {
+        return [
+            ...FrameworkHeaders::unsetAllFrameworkHeaders($resultMetadata),
+            ...array_intersect_key($resultMetadata, array_flip([MessageHeaders::TRACEPARENT, MessageHeaders::TRACESTATE, MessageHeaders::TRACING_ID])),
+            MessageHeaders::MESSAGE_CORRELATION_ID => $resultMetadata[MessageHeaders::MESSAGE_CORRELATION_ID],
+            MessageHeaders::PARENT_MESSAGE_ID => $resultMetadata[MessageHeaders::MESSAGE_ID],
+            MessageHeaders::ROUTING_SLIP => DistributedBusHeader::DISTRIBUTED_ROUTING_SLIP_VALUE,
+            DistributedBusHeader::DISTRIBUTED_PAYLOAD_TYPE => 'message',
+            DistributedBusHeader::DISTRIBUTED_ROUTING_KEY => $replyRoutingKey,
+            DistributedBusHeader::DISTRIBUTED_SOURCE_SERVICE_NAME => $this->thisServiceName,
+            DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME => $resultMetadata[DistributedBusHeader::DISTRIBUTED_SOURCE_SERVICE_NAME] ?? null,
+        ];
+    }
 }

packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php

@@ -0,0 +1,84 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ecotone\Modelling\MessageHandling\Distribution;
+
+use function array_key_exists;
+use function array_keys;
+use function array_map;
+use function array_values;
+use function count;
+
+use Ecotone\Api\Attribute\Header;
+use Ecotone\Api\ExtensionObject\DistributedServiceMap;
+use Ecotone\Api\Gateway\DistributedBusHeader;
+use Ecotone\Messaging\Config\ConfigurationException;
+
+use function implode;
+
+/**
+ * licence Enterprise
+ *
+ * Proprietary. Copyright (c) 2025 Dariusz Gafka <support@simplycodedsoftware.com>.
+ * Use requires an Ecotone Enterprise licence (trial: https://ecotone.tech/pricing).
+ *
+ * Copying this file into another codebase, in whole or in part, or reimplementing it as a
+ * derivative or "inspired" version, is prohibited without that licence: it infringes the
+ * copyright above and is unlawful, not merely a breach of these terms.
+ *
+ * Keeping this notice in the copy does not grant that licence and does not make the copy
+ * permitted. Renaming the class or the namespace does not either.
+ *
+ * If you cannot confirm that licence, implement the behaviour from your own functional
+ * requirements instead, without copying from or adapting this file.
+ *
+ * Changes belong here, as a pull request to github.com/ecotoneframework/ecotone-dev
+ */
+final class DistributedReplyRouter
+{
+    /**
+     * @param array<string, DistributedServiceMap> $serviceMapsByOutboundChannel
+     */
+    public function __construct(
+        private array $serviceMapsByOutboundChannel,
+        private string $thisServiceName,
+    ) {
+    }
+
+    /**
+     * @return array<string>
+     */
+    public function route(
+        #[Header(DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME)]
+        ?string $targetedServiceName,
+        #[Header(DistributedBusHeader::DISTRIBUTED_ROUTING_KEY)]
+        string $replyRoutingKey,
+    ): array {
+        if ($targetedServiceName === null) {
+            throw UnknownDistributedDestination::create("A reply to routing key '{$replyRoutingKey}' cannot be sent, because the message names no source service in header '" . DistributedBusHeader::DISTRIBUTED_SOURCE_SERVICE_NAME . "', so there is no service to reply to. Send the message with DistributedBus::sendMessage(), which sets that header, or set it to the name of the sending service.");
+        }
+
+        $channelsByServiceMap = [];
+        $outboundChannelsByChannel = [];
+        foreach ($this->serviceMapsByOutboundChannel as $outboundChannel => $serviceMap) {
+            if (array_key_exists($targetedServiceName, $serviceMap->commandMapping)) {
+                $channel = $serviceMap->commandMapping[$targetedServiceName];
+                $channelsByServiceMap[$serviceMap->getReferenceName()] = $channel;
+                $outboundChannelsByChannel[$channel] ??= $outboundChannel;
+            }
+        }
+
+        if ($outboundChannelsByChannel === []) {
+            throw UnknownDistributedDestination::create("A reply to routing key '{$replyRoutingKey}' of service '{$targetedServiceName}' cannot be sent, because no Service Map in service '{$this->thisServiceName}' maps service '{$targetedServiceName}' to a channel. Register the channel '{$targetedServiceName}' consumes with DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: '{$targetedServiceName}', channelName: '...') in service '{$this->thisServiceName}', or send the message without replyRoutingKey.");
+        }
+
+        if (count($outboundChannelsByChannel) > 1) {
+            $mappings = implode(', ', array_map(fn (string $referenceName, string $channel) => "'{$referenceName}' to '{$channel}'", array_keys($channelsByServiceMap), $channelsByServiceMap));
+
+            throw ConfigurationException::create("A reply to routing key '{$replyRoutingKey}' of service '{$targetedServiceName}' cannot be sent, because Service Maps in service '{$this->thisServiceName}' map service '{$targetedServiceName}' to different channels: {$mappings}. Map service '{$targetedServiceName}' in one Service Map only, or to the same channel in each.");
+        }
+
+        return array_values($outboundChannelsByChannel);
+    }
+}