Merge SignalQueue into SignalDispatcher

Author: roxblnfk

src/Internal/Declaration/WorkflowInstance/SignalDispatcher.php

@@ -19,14 +19,37 @@
 use Temporal\Internal\Declaration\Prototype\WorkflowPrototype;
 
 /**
+ * @psalm-type Consumer = callable(ValuesInterface): mixed
+ * @psalm-type OnSignalCallable = \Closure(non-empty-string $name, Consumer $handler, ValuesInterface $arguments): void
+ *
  * @internal
  */
 final class SignalDispatcher implements Destroyable
 {
     /** @var array<non-empty-string, SignalMethod> */
     private array $signalHandlers = [];
 
-    private readonly SignalQueue $signalQueue;
+    /**
+     * @var OnSignalCallable
+     */
+    private \Closure $onSignal;
+
+    /**
+     * @var array<int, SignalQueueItem>
+     */
+    private array $queue = [];
+
+    /**
+     * @var array<non-empty-string, Consumer>
+     */
+    private array $consumers = [];
+
+    /**
+     * A fallback consumer to handle signals when no consumer is attached.
+     *
+     * @var null|\Closure(non-empty-string, ValuesInterface): mixed
+     */
+    private ?\Closure $dynamicConsumer = null;
 
     /**
      * @param object $context Workflow instance.
@@ -35,16 +58,17 @@ public function __construct(
         WorkflowPrototype $prototype,
         private readonly object $context,
     ) {
-        $this->signalQueue = new SignalQueue();
-
         foreach ($prototype->getSignalHandlers() as $definition) {
             $this->addFromSignalDefinition($definition);
         }
     }
 
-    public function getSignalQueue(): SignalQueue
+    /**
+     * @param OnSignalCallable $handler
+     */
+    public function onSignal(\Closure $handler): void
     {
-        return $this->signalQueue;
+        $this->onSignal = $handler;
     }
 
     /**
@@ -53,7 +77,7 @@ public function getSignalQueue(): SignalQueue
      */
     public function getSignalHandler(string $name): \Closure
     {
-        return fn(ValuesInterface $values) => $this->signalQueue->push($name, $values);
+        return fn(ValuesInterface $values) => $this->push($name, $values);
     }
 
     /**
@@ -67,7 +91,7 @@ public function addSignalHandler(string $name, callable $handler, string $descri
             $handler,
             $description,
         );
-        $this->signalQueue->attach($name, $handler);
+        $this->attach($name, $handler);
     }
 
     public function addFromSignalDefinition(SignalDefinition $definition): void
@@ -79,20 +103,32 @@ public function addFromSignalDefinition(SignalDefinition $definition): void
             $handler,
             $definition->description,
         );
-        $this->signalQueue->attach($name, $handler);
+        $this->attach($name, $handler);
     }
 
     /**
      * @param callable(non-empty-string, ValuesInterface): mixed $handler
      */
     public function setDynamicSignalHandler(callable $handler): void
     {
-        $this->signalQueue->setFallback($handler(...));
+        $consumer = $handler(...);
+
+        $this->dynamicConsumer = $consumer;
+
+        // Flush all signals that have no consumer
+        foreach ($this->queue as $k => $item) {
+            if (\array_key_exists($item->name, $this->consumers)) {
+                continue;
+            }
+
+            unset($this->queue[$k]);
+            $this->consumeFallback($item->name, $item->values);
+        }
     }
 
     public function clearSignalQueue(): void
     {
-        $this->signalQueue->clear();
+        $this->queue = [];
     }
 
     /**
@@ -108,11 +144,10 @@ public function getSignalHandlers(): array
                 ->setDescription($handler->description);
         }
 
-        // todo
-        // if ($this->dynamic !== null) {
-        //     $handlers[] = (new WorkflowInteractionDefinition())
-        //         ->setDescription('Dynamic signal handler');
-        // }
+        if ($this->dynamicConsumer !== null) {
+            $handlers[] = (new WorkflowInteractionDefinition())
+                ->setDescription('Dynamic signal handler');
+        }
 
         \usort(
             $handlers,
@@ -127,8 +162,66 @@ public function getSignalHandlers(): array
 
     public function destroy(): void
     {
-        $this->signalQueue->clear();
-        $this->signalQueue->destroy();
+        $this->queue = [];
+        $this->consumers = [];
+        $this->dynamicConsumer = null;
+        unset($this->onSignal);
         $this->signalHandlers = [];
     }
+
+    /**
+     * @param non-empty-string $signal
+     */
+    private function push(string $signal, ValuesInterface $values): void
+    {
+        if (isset($this->consumers[$signal])) {
+            $this->consume($signal, $values, $this->consumers[$signal]);
+            return;
+        }
+
+        if ($this->dynamicConsumer !== null) {
+            $this->consumeFallback($signal, $values);
+            return;
+        }
+
+        $this->queue[] = new SignalQueueItem($signal, $values);
+    }
+
+    /**
+     * @param non-empty-string $signal
+     * @param Consumer $consumer
+     */
+    private function attach(string $signal, callable $consumer): void
+    {
+        $this->consumers[$signal] = $consumer; // overwrite
+
+        foreach ($this->queue as $k => $item) {
+            if ($item->name === $signal) {
+                unset($this->queue[$k]);
+                $this->consume($signal, $item->values, $consumer);
+            }
+        }
+    }
+
+    /**
+     * @param non-empty-string $signal
+     * @param Consumer $consumer
+     */
+    private function consume(string $signal, ValuesInterface $values, callable $consumer): void
+    {
+        ($this->onSignal)($signal, $consumer, $values);
+    }
+
+    /**
+     * @param non-empty-string $signal
+     */
+    private function consumeFallback(string $signal, ValuesInterface $values): void
+    {
+        $handler = $this->dynamicConsumer;
+        \assert($handler !== null);
+
+        // Wrap the fallback consumer to call interceptors
+        $consumer = static fn(ValuesInterface $values): mixed => $handler($signal, $values);
+        ($this->onSignal)($signal, $consumer, $values);
+    }
 }

src/Internal/Declaration/WorkflowInstance/SignalQueue.php

@@ -126,7 +126,7 @@ static function () use ($handler, $inboundPipeline, $input): mixed {
             });
 
         // Configure signal handler
-        $workflowInstance->getSignalDispatcher()->getSignalQueue()->onSignal(
+        $workflowInstance->getSignalDispatcher()->onSignal(
             function (string $name, callable $handler, ValuesInterface $arguments) use ($inboundPipeline): void {
                 // Define Context for interceptors Pipeline
                 Workflow::setCurrentContext($this->scopeContext);