From a847aeb543b83c93f074b13fa81e338e6f01dd12 Mon Sep 17 00:00:00 2001 From: Dariusz Gafka Date: Tue, 20 Oct 2026 17:10:00 +0200 Subject: [PATCH 01/12] feat(distribution): send the result of a distributed message back to the sender's reply-to channel --- .../Ecotone/Api/Gateway/DistributedBus.php | 2 +- .../Api/Gateway/DistributedBusHeader.php | 1 + .../Messaging/Support/FrameworkHeaders.php | 1 + .../DistributedMessageHandler.php | 25 +++++- .../Distribution/DistributedReplyRouter.php | 59 ++++++++++++++ .../DistributedBusWithServiceMapModule.php | 36 ++++++++- .../Module/DistributedHandlerModule.php | 1 + .../DistributedBusMessageReplyTest.php | 80 +++++++++++++++++++ 8 files changed, 202 insertions(+), 3 deletions(-) create mode 100644 packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php create mode 100644 packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php diff --git a/packages/Ecotone/Api/Gateway/DistributedBus.php b/packages/Ecotone/Api/Gateway/DistributedBus.php index a0bccd143c..1d6009f92d 100644 --- a/packages/Ecotone/Api/Gateway/DistributedBus.php +++ b/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 $targetChannelName, string $payload, string $sourceMediaType = MediaType::TEXT_PLAIN, array $metadata = [], ?string $replyToChannelName = null): void; } diff --git a/packages/Ecotone/Api/Gateway/DistributedBusHeader.php b/packages/Ecotone/Api/Gateway/DistributedBusHeader.php index f9d0eb9827..84744b762e 100644 --- a/packages/Ecotone/Api/Gateway/DistributedBusHeader.php +++ b/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_TO_CHANNEL = 'ecotone_distributed_reply_to_channel'; } diff --git a/packages/Ecotone/src/Messaging/Support/FrameworkHeaders.php b/packages/Ecotone/src/Messaging/Support/FrameworkHeaders.php index eb950a34d6..ed0f35665c 100644 --- a/packages/Ecotone/src/Messaging/Support/FrameworkHeaders.php +++ b/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_TO_CHANNEL], ); return $metadata; diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php index a202f4b894..9e7df8a466 100644 --- a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php @@ -8,6 +8,7 @@ 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 +17,8 @@ */ class DistributedMessageHandler { + public const REPLY_CHANNEL = 'ecotone.distributed.reply'; + public function __construct( private BusRoutingMap $distributedEventHandlerRoutingKeys, private BusRoutingMap $distributedCommandHandlerRoutingKeys, @@ -34,6 +37,8 @@ public function handle( string $contentType, #[Header(DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME)] ?string $targetedServiceName, + #[Header(DistributedBusHeader::DISTRIBUTED_REPLY_TO_CHANNEL)] + ?string $replyToChannelName, CommandBus $commandBus, EventBus $eventBus, MessagingEntrypointService $messagingEntrypoint @@ -57,9 +62,27 @@ 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); + $result = $messagingEntrypoint->sendWithHeaders($payload, $metadata, $routingKey); + + if ($replyToChannelName !== null && $result !== null) { + $messagingEntrypoint->sendWithHeaders($result, $this->replyHeaders($metadata, $replyToChannelName), self::REPLY_CHANNEL); + } } else { throw InvalidArgumentException::create("Trying to call distributed command handler for payload type {$payloadType} and allowed are event/command/message"); } } + + private function replyHeaders(array $requestMetadata, string $replyToChannelName): array + { + return [ + ...FrameworkHeaders::unsetAllFrameworkHeaders($requestMetadata), + MessageHeaders::MESSAGE_CORRELATION_ID => $requestMetadata[MessageHeaders::MESSAGE_CORRELATION_ID], + MessageHeaders::PARENT_MESSAGE_ID => $requestMetadata[MessageHeaders::MESSAGE_ID], + MessageHeaders::ROUTING_SLIP => DistributedBusHeader::DISTRIBUTED_ROUTING_SLIP_VALUE, + DistributedBusHeader::DISTRIBUTED_PAYLOAD_TYPE => 'message', + DistributedBusHeader::DISTRIBUTED_ROUTING_KEY => $replyToChannelName, + DistributedBusHeader::DISTRIBUTED_SOURCE_SERVICE_NAME => $this->thisServiceName, + DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME => $requestMetadata[DistributedBusHeader::DISTRIBUTED_SOURCE_SERVICE_NAME] ?? null, + ]; + } } diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php new file mode 100644 index 0000000000..a98d0f2d3b --- /dev/null +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php @@ -0,0 +1,59 @@ +. + * 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 $serviceMapsByOutboundChannel + */ + public function __construct( + private array $serviceMapsByOutboundChannel, + private string $thisServiceName, + ) { + } + + /** + * @return array + */ + public function route( + #[Header(DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME)] + ?string $targetedServiceName, + #[Header(DistributedBusHeader::DISTRIBUTED_ROUTING_KEY)] + string $replyToChannelName, + ): array { + foreach ($this->serviceMapsByOutboundChannel as $outboundChannel => $serviceMap) { + if (array_key_exists($targetedServiceName, $serviceMap->commandMapping)) { + return [$outboundChannel]; + } + } + + return []; + } +} diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php index e47103bda3..53a642cdfa 100644 --- a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php @@ -30,7 +30,9 @@ use Ecotone\Messaging\Handler\Router\RouterBuilder; use Ecotone\Messaging\Support\Assert; use Ecotone\Messaging\Support\LicensingException; +use Ecotone\Modelling\MessageHandling\Distribution\DistributedMessageHandler; use Ecotone\Modelling\MessageHandling\Distribution\DistributedOutboundRouter; +use Ecotone\Modelling\MessageHandling\Distribution\DistributedReplyRouter; #[ModuleAnnotation] /** @@ -65,6 +67,8 @@ public function prepare(Configuration $messagingConfiguration, array $extensionO $messageChannels = ExtensionObjectResolver::resolve(MessageChannelBuilder::class, $extensionObjects); $applicationConfiguration = ExtensionObjectResolver::resolveUnique(ServiceConfiguration::class, $extensionObjects, ServiceConfiguration::createWithDefaults()); + $this->registerReplyRouter($applicationConfiguration, $messagingConfiguration, $distributedServiceMaps, $interfaceToCallRegistry); + foreach ($distributedServiceMaps as $distributedServiceMap) { if (! $messagingConfiguration->isRunningForEnterpriseLicence()) { throw LicensingException::create('Distributed Bus with Service Map is available only as part of Ecotone Enterprise. See https://docs.ecotone.tech/enterprise for details.'); @@ -94,7 +98,7 @@ private function registerDistributedBus(ServiceConfiguration $applicationConfigu { Assert::isFalse($applicationConfiguration->serviceName === ServiceConfiguration::DEFAULT_SERVICE_NAME, 'Service name must be provided in ServiceConfiguration to make use of Distributed Bus', true); - $distributedChannelName = $distributedServiceMap->getReferenceName() . '_outboundDistribution'; + $distributedChannelName = self::outboundDistributionChannelFor($distributedServiceMap); $outboundRouterReference = $distributedServiceMap->getReferenceName() . '_' . DistributedOutboundRouter::class; $configuration->registerServiceDefinition( @@ -193,6 +197,7 @@ private function registerDistributedBus(ServiceConfiguration $applicationConfigu GatewayHeaderBuilder::create('sourceMediaType', MessageHeaders::CONTENT_TYPE), GatewayHeaderBuilder::create('targetChannelName', DistributedBusHeader::DISTRIBUTED_ROUTING_KEY), GatewayHeaderBuilder::create('targetServiceName', DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME), + GatewayHeaderBuilder::create('replyToChannelName', DistributedBusHeader::DISTRIBUTED_REPLY_TO_CHANNEL), GatewayHeaderValueBuilder::create(DistributedBusHeader::DISTRIBUTED_SOURCE_SERVICE_NAME, $applicationConfiguration->serviceName), GatewayHeaderValueBuilder::create(DistributedBusHeader::DISTRIBUTED_PAYLOAD_TYPE, 'message'), GatewayHeaderValueBuilder::create(MessageHeaders::ROUTING_SLIP, DistributedBusHeader::DISTRIBUTED_ROUTING_SLIP_VALUE), @@ -202,6 +207,35 @@ private function registerDistributedBus(ServiceConfiguration $applicationConfigu ); } + /** + * @param DistributedServiceMap[] $distributedServiceMaps + */ + private function registerReplyRouter(ServiceConfiguration $applicationConfiguration, Configuration $configuration, array $distributedServiceMaps, InterfaceToCallRegistry $interfaceToCallRegistry): void + { + $serviceMapsByOutboundChannel = []; + foreach ($distributedServiceMaps as $distributedServiceMap) { + $serviceMapsByOutboundChannel[self::outboundDistributionChannelFor($distributedServiceMap)] = $distributedServiceMap; + } + + $configuration->registerServiceDefinition( + DistributedReplyRouter::class, + Definition::createFor(DistributedReplyRouter::class, [$serviceMapsByOutboundChannel, $applicationConfiguration->serviceName]) + ); + $configuration->registerMessageHandler( + RouterBuilder::create(DistributedReplyRouter::class, $interfaceToCallRegistry->getFor(DistributedReplyRouter::class, 'route')) + ->withInputChannelName(DistributedMessageHandler::REPLY_CHANNEL) + ->withMethodParameterConverters([ + HeaderBuilder::createOptional('targetedServiceName', DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME), + HeaderBuilder::create('replyToChannelName', DistributedBusHeader::DISTRIBUTED_ROUTING_KEY), + ]) + ); + } + + private static function outboundDistributionChannelFor(DistributedServiceMap $distributedServiceMap): string + { + return $distributedServiceMap->getReferenceName() . '_outboundDistribution'; + } + public function isMessageChannelAvailable(array $messageChannels, mixed $channelName): bool { foreach ($messageChannels as $messageChannel) { diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedHandlerModule.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedHandlerModule.php index 1b757f9174..a6081ce0bd 100644 --- a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedHandlerModule.php +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedHandlerModule.php @@ -116,6 +116,7 @@ public function prepare(Configuration $messagingConfiguration, array $extensionO HeaderBuilder::create('routingKey', DistributedBusHeader::DISTRIBUTED_ROUTING_KEY), HeaderBuilder::create('contentType', MessageHeaders::CONTENT_TYPE), HeaderBuilder::createOptional('targetedServiceName', DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME), + HeaderBuilder::createOptional('replyToChannelName', DistributedBusHeader::DISTRIBUTED_REPLY_TO_CHANNEL), ReferenceBuilder::create('commandBus', CommandBus::class), ReferenceBuilder::create('eventBus', EventBus::class), ReferenceBuilder::create('messagingEntrypoint', MessagingEntrypointService::class), diff --git a/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php b/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php new file mode 100644 index 0000000000..8a556a04ab --- /dev/null +++ b/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php @@ -0,0 +1,80 @@ +. + * 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 DistributedBusMessageReplyTest extends TestCase +{ + public function test_the_result_of_a_distributed_message_triggers_the_reply_to_channel_in_the_sending_service(): void + { + $channels = [SimpleMessageChannelBuilder::createQueueChannel('ticket_channel'), SimpleMessageChannelBuilder::createQueueChannel('user_channel')]; + $tickets = new class () { + #[Distributed] + #[InternalHandler('ticket.create')] + public function create(string $description): string + { + return 'ticket for: ' . $description; + } + }; + $users = new class () { + public array $replies = []; + + #[InternalHandler('ticket.created')] + public function created(string $ticket): void + { + $this->replies[] = $ticket; + } + }; + $userService = $this->service('user_service', $users, $channels, DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'ticket_service', channelName: 'ticket_channel')); + $ticketService = $this->service('ticket_service', $tickets, $channels, DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel')); + + $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); + $userService->run('user_channel', ExecutionPollingMetadata::createWithTestingSetup()); + + self::assertSame(['ticket for: Printer is broken'], $users->replies); + } + + private function service(string $serviceName, object $handlers, array $channels, DistributedServiceMap ...$serviceMaps): FlowTestSupport + { + return EcotoneLite::bootstrapFlowTesting( + classesToResolve: [$handlers::class], + containerOrAvailableServices: [$handlers], + extensionObjects: [...$channels, ...$serviceMaps, ServiceConfiguration::createWithDefaults()->withServiceName($serviceName)], + enabledModules: [], + licenceKey: LicenceTesting::VALID_LICENCE, + ); + } +} From 83c92fa69cd718fe1b641620461a71e0995c108b Mon Sep 17 00:00:00 2001 From: Dariusz Gafka Date: Tue, 20 Oct 2026 17:25:00 +0200 Subject: [PATCH 02/12] feat(distribution): refuse a reply no Service Map can route, and prove what a reply carries --- .../Distribution/DistributedReplyRouter.php | 28 +- .../DistributedBusMessageReplyTest.php | 291 ++++++++++++++++++ 2 files changed, 317 insertions(+), 2 deletions(-) diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php index a98d0f2d3b..d047924d2c 100644 --- a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php @@ -7,8 +7,14 @@ use Ecotone\Api\Attribute\Header; use Ecotone\Api\ExtensionObject\DistributedServiceMap; use Ecotone\Api\Gateway\DistributedBusHeader; +use Ecotone\Messaging\Config\ConfigurationException; use function array_key_exists; +use function array_keys; +use function array_map; +use function array_values; +use function count; +use function implode; /** * licence Enterprise @@ -48,12 +54,30 @@ public function route( #[Header(DistributedBusHeader::DISTRIBUTED_ROUTING_KEY)] string $replyToChannelName, ): array { + if ($targetedServiceName === null) { + throw UnknownDistributedDestination::create("A reply to channel '{$replyToChannelName}' 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)) { - return [$outboundChannel]; + $channel = $serviceMap->commandMapping[$targetedServiceName]; + $channelsByServiceMap[$serviceMap->getReferenceName()] = $channel; + $outboundChannelsByChannel[$channel] ??= $outboundChannel; } } - return []; + if ($outboundChannelsByChannel === []) { + throw UnknownDistributedDestination::create("A reply to channel '{$replyToChannelName}' 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 replyToChannelName."); + } + + 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 channel '{$replyToChannelName}' 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); } } diff --git a/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php b/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php index 8a556a04ab..e9510b861f 100644 --- a/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php +++ b/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php @@ -5,14 +5,20 @@ namespace Test\Ecotone\Messaging\Unit\Distributed; use Ecotone\Api\Attribute\Distributed; +use Ecotone\Api\Attribute\Header; use Ecotone\Api\Attribute\InternalHandler; use Ecotone\Api\ExtensionObject\DistributedServiceMap; use Ecotone\Api\ExtensionObject\ExecutionPollingMetadata; +use Ecotone\Api\ExtensionObject\MediaType; use Ecotone\Api\ExtensionObject\ServiceConfiguration; use Ecotone\Api\ExtensionObject\SimpleMessageChannelBuilder; use Ecotone\Api\Lite\EcotoneLite; +use Ecotone\Api\Gateway\DistributedBusHeader; use Ecotone\Api\Lite\Test\FlowTestSupport; +use Ecotone\Api\Messaging\MessageHeaders; +use Ecotone\Test\ChannelTestSupport; use Ecotone\Test\LicenceTesting; +use Ecotone\Test\StaticMessageBuilder; use PHPUnit\Framework\TestCase; /** @@ -67,6 +73,284 @@ public function created(string $ticket): void self::assertSame(['ticket for: Printer is broken'], $users->replies); } + public function test_the_reply_carries_the_request_metadata_and_correlation_id(): void + { + $channels = $this->channels(); + $tickets = new class () { + public ?string $correlationId = null; + + #[Distributed] + #[InternalHandler('ticket.create')] + public function create(string $description, #[Header(MessageHeaders::MESSAGE_CORRELATION_ID)] string $correlationId): string + { + $this->correlationId = $correlationId; + + return 'ticket for: ' . $description; + } + }; + $users = new class () { + public array $replies = []; + + #[InternalHandler('ticket.created')] + public function created(string $ticket, #[Header('requestedBy')] string $requestedBy, #[Header(MessageHeaders::MESSAGE_CORRELATION_ID)] string $correlationId): void + { + $this->replies[] = [$ticket, $requestedBy, $correlationId]; + } + }; + $userService = $this->userService($users, $channels); + $ticketService = $this->ticketService($tickets, $channels); + + $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', metadata: ['requestedBy' => 'johny'], replyToChannelName: 'ticket.created'); + $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); + $userService->run('user_channel', ExecutionPollingMetadata::createWithTestingSetup()); + + self::assertSame([['ticket for: Printer is broken', 'johny', $tickets->correlationId]], $users->replies); + } + + public function test_a_handler_returning_null_sends_no_reply(): void + { + $channels = $this->channels(); + $tickets = new class () { + #[Distributed] + #[InternalHandler('ticket.create')] + public function create(string $description): ?string + { + return null; + } + }; + $users = new class () { + public array $replies = []; + + #[InternalHandler('ticket.created')] + public function created(string $ticket): void + { + $this->replies[] = $ticket; + } + }; + $userService = $this->userService($users, $channels); + $ticketService = $this->ticketService($tickets, $channels); + + $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); + $userService->run('user_channel', ExecutionPollingMetadata::createWithTestingSetup()); + + self::assertSame([], $users->replies); + } + + public function test_a_reply_is_not_answered_again_even_when_the_reply_handler_returns_a_value(): void + { + $channels = $this->channels(); + $tickets = new class () { + public int $created = 0; + + #[Distributed] + #[InternalHandler('ticket.create')] + public function create(string $description): string + { + $this->created++; + + return 'ticket for: ' . $description; + } + }; + $users = new class () { + public array $replies = []; + + #[Distributed] + #[InternalHandler('ticket.created')] + public function created(string $ticket): string + { + $this->replies[] = $ticket; + + return 'acknowledged'; + } + }; + $userService = $this->userService($users, $channels); + $ticketService = $this->ticketService($tickets, $channels); + + $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); + $userService->run('user_channel', ExecutionPollingMetadata::createWithTestingSetup()); + $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); + $userService->run('user_channel', ExecutionPollingMetadata::createWithTestingSetup()); + + self::assertSame(1, $tickets->created); + self::assertSame(['ticket for: Printer is broken'], $users->replies); + } + + public function test_an_object_result_reaches_the_reply_to_channel_as_the_same_object(): void + { + $channels = $this->channels(); + $tickets = new class () { + #[Distributed] + #[InternalHandler('ticket.create')] + public function create(string $description): TicketWasCreated + { + return new TicketWasCreated('ticket-1', $description); + } + }; + $users = new class () { + public array $replies = []; + + #[InternalHandler('ticket.created')] + public function created(TicketWasCreated $ticket): void + { + $this->replies[] = $ticket; + } + }; + $userService = $this->userService($users, $channels); + $ticketService = $this->ticketService($tickets, $channels); + + $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); + $userService->run('user_channel', ExecutionPollingMetadata::createWithTestingSetup()); + + self::assertEquals([new TicketWasCreated('ticket-1', 'Printer is broken')], $users->replies); + } + + public function test_sending_a_message_without_reply_to_channel_sends_no_reply(): void + { + $channels = $this->channels(); + $tickets = new class () { + public array $descriptions = []; + + #[Distributed] + #[InternalHandler('ticket.create')] + public function create(string $description): string + { + $this->descriptions[] = $description; + + return 'ticket for: ' . $description; + } + }; + $users = new class () { + public array $replies = []; + + #[InternalHandler('ticket.created')] + public function created(string $ticket): void + { + $this->replies[] = $ticket; + } + }; + $userService = $this->userService($users, $channels); + $ticketService = $this->ticketService($tickets, $channels); + + $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken'); + $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); + $userService->run('user_channel', ExecutionPollingMetadata::createWithTestingSetup()); + + self::assertSame(['Printer is broken'], $tickets->descriptions); + self::assertSame([], $users->replies); + } + + public function test_two_service_maps_mapping_the_sender_to_the_same_channel_send_one_reply(): void + { + $channels = $this->channels(); + $users = new class () { + public array $replies = []; + + #[InternalHandler('ticket.created')] + public function created(string $ticket): void + { + $this->replies[] = $ticket; + } + }; + $userService = $this->userService($users, $channels); + $ticketService = $this->service( + 'ticket_service', + $this->ticketsReturningDescription(), + $channels, + DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel'), + DistributedServiceMap::initialize('second_distributed_bus')->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel'), + ); + + $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); + $userService->run('user_channel', ExecutionPollingMetadata::createWithTestingSetup()); + + self::assertSame(['ticket for: Printer is broken'], $users->replies); + } + + public function test_a_reply_is_refused_when_no_service_map_of_the_receiver_maps_the_sender(): void + { + $channels = $this->channels(); + $userService = $this->userService(new class () { + }, $channels); + $ticketService = $this->service('ticket_service', $this->ticketsReturningDescription(), $channels); + + $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + + $this->expectExceptionMessage("A reply to channel 'ticket.created' of service 'user_service' cannot be sent, because no Service Map in service 'ticket_service' maps service 'user_service' to a channel. Register the channel 'user_service' consumes with DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'user_service', channelName: '...') in service 'ticket_service', or send the message without replyToChannelName."); + + $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); + } + + public function test_a_reply_is_refused_when_two_service_maps_of_the_receiver_map_the_sender_to_different_channels(): void + { + $channels = [...$this->channels(), SimpleMessageChannelBuilder::createQueueChannel('user_channel_2')]; + $userService = $this->userService(new class () { + }, $channels); + $ticketService = $this->service( + 'ticket_service', + $this->ticketsReturningDescription(), + $channels, + DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel'), + DistributedServiceMap::initialize('second_distributed_bus')->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel_2'), + ); + + $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + + $this->expectExceptionMessage("A reply to channel 'ticket.created' of service 'user_service' cannot be sent, because Service Maps in service 'ticket_service' map service 'user_service' to different channels: 'Ecotone\\Api\\Gateway\\DistributedBus' to 'user_channel', 'second_distributed_bus' to 'user_channel_2'. Map service 'user_service' in one Service Map only, or to the same channel in each."); + + $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); + } + + public function test_a_reply_is_refused_when_the_message_names_no_source_service(): void + { + $channels = $this->channels(); + $ticketService = $this->ticketService($this->ticketsReturningDescription(), $channels); + ChannelTestSupport::of($ticketService)->getMessageChannel('ticket_channel')->send( + StaticMessageBuilder::withPayload('Printer is broken') + ->setHeader(MessageHeaders::ROUTING_SLIP, DistributedBusHeader::DISTRIBUTED_ROUTING_SLIP_VALUE) + ->setHeader(DistributedBusHeader::DISTRIBUTED_ROUTING_KEY, 'ticket.create') + ->setHeader(DistributedBusHeader::DISTRIBUTED_PAYLOAD_TYPE, 'message') + ->setHeader(DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME, 'ticket_service') + ->setHeader(DistributedBusHeader::DISTRIBUTED_REPLY_TO_CHANNEL, 'ticket.created') + ->setHeader(MessageHeaders::CONTENT_TYPE, MediaType::TEXT_PLAIN) + ->build() + ); + + $this->expectExceptionMessage("A reply to channel 'ticket.created' 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."); + + $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); + } + + private function ticketsReturningDescription(): object + { + return new class () { + #[Distributed] + #[InternalHandler('ticket.create')] + public function create(string $description): string + { + return 'ticket for: ' . $description; + } + }; + } + + private function channels(): array + { + return [SimpleMessageChannelBuilder::createQueueChannel('ticket_channel'), SimpleMessageChannelBuilder::createQueueChannel('user_channel')]; + } + + private function userService(object $handlers, array $channels): FlowTestSupport + { + return $this->service('user_service', $handlers, $channels, DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'ticket_service', channelName: 'ticket_channel')); + } + + private function ticketService(object $handlers, array $channels): FlowTestSupport + { + return $this->service('ticket_service', $handlers, $channels, DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel')); + } + private function service(string $serviceName, object $handlers, array $channels, DistributedServiceMap ...$serviceMaps): FlowTestSupport { return EcotoneLite::bootstrapFlowTesting( @@ -78,3 +362,10 @@ classesToResolve: [$handlers::class], ); } } + +final class TicketWasCreated +{ + public function __construct(public readonly string $ticketId, public readonly string $description) + { + } +} From 0fe1d414eff55c8f85020d4d2c8e369715073fe9 Mon Sep 17 00:00:00 2001 From: Dariusz Gafka Date: Tue, 20 Oct 2026 17:40:00 +0200 Subject: [PATCH 03/12] feat(flow): draw distributed internal handlers and their replies, and prove a reply-to flow on every flow suite --- .../Config/Flow/MessageFlowDiscovery.php | 14 +++++ .../Test/Flow/DistributedFlowScenarios.php | 55 +++++++++++++++++++ 2 files changed, 69 insertions(+) diff --git a/packages/Ecotone/src/Messaging/Config/Flow/MessageFlowDiscovery.php b/packages/Ecotone/src/Messaging/Config/Flow/MessageFlowDiscovery.php index 5aace82c98..23006719b7 100644 --- a/packages/Ecotone/src/Messaging/Config/Flow/MessageFlowDiscovery.php +++ b/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' => 'replyToChannelName']); + } + } + private function discoverRouters(): void { foreach ($this->applicationMethods(Router::class) as $registration) { diff --git a/packages/Ecotone/src/Test/Flow/DistributedFlowScenarios.php b/packages/Ecotone/src/Test/Flow/DistributedFlowScenarios.php index 507deed261..6e5d2fe70e 100644 --- a/packages/Ecotone/src/Test/Flow/DistributedFlowScenarios.php +++ b/packages/Ecotone/src/Test/Flow/DistributedFlowScenarios.php @@ -7,6 +7,7 @@ use Ecotone\Api\Attribute\CommandHandler; use Ecotone\Api\Attribute\Distributed; use Ecotone\Api\Attribute\EventHandler; +use Ecotone\Api\Attribute\InternalHandler; use Ecotone\Api\ExtensionObject\DistributedServiceMap; use Ecotone\Api\ExtensionObject\ExecutionPollingMetadata; use Ecotone\Api\ExtensionObject\ServiceConfiguration; @@ -28,6 +29,7 @@ public static function all(): array return [ self::distributedCommandConsumedByTheTargetService(), self::distributedEventDeliveredToEverySubscribedChannel(), + self::distributedMessageAnsweredOnItsReplyToChannel(), ]; } @@ -115,4 +117,57 @@ public function update(string $details): void ], ); } + + private static function distributedMessageAnsweredOnItsReplyToChannel(): FlowScenario + { + $tickets = new class () { + #[Distributed] + #[InternalHandler('ticket.create', endpointId: 'ticket.creation')] + public function create(string $description): string + { + return 'ticket for: ' . $description; + } + + #[Distributed] + #[InternalHandler('ticket.created', endpointId: 'ticket.confirmation')] + public function confirm(string $ticket): void + { + } + }; + + return new FlowScenario( + 'distributed message answered on its reply-to channel', + [$tickets::class], + [$tickets], + [ + SimpleMessageChannelBuilder::createQueueChannel('distributed_channel'), + DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'ticket_service', channelName: 'distributed_channel'), + ServiceConfiguration::createWithDefaults()->withServiceName('ticket_service'), + ], + [ + FlowStage::when(fn (FlowTestSupport $ecotone, array $metadata) => $ecotone->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', metadata: $metadata, replyToChannelName: 'ticket.created'))->leavingWorkPending() + ->expect(ExpectedSpan::named('send distributed')->ofKind(SpanKind::Producer)->containing(ExpectedSpan::named('send distributed_channel')->ofKind(SpanKind::Producer))) + ->withNoOtherSpans(), + FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->run('distributed_channel', ExecutionPollingMetadata::createWithTestingSetup(handledMessageLimit: 1)))->leavingWorkPending() + ->expect( + ExpectedSpan::named('receive distributed_channel')->ofKind(SpanKind::Consumer)->containing( + ExpectedSpan::named('process class@anonymous::create')->withAttributes([SpanAttribute::ENDPOINT_ID => 'ticket.creation']), + ExpectedSpan::named('send distributed_channel')->ofKind(SpanKind::Producer), + ), + ) + ->withNoOtherSpans(), + FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->run('distributed_channel', ExecutionPollingMetadata::createWithTestingSetup(handledMessageLimit: 1))) + ->expect( + ExpectedSpan::named('receive distributed_channel')->ofKind(SpanKind::Consumer)->containing( + ExpectedSpan::named('process class@anonymous::create')->withAttributes([SpanAttribute::ENDPOINT_ID => 'ticket.creation']), + ExpectedSpan::named('send distributed_channel')->ofKind(SpanKind::Producer)->containing( + ExpectedSpan::named('receive distributed_channel')->ofKind(SpanKind::Consumer)->containing( + ExpectedSpan::named('process class@anonymous::confirm')->withAttributes([SpanAttribute::ENDPOINT_ID => 'ticket.confirmation']), + ), + ), + ), + ), + ], + ); + } } From 273059a9943ca3df3c72fc4d9a8761e47ab98b9f Mon Sep 17 00:00:00 2001 From: Dariusz Gafka Date: Tue, 20 Oct 2026 17:50:00 +0200 Subject: [PATCH 04/12] test(dbal): prove a distributed message reply over DBAL channels --- .../DbalDistributedMessageReplyTest.php | 94 +++++++++++++++++++ 1 file changed, 94 insertions(+) create mode 100644 packages/Dbal/tests/Integration/DbalDistributedMessageReplyTest.php diff --git a/packages/Dbal/tests/Integration/DbalDistributedMessageReplyTest.php b/packages/Dbal/tests/Integration/DbalDistributedMessageReplyTest.php new file mode 100644 index 0000000000..4e60f40940 --- /dev/null +++ b/packages/Dbal/tests/Integration/DbalDistributedMessageReplyTest.php @@ -0,0 +1,94 @@ +. + * 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_to_channel_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')); + $ticketService = $this->service('ticket_service', $tickets, DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel')); + + $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', metadata: ['requestedBy' => 'johny'], replyToChannelName: '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); + } + + private function service(string $serviceName, object $handlers, DistributedServiceMap $serviceMap): FlowTestSupport + { + return $this->bootstrapFlowTesting( + classesToResolve: [$handlers::class], + containerOrAvailableServices: [$handlers, DbalConnectionFactory::class => $this->getConnectionFactory()], + 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) + { + } +} From f0007ada9fc8b19f1a6556c6e3106072824989ba Mon Sep 17 00:00:00 2001 From: Dariusz Gafka Date: Tue, 20 Oct 2026 17:55:00 +0200 Subject: [PATCH 05/12] style(distribution): apply php-cs-fixer --- .../Distribution/DistributedReplyRouter.php | 11 ++++++----- .../Distributed/DistributedBusMessageReplyTest.php | 2 +- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php index d047924d2c..346eb39afd 100644 --- a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php @@ -4,16 +4,17 @@ namespace Ecotone\Modelling\MessageHandling\Distribution; -use Ecotone\Api\Attribute\Header; -use Ecotone\Api\ExtensionObject\DistributedServiceMap; -use Ecotone\Api\Gateway\DistributedBusHeader; -use Ecotone\Messaging\Config\ConfigurationException; - 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; /** diff --git a/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php b/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php index e9510b861f..6d7352d022 100644 --- a/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php +++ b/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php @@ -12,8 +12,8 @@ use Ecotone\Api\ExtensionObject\MediaType; use Ecotone\Api\ExtensionObject\ServiceConfiguration; use Ecotone\Api\ExtensionObject\SimpleMessageChannelBuilder; -use Ecotone\Api\Lite\EcotoneLite; use Ecotone\Api\Gateway\DistributedBusHeader; +use Ecotone\Api\Lite\EcotoneLite; use Ecotone\Api\Lite\Test\FlowTestSupport; use Ecotone\Api\Messaging\MessageHeaders; use Ecotone\Test\ChannelTestSupport; From 39d9fdaf6931c57c4c69f159e8cf42aff3e92ff2 Mon Sep 17 00:00:00 2001 From: Dariusz Gafka Date: Tue, 20 Oct 2026 18:05:00 +0200 Subject: [PATCH 06/12] fix(distribution): reply from an asynchronous target by carrying the reply on the routing slip --- .../DistributedMessageHandler.php | 27 ++++++++------ .../DistributedBusWithServiceMapModule.php | 2 +- .../Module/DistributedHandlerModule.php | 10 ++++++ .../DistributedBusMessageReplyTest.php | 35 +++++++++++++++++++ 4 files changed, 63 insertions(+), 11 deletions(-) diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php index 9e7df8a466..c1418bff59 100644 --- a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php @@ -18,6 +18,7 @@ class DistributedMessageHandler { public const REPLY_CHANNEL = 'ecotone.distributed.reply'; + public const REPLY_ROUTING_CHANNEL = 'ecotone.distributed.reply.route'; public function __construct( private BusRoutingMap $distributedEventHandlerRoutingKeys, @@ -62,27 +63,33 @@ public function handle( throw InvalidArgumentException::create("Received command message which targets {$targetedServiceName} Service, but consuming in {$this->thisServiceName}. Message was wrongly distributed."); } - $result = $messagingEntrypoint->sendWithHeaders($payload, $metadata, $routingKey); - - if ($replyToChannelName !== null && $result !== null) { - $messagingEntrypoint->sendWithHeaders($result, $this->replyHeaders($metadata, $replyToChannelName), self::REPLY_CHANNEL); - } + $messagingEntrypoint->sendWithHeaders($payload, $metadata, $routingKey, $replyToChannelName === 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"); } } - private function replyHeaders(array $requestMetadata, string $replyToChannelName): array + public function reply( + mixed $result, + array $metadata, + #[Header(DistributedBusHeader::DISTRIBUTED_REPLY_TO_CHANNEL)] + string $replyToChannelName, + MessagingEntrypointService $messagingEntrypoint + ): void { + $messagingEntrypoint->sendWithHeaders($result, $this->replyHeaders($metadata, $replyToChannelName), self::REPLY_ROUTING_CHANNEL); + } + + private function replyHeaders(array $resultMetadata, string $replyToChannelName): array { return [ - ...FrameworkHeaders::unsetAllFrameworkHeaders($requestMetadata), - MessageHeaders::MESSAGE_CORRELATION_ID => $requestMetadata[MessageHeaders::MESSAGE_CORRELATION_ID], - MessageHeaders::PARENT_MESSAGE_ID => $requestMetadata[MessageHeaders::MESSAGE_ID], + ...FrameworkHeaders::unsetAllFrameworkHeaders($resultMetadata), + 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 => $replyToChannelName, DistributedBusHeader::DISTRIBUTED_SOURCE_SERVICE_NAME => $this->thisServiceName, - DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME => $requestMetadata[DistributedBusHeader::DISTRIBUTED_SOURCE_SERVICE_NAME] ?? null, + DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME => $resultMetadata[DistributedBusHeader::DISTRIBUTED_SOURCE_SERVICE_NAME] ?? null, ]; } } diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php index 53a642cdfa..ecab7ffe84 100644 --- a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php @@ -223,7 +223,7 @@ private function registerReplyRouter(ServiceConfiguration $applicationConfigurat ); $configuration->registerMessageHandler( RouterBuilder::create(DistributedReplyRouter::class, $interfaceToCallRegistry->getFor(DistributedReplyRouter::class, 'route')) - ->withInputChannelName(DistributedMessageHandler::REPLY_CHANNEL) + ->withInputChannelName(DistributedMessageHandler::REPLY_ROUTING_CHANNEL) ->withMethodParameterConverters([ HeaderBuilder::createOptional('targetedServiceName', DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME), HeaderBuilder::create('replyToChannelName', DistributedBusHeader::DISTRIBUTED_ROUTING_KEY), diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedHandlerModule.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedHandlerModule.php index a6081ce0bd..d16b33219e 100644 --- a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedHandlerModule.php +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedHandlerModule.php @@ -122,6 +122,16 @@ public function prepare(Configuration $messagingConfiguration, array $extensionO ReferenceBuilder::create('messagingEntrypoint', MessagingEntrypointService::class), ]) ); + $messagingConfiguration->registerMessageHandler( + MessageProcessorActivatorBuilder::createForMethod(DistributedMessageHandler::class, $interfaceToCallRegistry->getFor(DistributedMessageHandler::class, 'reply')) + ->withInputChannelName(DistributedMessageHandler::REPLY_CHANNEL) + ->withMethodParameterConverters([ + PayloadBuilder::create('result'), + AllHeadersBuilder::createWith('metadata'), + HeaderBuilder::create('replyToChannelName', DistributedBusHeader::DISTRIBUTED_REPLY_TO_CHANNEL), + ReferenceBuilder::create('messagingEntrypoint', MessagingEntrypointService::class), + ]) + ); } public function getModulePackageName(): string diff --git a/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php b/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php index 6d7352d022..7257ba682a 100644 --- a/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php +++ b/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php @@ -4,6 +4,7 @@ namespace Test\Ecotone\Messaging\Unit\Distributed; +use Ecotone\Api\Attribute\Asynchronous; use Ecotone\Api\Attribute\Distributed; use Ecotone\Api\Attribute\Header; use Ecotone\Api\Attribute\InternalHandler; @@ -73,6 +74,40 @@ public function created(string $ticket): void self::assertSame(['ticket for: Printer is broken'], $users->replies); } + public function test_an_asynchronous_target_replies_once_its_work_is_consumed(): void + { + $channels = [...$this->channels(), SimpleMessageChannelBuilder::createQueueChannel('ticket_work')]; + $tickets = new class () { + #[Asynchronous('ticket_work')] + #[Distributed] + #[InternalHandler('ticket.create', endpointId: 'ticket.creation')] + public function create(string $description): string + { + return 'ticket for: ' . $description; + } + }; + $users = new class () { + public array $replies = []; + + #[InternalHandler('ticket.created')] + public function created(string $ticket): void + { + $this->replies[] = $ticket; + } + }; + $userService = $this->userService($users, $channels); + $ticketService = $this->ticketService($tickets, $channels); + + $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); + $userService->run('user_channel', ExecutionPollingMetadata::createWithTestingSetup()); + self::assertSame([], $users->replies); + + $ticketService->run('ticket_work', ExecutionPollingMetadata::createWithTestingSetup()); + $userService->run('user_channel', ExecutionPollingMetadata::createWithTestingSetup()); + self::assertSame(['ticket for: Printer is broken'], $users->replies); + } + public function test_the_reply_carries_the_request_metadata_and_correlation_id(): void { $channels = $this->channels(); From 847eec7d0cf0d5f8ef2e705c92056f193626e26d Mon Sep 17 00:00:00 2001 From: Dariusz Gafka Date: Tue, 20 Oct 2026 18:15:00 +0200 Subject: [PATCH 07/12] fix(distribution): continue the sender's trace on a reply from a receiver without tracing --- .../DistributedMessageHandler.php | 4 ++ .../DistributedBusMessageReplyTest.php | 39 +++++++++++++++++++ 2 files changed, 43 insertions(+) diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php index c1418bff59..2371c33b60 100644 --- a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php @@ -12,6 +12,9 @@ use Ecotone\Messaging\Support\InvalidArgumentException; use Ecotone\Modelling\Config\Routing\BusRoutingMap; +use function array_flip; +use function array_intersect_key; + /** * licence Apache-2.0 */ @@ -83,6 +86,7 @@ private function replyHeaders(array $resultMetadata, string $replyToChannelName) { 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, diff --git a/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php b/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php index 7257ba682a..97c33c43cc 100644 --- a/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php +++ b/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php @@ -17,6 +17,9 @@ use Ecotone\Api\Lite\EcotoneLite; use Ecotone\Api\Lite\Test\FlowTestSupport; use Ecotone\Api\Messaging\MessageHeaders; +use Ecotone\Api\Tracing\InMemorySpanRecorder; +use Ecotone\Api\Tracing\Span; +use Ecotone\Api\Tracing\TracingConfiguration; use Ecotone\Test\ChannelTestSupport; use Ecotone\Test\LicenceTesting; use Ecotone\Test\StaticMessageBuilder; @@ -108,6 +111,42 @@ public function created(string $ticket): void self::assertSame(['ticket for: Printer is broken'], $users->replies); } + public function test_the_reply_continues_the_senders_trace_through_a_receiver_without_tracing(): void + { + $channels = $this->channels(); + $recorder = new InMemorySpanRecorder(); + $users = new class () { + public array $replies = []; + + #[InternalHandler('ticket.created')] + public function created(string $ticket): void + { + $this->replies[] = $ticket; + } + }; + $userService = EcotoneLite::bootstrapFlowTesting( + classesToResolve: [$users::class], + containerOrAvailableServices: [$users, InMemorySpanRecorder::class => $recorder], + extensionObjects: [ + ...$channels, + DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'ticket_service', channelName: 'ticket_channel'), + ServiceConfiguration::createWithDefaults()->withServiceName('user_service'), + TracingConfiguration::createWithDefaults()->withSpanRecorder(InMemorySpanRecorder::class), + ], + enabledModules: [], + licenceKey: LicenceTesting::VALID_LICENCE, + ); + $ticketService = $this->ticketService($this->ticketsReturningDescription(), $channels); + + $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); + $userService->run('user_channel', ExecutionPollingMetadata::createWithTestingSetup()); + + self::assertSame(['ticket for: Printer is broken'], $users->replies); + self::assertCount(1, array_unique(array_map(fn (Span $span) => $span->traceId, $recorder->recordedSpans()))); + self::assertCount(1, array_filter($recorder->recordedSpans(), fn (Span $span) => $span->isRoot())); + } + public function test_the_reply_carries_the_request_metadata_and_correlation_id(): void { $channels = $this->channels(); From 10c269af0457b5b957ef8f7a25bc83ba425239be Mon Sep 17 00:00:00 2001 From: Dariusz Gafka Date: Tue, 20 Oct 2026 18:25:00 +0200 Subject: [PATCH 08/12] test(dbal): prove a reply that cannot be sent rolls back the handler and the retried request replies once --- .../DbalDistributedMessageReplyTest.php | 92 ++++++++++++++++++- 1 file changed, 88 insertions(+), 4 deletions(-) diff --git a/packages/Dbal/tests/Integration/DbalDistributedMessageReplyTest.php b/packages/Dbal/tests/Integration/DbalDistributedMessageReplyTest.php index 4e60f40940..8bf0a28d42 100644 --- a/packages/Dbal/tests/Integration/DbalDistributedMessageReplyTest.php +++ b/packages/Dbal/tests/Integration/DbalDistributedMessageReplyTest.php @@ -4,6 +4,9 @@ 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; @@ -15,6 +18,7 @@ use Ecotone\Api\ExtensionObject\ServiceConfiguration; use Ecotone\Api\Lite\Test\FlowTestSupport; use Ecotone\Test\LicenceTesting; +use RuntimeException; use Test\Ecotone\Dbal\DbalMessagingTestCase; /** @@ -59,8 +63,8 @@ public function created(DbalTicketWasCreated $ticket, #[Header('requestedBy')] s $this->replies[] = [$ticket, $requestedBy]; } }; - $userService = $this->service('user_service', $users, DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'ticket_service', channelName: 'ticket_channel')); - $ticketService = $this->service('ticket_service', $tickets, DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel')); + $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'], replyToChannelName: 'ticket.created'); $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); @@ -69,11 +73,58 @@ public function created(DbalTicketWasCreated $ticket, #[Header('requestedBy')] s self::assertEquals([[new DbalTicketWasCreated('ticket-1', 'Printer is broken'), 'johny']], $users->replies); } - private function service(string $serviceName, object $handlers, DistributedServiceMap $serviceMap): FlowTestSupport + 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', replyToChannelName: '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 => $this->getConnectionFactory()], + containerOrAvailableServices: [$handlers, DbalConnectionFactory::class => $connectionFactory], extensionObjects: [ DbalBackedMessageChannelBuilder::create('ticket_channel'), DbalBackedMessageChannelBuilder::create('user_channel'), @@ -92,3 +143,36 @@ public function __construct(public readonly string $ticketId, public readonly st { } } + +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'); + } + } +} From 4acf1811b0f7a0d80e639c448bd443bbf4860254 Mon Sep 17 00:00:00 2001 From: Dariusz Gafka Date: Tue, 20 Oct 2026 18:35:00 +0200 Subject: [PATCH 09/12] test(distribution): prove a reply skips the receiver's asynchronous bus channel, and send the source-less request through an application gateway --- .../DistributedBusMessageReplyTest.php | 66 +++++++++++++++---- 1 file changed, 54 insertions(+), 12 deletions(-) diff --git a/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php b/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php index 97c33c43cc..923a7e6f6c 100644 --- a/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php +++ b/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php @@ -7,7 +7,10 @@ use Ecotone\Api\Attribute\Asynchronous; use Ecotone\Api\Attribute\Distributed; use Ecotone\Api\Attribute\Header; +use Ecotone\Api\Attribute\Headers; use Ecotone\Api\Attribute\InternalHandler; +use Ecotone\Api\Attribute\MessageGateway; +use Ecotone\Api\Attribute\Payload; use Ecotone\Api\ExtensionObject\DistributedServiceMap; use Ecotone\Api\ExtensionObject\ExecutionPollingMetadata; use Ecotone\Api\ExtensionObject\MediaType; @@ -20,9 +23,7 @@ use Ecotone\Api\Tracing\InMemorySpanRecorder; use Ecotone\Api\Tracing\Span; use Ecotone\Api\Tracing\TracingConfiguration; -use Ecotone\Test\ChannelTestSupport; use Ecotone\Test\LicenceTesting; -use Ecotone\Test\StaticMessageBuilder; use PHPUnit\Framework\TestCase; /** @@ -344,6 +345,33 @@ public function created(string $ticket): void self::assertSame(['ticket for: Printer is broken'], $users->replies); } + public function test_a_reply_is_sent_directly_even_when_the_receivers_service_map_sends_asynchronously(): void + { + $channels = [...$this->channels(), SimpleMessageChannelBuilder::createQueueChannel('distributed_bus_async')]; + $users = new class () { + public array $replies = []; + + #[InternalHandler('ticket.created')] + public function created(string $ticket): void + { + $this->replies[] = $ticket; + } + }; + $userService = $this->userService($users, $channels); + $ticketService = $this->service( + 'ticket_service', + $this->ticketsReturningDescription(), + $channels, + DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel')->withAsynchronousChannel('distributed_bus_async'), + ); + + $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); + $userService->run('user_channel', ExecutionPollingMetadata::createWithTestingSetup()); + + self::assertSame(['ticket for: Printer is broken'], $users->replies); + } + public function test_a_reply_is_refused_when_no_service_map_of_the_receiver_maps_the_sender(): void { $channels = $this->channels(); @@ -381,17 +409,25 @@ public function test_a_reply_is_refused_when_two_service_maps_of_the_receiver_ma public function test_a_reply_is_refused_when_the_message_names_no_source_service(): void { $channels = $this->channels(); - $ticketService = $this->ticketService($this->ticketsReturningDescription(), $channels); - ChannelTestSupport::of($ticketService)->getMessageChannel('ticket_channel')->send( - StaticMessageBuilder::withPayload('Printer is broken') - ->setHeader(MessageHeaders::ROUTING_SLIP, DistributedBusHeader::DISTRIBUTED_ROUTING_SLIP_VALUE) - ->setHeader(DistributedBusHeader::DISTRIBUTED_ROUTING_KEY, 'ticket.create') - ->setHeader(DistributedBusHeader::DISTRIBUTED_PAYLOAD_TYPE, 'message') - ->setHeader(DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME, 'ticket_service') - ->setHeader(DistributedBusHeader::DISTRIBUTED_REPLY_TO_CHANNEL, 'ticket.created') - ->setHeader(MessageHeaders::CONTENT_TYPE, MediaType::TEXT_PLAIN) - ->build() + $tickets = $this->ticketsReturningDescription(); + $ticketService = EcotoneLite::bootstrapFlowTesting( + classesToResolve: [$tickets::class, ExternalTicketProducer::class], + containerOrAvailableServices: [$tickets], + extensionObjects: [ + ...$channels, + DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel'), + ServiceConfiguration::createWithDefaults()->withServiceName('ticket_service'), + ], + enabledModules: [], + licenceKey: LicenceTesting::VALID_LICENCE, ); + $ticketService->getGateway(ExternalTicketProducer::class)->send('Printer is broken', DistributedBusHeader::DISTRIBUTED_ROUTING_SLIP_VALUE, [ + DistributedBusHeader::DISTRIBUTED_ROUTING_KEY => 'ticket.create', + DistributedBusHeader::DISTRIBUTED_PAYLOAD_TYPE => 'message', + DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME => 'ticket_service', + DistributedBusHeader::DISTRIBUTED_REPLY_TO_CHANNEL => 'ticket.created', + MessageHeaders::CONTENT_TYPE => MediaType::TEXT_PLAIN, + ]); $this->expectExceptionMessage("A reply to channel 'ticket.created' 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."); @@ -443,3 +479,9 @@ public function __construct(public readonly string $ticketId, public readonly st { } } + +interface ExternalTicketProducer +{ + #[MessageGateway('ticket_channel')] + public function send(#[Payload] string $description, #[Header(MessageHeaders::ROUTING_SLIP)] string $routingSlip, #[Headers] array $headers): void; +} From db040d9f3fbd3c9110426afc7b4d81441d0de9ca Mon Sep 17 00:00:00 2001 From: Dariusz Gafka Date: Tue, 20 Oct 2026 18:37:00 +0200 Subject: [PATCH 10/12] style(distribution): apply php-cs-fixer --- .../Distribution/DistributedMessageHandler.php | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php index 2371c33b60..ea79520edd 100644 --- a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php @@ -2,6 +2,9 @@ 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; @@ -12,9 +15,6 @@ use Ecotone\Messaging\Support\InvalidArgumentException; use Ecotone\Modelling\Config\Routing\BusRoutingMap; -use function array_flip; -use function array_intersect_key; - /** * licence Apache-2.0 */ From dbc15b156cace0ef501a2d534c3f307602f0afab Mon Sep 17 00:00:00 2001 From: Dariusz Gafka Date: Tue, 20 Oct 2026 18:45:00 +0200 Subject: [PATCH 11/12] refactor(distribution): name the reply target replyRoutingKey --- .../DbalDistributedMessageReplyTest.php | 6 ++-- .../Ecotone/Api/Gateway/DistributedBus.php | 2 +- .../Api/Gateway/DistributedBusHeader.php | 2 +- .../Config/Flow/MessageFlowDiscovery.php | 2 +- .../Messaging/Support/FrameworkHeaders.php | 2 +- .../DistributedMessageHandler.php | 16 ++++----- .../Distribution/DistributedReplyRouter.php | 8 ++--- .../DistributedBusWithServiceMapModule.php | 4 +-- .../Module/DistributedHandlerModule.php | 4 +-- .../Test/Flow/DistributedFlowScenarios.php | 4 +-- .../DistributedBusMessageReplyTest.php | 36 +++++++++---------- 11 files changed, 43 insertions(+), 43 deletions(-) diff --git a/packages/Dbal/tests/Integration/DbalDistributedMessageReplyTest.php b/packages/Dbal/tests/Integration/DbalDistributedMessageReplyTest.php index 8bf0a28d42..9a80b8bbf5 100644 --- a/packages/Dbal/tests/Integration/DbalDistributedMessageReplyTest.php +++ b/packages/Dbal/tests/Integration/DbalDistributedMessageReplyTest.php @@ -43,7 +43,7 @@ */ final class DbalDistributedMessageReplyTest extends DbalMessagingTestCase { - public function test_the_result_of_a_distributed_message_reaches_the_reply_to_channel_over_dbal_channels(): void + public function test_the_result_of_a_distributed_message_reaches_the_reply_routing_key_over_dbal_channels(): void { $tickets = new class () { #[Distributed] @@ -66,7 +66,7 @@ public function created(DbalTicketWasCreated $ticket, #[Header('requestedBy')] s $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'], replyToChannelName: 'ticket.created'); + $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()); @@ -112,7 +112,7 @@ public function created(string $ticket): void $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', replyToChannelName: 'ticket.created'); + $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()); diff --git a/packages/Ecotone/Api/Gateway/DistributedBus.php b/packages/Ecotone/Api/Gateway/DistributedBus.php index 1d6009f92d..f95d7e4620 100644 --- a/packages/Ecotone/Api/Gateway/DistributedBus.php +++ b/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 = [], ?string $replyToChannelName = null): void; + public function sendMessage(string $targetServiceName, string $targetChannelName, string $payload, string $sourceMediaType = MediaType::TEXT_PLAIN, array $metadata = [], ?string $replyRoutingKey = null): void; } diff --git a/packages/Ecotone/Api/Gateway/DistributedBusHeader.php b/packages/Ecotone/Api/Gateway/DistributedBusHeader.php index 84744b762e..19549076b7 100644 --- a/packages/Ecotone/Api/Gateway/DistributedBusHeader.php +++ b/packages/Ecotone/Api/Gateway/DistributedBusHeader.php @@ -16,5 +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_TO_CHANNEL = 'ecotone_distributed_reply_to_channel'; + public const DISTRIBUTED_REPLY_ROUTING_KEY = 'ecotone_distributed_reply_routing_key'; } diff --git a/packages/Ecotone/src/Messaging/Config/Flow/MessageFlowDiscovery.php b/packages/Ecotone/src/Messaging/Config/Flow/MessageFlowDiscovery.php index 23006719b7..4f35c05125 100644 --- a/packages/Ecotone/src/Messaging/Config/Flow/MessageFlowDiscovery.php +++ b/packages/Ecotone/src/Messaging/Config/Flow/MessageFlowDiscovery.php @@ -277,7 +277,7 @@ private function registerDistributedChannelHandler(string $handlerId, string $in $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' => 'replyToChannelName']); + $this->edges[] = new FlowEdge($handlerId, $busId, FlowEdgeKind::SENDS_TO, ['via' => 'replyRoutingKey']); } } diff --git a/packages/Ecotone/src/Messaging/Support/FrameworkHeaders.php b/packages/Ecotone/src/Messaging/Support/FrameworkHeaders.php index ed0f35665c..7c5b342d71 100644 --- a/packages/Ecotone/src/Messaging/Support/FrameworkHeaders.php +++ b/packages/Ecotone/src/Messaging/Support/FrameworkHeaders.php @@ -110,7 +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_TO_CHANNEL], + $metadata[DistributedBusHeader::DISTRIBUTED_REPLY_ROUTING_KEY], ); return $metadata; diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php index ea79520edd..02cd9f60ef 100644 --- a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedMessageHandler.php @@ -41,8 +41,8 @@ public function handle( string $contentType, #[Header(DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME)] ?string $targetedServiceName, - #[Header(DistributedBusHeader::DISTRIBUTED_REPLY_TO_CHANNEL)] - ?string $replyToChannelName, + #[Header(DistributedBusHeader::DISTRIBUTED_REPLY_ROUTING_KEY)] + ?string $replyRoutingKey, CommandBus $commandBus, EventBus $eventBus, MessagingEntrypointService $messagingEntrypoint @@ -66,7 +66,7 @@ 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, $replyToChannelName === null ? null : self::REPLY_CHANNEL); + $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"); } @@ -75,14 +75,14 @@ public function handle( public function reply( mixed $result, array $metadata, - #[Header(DistributedBusHeader::DISTRIBUTED_REPLY_TO_CHANNEL)] - string $replyToChannelName, + #[Header(DistributedBusHeader::DISTRIBUTED_REPLY_ROUTING_KEY)] + string $replyRoutingKey, MessagingEntrypointService $messagingEntrypoint ): void { - $messagingEntrypoint->sendWithHeaders($result, $this->replyHeaders($metadata, $replyToChannelName), self::REPLY_ROUTING_CHANNEL); + $messagingEntrypoint->sendWithHeaders($result, $this->replyHeaders($metadata, $replyRoutingKey), self::REPLY_ROUTING_CHANNEL); } - private function replyHeaders(array $resultMetadata, string $replyToChannelName): array + private function replyHeaders(array $resultMetadata, string $replyRoutingKey): array { return [ ...FrameworkHeaders::unsetAllFrameworkHeaders($resultMetadata), @@ -91,7 +91,7 @@ private function replyHeaders(array $resultMetadata, string $replyToChannelName) 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 => $replyToChannelName, + DistributedBusHeader::DISTRIBUTED_ROUTING_KEY => $replyRoutingKey, DistributedBusHeader::DISTRIBUTED_SOURCE_SERVICE_NAME => $this->thisServiceName, DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME => $resultMetadata[DistributedBusHeader::DISTRIBUTED_SOURCE_SERVICE_NAME] ?? null, ]; diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php index 346eb39afd..46f90b25f9 100644 --- a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/DistributedReplyRouter.php @@ -53,10 +53,10 @@ public function route( #[Header(DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME)] ?string $targetedServiceName, #[Header(DistributedBusHeader::DISTRIBUTED_ROUTING_KEY)] - string $replyToChannelName, + string $replyRoutingKey, ): array { if ($targetedServiceName === null) { - throw UnknownDistributedDestination::create("A reply to channel '{$replyToChannelName}' 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."); + 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 = []; @@ -70,13 +70,13 @@ public function route( } if ($outboundChannelsByChannel === []) { - throw UnknownDistributedDestination::create("A reply to channel '{$replyToChannelName}' 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 replyToChannelName."); + 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 channel '{$replyToChannelName}' 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."); + 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); diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php index ecab7ffe84..c73aaefe0e 100644 --- a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php @@ -197,7 +197,7 @@ private function registerDistributedBus(ServiceConfiguration $applicationConfigu GatewayHeaderBuilder::create('sourceMediaType', MessageHeaders::CONTENT_TYPE), GatewayHeaderBuilder::create('targetChannelName', DistributedBusHeader::DISTRIBUTED_ROUTING_KEY), GatewayHeaderBuilder::create('targetServiceName', DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME), - GatewayHeaderBuilder::create('replyToChannelName', DistributedBusHeader::DISTRIBUTED_REPLY_TO_CHANNEL), + GatewayHeaderBuilder::create('replyRoutingKey', DistributedBusHeader::DISTRIBUTED_REPLY_ROUTING_KEY), GatewayHeaderValueBuilder::create(DistributedBusHeader::DISTRIBUTED_SOURCE_SERVICE_NAME, $applicationConfiguration->serviceName), GatewayHeaderValueBuilder::create(DistributedBusHeader::DISTRIBUTED_PAYLOAD_TYPE, 'message'), GatewayHeaderValueBuilder::create(MessageHeaders::ROUTING_SLIP, DistributedBusHeader::DISTRIBUTED_ROUTING_SLIP_VALUE), @@ -226,7 +226,7 @@ private function registerReplyRouter(ServiceConfiguration $applicationConfigurat ->withInputChannelName(DistributedMessageHandler::REPLY_ROUTING_CHANNEL) ->withMethodParameterConverters([ HeaderBuilder::createOptional('targetedServiceName', DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME), - HeaderBuilder::create('replyToChannelName', DistributedBusHeader::DISTRIBUTED_ROUTING_KEY), + HeaderBuilder::create('replyRoutingKey', DistributedBusHeader::DISTRIBUTED_ROUTING_KEY), ]) ); } diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedHandlerModule.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedHandlerModule.php index d16b33219e..fd510056cd 100644 --- a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedHandlerModule.php +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedHandlerModule.php @@ -116,7 +116,7 @@ public function prepare(Configuration $messagingConfiguration, array $extensionO HeaderBuilder::create('routingKey', DistributedBusHeader::DISTRIBUTED_ROUTING_KEY), HeaderBuilder::create('contentType', MessageHeaders::CONTENT_TYPE), HeaderBuilder::createOptional('targetedServiceName', DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME), - HeaderBuilder::createOptional('replyToChannelName', DistributedBusHeader::DISTRIBUTED_REPLY_TO_CHANNEL), + HeaderBuilder::createOptional('replyRoutingKey', DistributedBusHeader::DISTRIBUTED_REPLY_ROUTING_KEY), ReferenceBuilder::create('commandBus', CommandBus::class), ReferenceBuilder::create('eventBus', EventBus::class), ReferenceBuilder::create('messagingEntrypoint', MessagingEntrypointService::class), @@ -128,7 +128,7 @@ public function prepare(Configuration $messagingConfiguration, array $extensionO ->withMethodParameterConverters([ PayloadBuilder::create('result'), AllHeadersBuilder::createWith('metadata'), - HeaderBuilder::create('replyToChannelName', DistributedBusHeader::DISTRIBUTED_REPLY_TO_CHANNEL), + HeaderBuilder::create('replyRoutingKey', DistributedBusHeader::DISTRIBUTED_REPLY_ROUTING_KEY), ReferenceBuilder::create('messagingEntrypoint', MessagingEntrypointService::class), ]) ); diff --git a/packages/Ecotone/src/Test/Flow/DistributedFlowScenarios.php b/packages/Ecotone/src/Test/Flow/DistributedFlowScenarios.php index 6e5d2fe70e..30038e10dd 100644 --- a/packages/Ecotone/src/Test/Flow/DistributedFlowScenarios.php +++ b/packages/Ecotone/src/Test/Flow/DistributedFlowScenarios.php @@ -136,7 +136,7 @@ public function confirm(string $ticket): void }; return new FlowScenario( - 'distributed message answered on its reply-to channel', + 'distributed message answered on its reply routing key', [$tickets::class], [$tickets], [ @@ -145,7 +145,7 @@ public function confirm(string $ticket): void ServiceConfiguration::createWithDefaults()->withServiceName('ticket_service'), ], [ - FlowStage::when(fn (FlowTestSupport $ecotone, array $metadata) => $ecotone->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', metadata: $metadata, replyToChannelName: 'ticket.created'))->leavingWorkPending() + FlowStage::when(fn (FlowTestSupport $ecotone, array $metadata) => $ecotone->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', metadata: $metadata, replyRoutingKey: 'ticket.created'))->leavingWorkPending() ->expect(ExpectedSpan::named('send distributed')->ofKind(SpanKind::Producer)->containing(ExpectedSpan::named('send distributed_channel')->ofKind(SpanKind::Producer))) ->withNoOtherSpans(), FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->run('distributed_channel', ExecutionPollingMetadata::createWithTestingSetup(handledMessageLimit: 1)))->leavingWorkPending() diff --git a/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php b/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php index 923a7e6f6c..2d78b839e0 100644 --- a/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php +++ b/packages/Ecotone/tests/Messaging/Unit/Distributed/DistributedBusMessageReplyTest.php @@ -48,7 +48,7 @@ */ final class DistributedBusMessageReplyTest extends TestCase { - public function test_the_result_of_a_distributed_message_triggers_the_reply_to_channel_in_the_sending_service(): void + public function test_the_result_of_a_distributed_message_triggers_the_reply_routing_key_in_the_sending_service(): void { $channels = [SimpleMessageChannelBuilder::createQueueChannel('ticket_channel'), SimpleMessageChannelBuilder::createQueueChannel('user_channel')]; $tickets = new class () { @@ -71,7 +71,7 @@ public function created(string $ticket): void $userService = $this->service('user_service', $users, $channels, DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'ticket_service', channelName: 'ticket_channel')); $ticketService = $this->service('ticket_service', $tickets, $channels, DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel')); - $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $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()); @@ -102,7 +102,7 @@ public function created(string $ticket): void $userService = $this->userService($users, $channels); $ticketService = $this->ticketService($tickets, $channels); - $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $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([], $users->replies); @@ -139,7 +139,7 @@ classesToResolve: [$users::class], ); $ticketService = $this->ticketService($this->ticketsReturningDescription(), $channels); - $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $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()); @@ -175,7 +175,7 @@ public function created(string $ticket, #[Header('requestedBy')] string $request $userService = $this->userService($users, $channels); $ticketService = $this->ticketService($tickets, $channels); - $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', metadata: ['requestedBy' => 'johny'], replyToChannelName: 'ticket.created'); + $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()); @@ -205,7 +205,7 @@ public function created(string $ticket): void $userService = $this->userService($users, $channels); $ticketService = $this->ticketService($tickets, $channels); - $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $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()); @@ -242,7 +242,7 @@ public function created(string $ticket): string $userService = $this->userService($users, $channels); $ticketService = $this->ticketService($tickets, $channels); - $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $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()); $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); @@ -252,7 +252,7 @@ public function created(string $ticket): string self::assertSame(['ticket for: Printer is broken'], $users->replies); } - public function test_an_object_result_reaches_the_reply_to_channel_as_the_same_object(): void + public function test_an_object_result_reaches_the_reply_routing_key_as_the_same_object(): void { $channels = $this->channels(); $tickets = new class () { @@ -275,14 +275,14 @@ public function created(TicketWasCreated $ticket): void $userService = $this->userService($users, $channels); $ticketService = $this->ticketService($tickets, $channels); - $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $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::assertEquals([new TicketWasCreated('ticket-1', 'Printer is broken')], $users->replies); } - public function test_sending_a_message_without_reply_to_channel_sends_no_reply(): void + public function test_sending_a_message_without_reply_routing_key_sends_no_reply(): void { $channels = $this->channels(); $tickets = new class () { @@ -338,7 +338,7 @@ public function created(string $ticket): void DistributedServiceMap::initialize('second_distributed_bus')->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel'), ); - $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $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()); @@ -365,7 +365,7 @@ public function created(string $ticket): void DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel')->withAsynchronousChannel('distributed_bus_async'), ); - $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $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()); @@ -379,9 +379,9 @@ public function test_a_reply_is_refused_when_no_service_map_of_the_receiver_maps }, $channels); $ticketService = $this->service('ticket_service', $this->ticketsReturningDescription(), $channels); - $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyRoutingKey: 'ticket.created'); - $this->expectExceptionMessage("A reply to channel 'ticket.created' of service 'user_service' cannot be sent, because no Service Map in service 'ticket_service' maps service 'user_service' to a channel. Register the channel 'user_service' consumes with DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'user_service', channelName: '...') in service 'ticket_service', or send the message without replyToChannelName."); + $this->expectExceptionMessage("A reply to routing key 'ticket.created' of service 'user_service' cannot be sent, because no Service Map in service 'ticket_service' maps service 'user_service' to a channel. Register the channel 'user_service' consumes with DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'user_service', channelName: '...') in service 'ticket_service', or send the message without replyRoutingKey."); $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); } @@ -399,9 +399,9 @@ public function test_a_reply_is_refused_when_two_service_maps_of_the_receiver_ma DistributedServiceMap::initialize('second_distributed_bus')->withCommandMapping(targetServiceName: 'user_service', channelName: 'user_channel_2'), ); - $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyToChannelName: 'ticket.created'); + $userService->getDistributedBus()->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyRoutingKey: 'ticket.created'); - $this->expectExceptionMessage("A reply to channel 'ticket.created' of service 'user_service' cannot be sent, because Service Maps in service 'ticket_service' map service 'user_service' to different channels: 'Ecotone\\Api\\Gateway\\DistributedBus' to 'user_channel', 'second_distributed_bus' to 'user_channel_2'. Map service 'user_service' in one Service Map only, or to the same channel in each."); + $this->expectExceptionMessage("A reply to routing key 'ticket.created' of service 'user_service' cannot be sent, because Service Maps in service 'ticket_service' map service 'user_service' to different channels: 'Ecotone\\Api\\Gateway\\DistributedBus' to 'user_channel', 'second_distributed_bus' to 'user_channel_2'. Map service 'user_service' in one Service Map only, or to the same channel in each."); $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); } @@ -425,11 +425,11 @@ classesToResolve: [$tickets::class, ExternalTicketProducer::class], DistributedBusHeader::DISTRIBUTED_ROUTING_KEY => 'ticket.create', DistributedBusHeader::DISTRIBUTED_PAYLOAD_TYPE => 'message', DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME => 'ticket_service', - DistributedBusHeader::DISTRIBUTED_REPLY_TO_CHANNEL => 'ticket.created', + DistributedBusHeader::DISTRIBUTED_REPLY_ROUTING_KEY => 'ticket.created', MessageHeaders::CONTENT_TYPE => MediaType::TEXT_PLAIN, ]); - $this->expectExceptionMessage("A reply to channel 'ticket.created' 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."); + $this->expectExceptionMessage("A reply to routing key 'ticket.created' 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."); $ticketService->run('ticket_channel', ExecutionPollingMetadata::createWithTestingSetup()); } From 2737721e5494a06ec0760caf1380c7040e1208ef Mon Sep 17 00:00:00 2001 From: Dariusz Gafka Date: Tue, 20 Oct 2026 18:52:00 +0200 Subject: [PATCH 12/12] refactor(distribution): name sendMessage's target routingKey, like every other DistributedBus method --- packages/Ecotone/Api/Gateway/DistributedBus.php | 2 +- .../Distribution/Module/DistributedBusWithServiceMapModule.php | 2 +- upgrade-2.0.md | 1 + 3 files changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/Ecotone/Api/Gateway/DistributedBus.php b/packages/Ecotone/Api/Gateway/DistributedBus.php index f95d7e4620..973d81817c 100644 --- a/packages/Ecotone/Api/Gateway/DistributedBus.php +++ b/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 = [], ?string $replyRoutingKey = null): void; + public function sendMessage(string $targetServiceName, string $routingKey, string $payload, string $sourceMediaType = MediaType::TEXT_PLAIN, array $metadata = [], ?string $replyRoutingKey = null): void; } diff --git a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php index c73aaefe0e..2c5e5ebade 100644 --- a/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php +++ b/packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php @@ -195,7 +195,7 @@ private function registerDistributedBus(ServiceConfiguration $applicationConfigu GatewayPayloadBuilder::create('payload'), GatewayHeadersBuilder::create('metadata'), GatewayHeaderBuilder::create('sourceMediaType', MessageHeaders::CONTENT_TYPE), - GatewayHeaderBuilder::create('targetChannelName', DistributedBusHeader::DISTRIBUTED_ROUTING_KEY), + GatewayHeaderBuilder::create('routingKey', DistributedBusHeader::DISTRIBUTED_ROUTING_KEY), GatewayHeaderBuilder::create('targetServiceName', DistributedBusHeader::DISTRIBUTED_TARGET_SERVICE_NAME), GatewayHeaderBuilder::create('replyRoutingKey', DistributedBusHeader::DISTRIBUTED_REPLY_ROUTING_KEY), GatewayHeaderValueBuilder::create(DistributedBusHeader::DISTRIBUTED_SOURCE_SERVICE_NAME, $applicationConfiguration->serviceName), diff --git a/upgrade-2.0.md b/upgrade-2.0.md index 79f221be0a..0837b2d60e 100644 --- a/upgrade-2.0.md +++ b/upgrade-2.0.md @@ -2858,6 +2858,7 @@ described. A call that uses an old name fails loudly — PHP's `Unknown named pa | `EventSourcedRepository::findBy($aggregateClassName, $identifiers, fromAggregateVersion: 3)` | `findBy(…, fromVersion: 3)`; an implementing class keeps working whatever it names the parameter | | `DbalDeadLetterBuilder::createList()`, `createShow()`, `createCount()`, `createDelete()`, `createDeleteAll()`, `createReply()`, `createReplyAll()` — each `($referenceName, $connectionReferenceName)` | `createList($referenceName)` and so on, with `createReplay()` / `createReplayAll()` for the last two (§15). The connection is no longer an argument: the builder addresses the dead letter registered under `$referenceName`, which carries its own connection. PHP ignores a surplus positional argument, so a 1.x positional call still runs and its connection argument is dropped; a named `connectionReferenceName:` fails | | `DbalDeadLetterBuilder::createStore($connectionReferenceName)` | `createStore($referenceName, $inputChannelName)`; a 1.x one-argument call fails with PHP's `Too few arguments` | +| `DistributedBus::sendMessage($targetServiceName, targetChannelName: 'ticket.create', …)` | `sendMessage($targetServiceName, routingKey: 'ticket.create', …)`, the name every other `DistributedBus` method gives its routing argument; positional calls are unchanged | **How to adapt:** rename the call. `DbalDeadLetterBuilder`'s factories are the dead letter module's own plumbing; an application reaches the dead letter through `DeadLetterGateway`, and through the builder only for its channel-name