Feat/streaming channels for distributed services (#590)

* streaming test

* feat: streaming channels for distributed services

* fixes

Author: dgafka

packages/Ecotone/src/Modelling/Api/Distribution/DistributedServiceMap.php

@@ -5,6 +5,7 @@
 namespace Ecotone\Modelling\Api\Distribution;
 
 use Ecotone\Messaging\Attribute\Asynchronous;
+use Ecotone\Messaging\Config\ConfigurationException;
 use Ecotone\Messaging\Config\Container\AttributeDefinition;
 use Ecotone\Messaging\Config\Container\DefinedObject;
 use Ecotone\Messaging\Config\Container\Definition;
@@ -19,15 +20,17 @@
 final class DistributedServiceMap implements DefinedObject
 {
     /**
-     * @param array<string, string> $serviceMapping
-     * @param array<string, array<string>> $subscriptionRoutingKeys
+     * @param array<string, string> $commandMapping - service name -> channel name (for command routing)
+     * @param array<string, array{keys: ?array<string>, exclude: array<string>}> $eventSubscriptions - channel name -> ['keys' => [...] or null, 'exclude' => [...]]
      * @param array<object> $distributedBusAnnotations
+     * @param bool|null $legacyMode - null = not set, true = legacy (withServiceMapping), false = new API (withCommandMapping/withEventMapping)
      */
     public function __construct(
         private string $referenceName,
-        private array $serviceMapping = [],
-        private ?array $subscriptionRoutingKeys = null,
+        private array $commandMapping = [],
+        private array $eventSubscriptions = [],
         private array $distributedBusAnnotations = [],
+        private ?bool $legacyMode = null,
     ) {
         Assert::allObjects($this->distributedBusAnnotations, 'Annotations passed to DistributedServiceMap, must all be objects');
     }
@@ -38,13 +41,56 @@ public static function initialize(string $referenceName = DistributedBus::class)
     }
 
     /**
+     * @deprecated Use withCommandMapping() and withEventMapping() instead
      * @param array|null $subscriptionRoutingKeys If null subscribing to all events, if empty array to none, if non empty array then keys will be used to match the name
      */
     public function withServiceMapping(string $serviceName, string $channelName, ?array $subscriptionRoutingKeys = null): self
     {
         $self = clone $this;
-        $self->serviceMapping[$serviceName] = $channelName;
-        $self->subscriptionRoutingKeys[$serviceName] = $subscriptionRoutingKeys;
+        $self->assertNotInNewMode('withServiceMapping');
+        $self->legacyMode = true;
+
+        $self->commandMapping[$serviceName] = $channelName;
+        $self->eventSubscriptions[$channelName] = [
+            'keys' => $subscriptionRoutingKeys,
+            'exclude' => [],
+        ];
+
+        return $self;
+    }
+
+    /**
+     * Maps a service to a channel for command routing only.
+     * Does NOT create any event subscription.
+     */
+    public function withCommandMapping(string $targetServiceName, string $channelName): self
+    {
+        $self = clone $this;
+        $self->assertNotInLegacyMode('withCommandMapping');
+        $self->legacyMode = false;
+
+        $self->commandMapping[$targetServiceName] = $channelName;
+
+        return $self;
+    }
+
+    /**
+     * Creates an event subscription for a channel with explicit subscription keys.
+     *
+     * @param string $channelName Target channel to send events to
+     * @param array<string> $subscriptionKeys Routing key patterns to match
+     * @param array<string> $excludeEventsFromServices Service names whose events should NOT be sent to this channel
+     */
+    public function withEventMapping(string $channelName, array $subscriptionKeys, array $excludeEventsFromServices = []): self
+    {
+        $self = clone $this;
+        $self->assertNotInLegacyMode('withEventMapping');
+        $self->legacyMode = false;
+
+        $self->eventSubscriptions[$channelName] = [
+            'keys' => $subscriptionKeys,
+            'exclude' => $excludeEventsFromServices,
+        ];
 
         return $self;
     }
@@ -60,43 +106,86 @@ public function withAsynchronousChannel(string $channelName): self
     /**
      * @return array<string, string>
      */
-    public function getServiceMapping(): array
+    public function getCommandMapping(): array
     {
-        return $this->serviceMapping;
+        return $this->commandMapping;
     }
 
+    /**
+     * LEGACY MODE ONLY - Get all channels except the one belonging to the given service.
+     * Uses service name to channel mapping for exclusion.
+     *
+     * @deprecated For new mode, use getAllSubscriptionChannels() instead
+     */
     public function getAllChannelNamesBesides(string $serviceName, string $routingKey): array
     {
         $filteredChannels = [];
+        $excludeChannel = $this->commandMapping[$serviceName] ?? null;
+
+        foreach ($this->eventSubscriptions as $channel => $config) {
+            if ($channel === $excludeChannel) {
+                continue;
+            }
+
+            $keys = $config['keys'];
+
+            if ($keys === null) {
+                $filteredChannels[] = $channel;
+
+                continue;
+            }
 
-        foreach ($this->serviceMapping as $service => $channel) {
-            if ($service !== $serviceName) {
-                if ($this->subscriptionRoutingKeys[$service] === null) {
+            foreach ($keys as $subscriptionEventFilter) {
+                if (BusRoutingMap::globMatch($subscriptionEventFilter, $routingKey)) {
                     $filteredChannels[] = $channel;
 
-                    continue;
+                    break;
                 }
+            }
+        }
+
+        return $filteredChannels;
+    }
+
+    /**
+     * NEW MODE ONLY - Get all subscription channels for an event.
+     * Uses explicit exclude list from eventSubscriptions config.
+     *
+     * @param string $sourceServiceName The service publishing the event
+     * @param string $routingKey The event routing key
+     * @return array<string>
+     */
+    public function getAllSubscriptionChannels(string $sourceServiceName, string $routingKey): array
+    {
+        $filteredChannels = [];
+
+        foreach ($this->eventSubscriptions as $channel => $config) {
+            $keys = $config['keys'];
+            $exclude = $config['exclude'];
+
+            if (\in_array($sourceServiceName, $exclude, true)) {
+                continue;
+            }
 
-                foreach ($this->subscriptionRoutingKeys[$service] as $subscriptionEventFilter) {
-                    if (BusRoutingMap::globMatch($subscriptionEventFilter, $routingKey)) {
-                        $filteredChannels[] = $channel;
+            foreach ($keys as $subscriptionEventFilter) {
+                if (BusRoutingMap::globMatch($subscriptionEventFilter, $routingKey)) {
+                    $filteredChannels[] = $channel;
 
-                        break;
-                    }
+                    break;
                 }
             }
         }
 
-        return $filteredChannels;
+        return array_unique($filteredChannels);
     }
 
     public function getChannelNameFor(string $serviceName): string
     {
-        if (! array_key_exists($serviceName, $this->serviceMapping)) {
+        if (! \array_key_exists($serviceName, $this->commandMapping)) {
             throw new UnknownDistributedDestination("Service {$serviceName} is not registered in distributed service map");
         }
 
-        return $this->serviceMapping[$serviceName];
+        return $this->commandMapping[$serviceName];
     }
 
     public function getReferenceName(): string
@@ -112,16 +201,42 @@ public function getDistributedBusAnnotations(): array
         return $this->distributedBusAnnotations;
     }
 
+    public function isLegacyMode(): bool
+    {
+        return $this->legacyMode === true;
+    }
+
     public function getDefinition(): Definition
     {
         return Definition::createFor(
             self::class,
             [
                 $this->referenceName,
-                $this->serviceMapping,
-                $this->subscriptionRoutingKeys,
+                $this->commandMapping,
+                $this->eventSubscriptions,
                 $this->distributedBusAnnotations,
+                $this->legacyMode,
             ]
         );
     }
+
+    private function assertNotInLegacyMode(string $methodName): void
+    {
+        if ($this->legacyMode === true) {
+            throw ConfigurationException::create(
+                "Cannot use {$methodName}() after withServiceMapping(). " .
+                'Use either legacy API (withServiceMapping) or new API (withCommandMapping/withEventMapping), not both.'
+            );
+        }
+    }
+
+    private function assertNotInNewMode(string $methodName): void
+    {
+        if ($this->legacyMode === false) {
+            throw ConfigurationException::create(
+                "Cannot use {$methodName}() after withCommandMapping() or withEventMapping(). " .
+                'Use either legacy API (withServiceMapping) or new API (withCommandMapping/withEventMapping), not both.'
+            );
+        }
+    }
 }

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

@@ -31,9 +31,13 @@ public function route(
         $routingKey,
     ): array {
         if ($payloadType === 'event') {
-            return $this->distributedServiceMap->getAllChannelNamesBesides($this->thisServiceName, $routingKey);
-        } elseif (in_array($payloadType, ['command', 'message'])) {
-            Assert::isTrue($targetedServiceName !== null, sprintf('
+            if ($this->distributedServiceMap->isLegacyMode()) {
+                return $this->distributedServiceMap->getAllChannelNamesBesides($this->thisServiceName, $routingKey);
+            } else {
+                return $this->distributedServiceMap->getAllSubscriptionChannels($this->thisServiceName, $routingKey);
+            }
+        } elseif (\in_array($payloadType, ['command', 'message'])) {
+            Assert::isTrue($targetedServiceName !== null, \sprintf('
                 Cannot send commands to shared channel - `%s`. Commands follow point-to-point semantics, and shared channels are reserved for events only.
                 Change your channel to standard pollable channel.
             ', $targetedServiceName));

packages/Ecotone/src/Modelling/MessageHandling/Distribution/Module/DistributedBusWithServiceMapModule.php

@@ -56,7 +56,7 @@ public function prepare(Configuration $messagingConfiguration, array $extensionO
                 throw LicensingException::create('Distributed Bus with Service Map is available only as part of Ecotone Enterprise.');
             }
 
-            foreach ($distributedServiceMap->getServiceMapping() as $serviceName => $channelName) {
+            foreach ($distributedServiceMap->getCommandMapping() as $serviceName => $channelName) {
                 if (! $this->isMessageChannelAvailable($messageChannels, $channelName)) {
                     throw ConfigurationException::create("Service Map has Service {$serviceName} mapped to channel {$channelName} but it is not available in message channels. Have you forgot to register it?");
                 }