Merge pull request #581: Add methods to register dynamic handlers

Author: roxblnfk

composer.json

@@ -57,7 +57,7 @@
         "dereuromark/composer-prefer-lowest": "^0.1.10",
         "doctrine/annotations": "^1.14.4 || ^2.0.2",
         "internal/dload": "^1.0",
-        "jetbrains/phpstorm-attributes": "dev-master@dev",
+        "jetbrains/phpstorm-attributes": "dev-master",
         "laminas/laminas-code": "^4.16",
         "phpunit/phpunit": "^10.5.41",
         "spiral/code-style": "~2.1.2",
@@ -77,8 +77,9 @@
         }
     },
     "suggest": {
-        "doctrine/annotations": "For Doctrine metadata driver support",
-        "ext-grpc": "For Client calls"
+        "ext-grpc": "For Client calls",
+        "ext-protobuf": "For better performance",
+        "buggregator/trap": "For better debugging"
     },
     "scripts": {
         "post-update-cmd": "Temporal\\Worker\\Transport\\RoadRunnerVersionChecker::postUpdate",

psalm-baseline.xml

@@ -1,5 +1,5 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<files psalm-version="6.9.1@81c8a77c0793d450fee40265cfe68891df11d505">
+<files psalm-version="6.10.0@9c0add4eb88d4b169ac04acb7c679918cbb9c252">
   <file src="src/Activity.php">
     <ImplicitToStringCast>
       <code><![CDATA[$type]]></code>
@@ -606,23 +606,16 @@
     </RedundantConditionGivenDocblockType>
   </file>
   <file src="src/Internal/Declaration/WorkflowInstance.php">
-    <MissingClosureReturnType>
-      <code><![CDATA[function (QueryInput $input) use ($fn) {]]></code>
-    </MissingClosureReturnType>
+    <LessSpecificImplementedReturnType>
+      <code><![CDATA[UpdateHandler|null]]></code>
+    </LessSpecificImplementedReturnType>
     <PropertyNotSetInConstructor>
       <code><![CDATA[$queryExecutor]]></code>
       <code><![CDATA[$updateExecutor]]></code>
       <code><![CDATA[$updateValidator]]></code>
     </PropertyNotSetInConstructor>
-    <PropertyTypeCoercion>
-      <code><![CDATA[$this->queryHandlers]]></code>
-      <code><![CDATA[$this->signalHandlers]]></code>
-    </PropertyTypeCoercion>
   </file>
   <file src="src/Internal/Declaration/WorkflowInstance/SignalQueue.php">
-    <ArgumentTypeCoercion>
-      <code><![CDATA[$signal]]></code>
-    </ArgumentTypeCoercion>
     <MissingConstructor>
       <code><![CDATA[$onSignal]]></code>
     </MissingConstructor>
@@ -1187,9 +1180,6 @@
     </UnevaluatedCode>
   </file>
   <file src="src/Internal/Workflow/Process/Process.php">
-    <MissingClosureParamType>
-      <code><![CDATA[$result]]></code>
-    </MissingClosureParamType>
     <PropertyNotSetInConstructor>
       <code><![CDATA[Process]]></code>
     </PropertyNotSetInConstructor>
@@ -1521,7 +1511,7 @@
     <InvalidReturnStatement>
       <code><![CDATA[self::getCurrentContext()->newActivityStub($class, $options)]]></code>
       <code><![CDATA[self::getCurrentContext()->registerQuery($queryType, $handler)]]></code>
-      <code><![CDATA[self::getCurrentContext()->registerSignal($queryType, $handler)]]></code>
+      <code><![CDATA[self::getCurrentContext()->registerSignal($name, $handler)]]></code>
     </InvalidReturnStatement>
     <InvalidReturnType>
       <code><![CDATA[ScopedContextInterface]]></code>

src/Interceptor/WorkflowInbound/QueryInput.php

@@ -24,6 +24,7 @@ class QueryInput
      * @internal Don't use the constructor. Use {@see self::with()} instead.
      */
     public function __construct(
+        /** @var non-empty-string */
         public readonly string $queryName,
         public readonly ValuesInterface $arguments,
         public readonly WorkflowInfo $info,

src/Internal/Declaration/WorkflowInstance.php

@@ -24,35 +24,36 @@
 
 /**
  * @psalm-type QueryHandler = \Closure(QueryInput): mixed
- * @psalm-type UpdateHandler = \Closure(UpdateInput, Deferred): PromiseInterface
+ * @psalm-type UpdateHandler = \Closure(UpdateInput, Deferred): mixed
  * @psalm-type ValidateUpdateHandler = \Closure(UpdateInput): void
  * @psalm-type QueryExecutor = \Closure(QueryInput, callable(ValuesInterface): mixed): mixed
  * @psalm-type UpdateExecutor = \Closure(UpdateInput, callable(ValuesInterface): mixed, Deferred): PromiseInterface
- * @psalm-type ValidateUpdateExecutor = \Closure(UpdateInput, callable(ValuesInterface): mixed): mixed
+ * @psalm-type ValidateUpdateExecutor = \Closure(UpdateInput, callable(ValuesInterface): mixed): void
  * @psalm-type UpdateValidator = \Closure(UpdateInput, UpdateHandler): void
  *
  * @internal
  */
 final class WorkflowInstance extends Instance implements WorkflowInstanceInterface
 {
-    /**
-     * @var array<non-empty-string, QueryHandler>
-     */
+    /** @var array<non-empty-string, QueryHandler> */
     private array $queryHandlers = [];
 
-    /**
-     * @var array<non-empty-string, MethodHandler>
-     */
+    /** @var null|QueryHandler */
+    private ?\Closure $queryDynamicHandler = null;
+
+    /** @var null|UpdateHandler */
+    private ?\Closure $updateDynamicHandler = null;
+
+    /** @var null|ValidateUpdateHandler */
+    private ?\Closure $updateDynamicValidator = null;
+
+    /** @var array<non-empty-string, MethodHandler> */
     private array $signalHandlers = [];
 
-    /**
-     * @var array<non-empty-string, UpdateHandler>
-     */
+    /** @var array<non-empty-string, UpdateHandler> */
     private array $updateHandlers = [];
 
-    /**
-     * @var array<non-empty-string, null|ValidateUpdateHandler>
-     */
+    /** @var array<non-empty-string, null|ValidateUpdateHandler> */
     private array $validateUpdateHandlers = [];
 
     private SignalQueue $signalQueue;
@@ -174,7 +175,7 @@ public function getSignalQueue(): SignalQueue
      */
     public function findQueryHandler(string $name): ?\Closure
     {
-        return $this->queryHandlers[$name] ?? null;
+        return $this->queryHandlers[$name] ?? $this->queryDynamicHandler;
     }
 
     /**
@@ -185,7 +186,7 @@ public function findQueryHandler(string $name): ?\Closure
      */
     public function findUpdateHandler(string $name): ?\Closure
     {
-        return $this->updateHandlers[$name] ?? null;
+        return $this->updateHandlers[$name] ?? $this->updateDynamicHandler;
     }
 
     /**
@@ -196,7 +197,11 @@ public function findUpdateHandler(string $name): ?\Closure
      */
     public function findValidateUpdateHandler(string $name): ?\Closure
     {
-        return $this->validateUpdateHandlers[$name] ?? null;
+        return $this->validateUpdateHandlers[$name] ?? (
+            \array_key_exists($name, $this->updateHandlers)
+                ? null
+                : $this->updateDynamicValidator
+        );
     }
 
     /**
@@ -207,16 +212,13 @@ public function addQueryHandler(string $name, callable $handler): void
         $fn = $this->createCallableHandler($handler);
 
         $this->queryHandlers[$name] = $this->pipeline->with(
-            function (QueryInput $input) use ($fn) {
-                return ($this->queryExecutor)($input, $fn);
-            },
+            fn(QueryInput $input): mixed => ($this->queryExecutor)($input, $fn),
             /** @see WorkflowInboundCallsInterceptor::handleQuery() */
             'handleQuery',
         )(...);
     }
 
     /**
-     * @param non-empty-string $name
      * @throws \ReflectionException
      */
     public function addUpdateHandler(string $name, callable $handler): void
@@ -233,7 +235,6 @@ function (UpdateInput $input, Deferred $deferred) use ($fn) {
     }
 
     /**
-     * @param non-empty-string $name
      * @throws \ReflectionException
      */
     public function addValidateUpdateHandler(string $name, callable $handler): void
@@ -263,15 +264,49 @@ public function getSignalHandler(string $name): \Closure
         return fn(ValuesInterface $values) => $this->signalQueue->push($name, $values);
     }
 
-    /**
-     * @throws \ReflectionException
-     */
     public function addSignalHandler(string $name, callable $handler): void
     {
         $this->signalHandlers[$name] = $this->createCallableHandler($handler);
         $this->signalQueue->attach($name, $this->signalHandlers[$name]);
     }
 
+    public function setDynamicSignalHandler(callable $handler): void
+    {
+        $this->signalQueue->setFallback($handler(...));
+    }
+
+    public function setDynamicQueryHandler(callable $handler): void
+    {
+        $this->queryDynamicHandler = $this->pipeline->with(
+            fn(QueryInput $input): mixed => ($this->queryExecutor)(
+                $input,
+                static fn(ValuesInterface $arguments): mixed => $handler($input->queryName, $arguments),
+            ),
+            /** @see WorkflowInboundCallsInterceptor::handleQuery() */
+            'handleQuery',
+        )(...);
+    }
+
+    public function setDynamicUpdateHandler(callable $handler, ?callable $validator = null): void
+    {
+        $this->updateDynamicValidator = $validator === null
+            ? null
+            : fn(UpdateInput $input): mixed => ($this->updateValidator)(
+                $input,
+                static fn(ValuesInterface $arguments): mixed => $validator($input->updateName, $arguments),
+            );
+
+        $this->updateDynamicHandler = $this->pipeline->with(
+            fn(UpdateInput $input, Deferred $deferred): mixed => ($this->updateExecutor)(
+                $input,
+                static fn(ValuesInterface $arguments): mixed => $handler($input->updateName, $arguments),
+                $deferred,
+            ),
+            /** @see WorkflowInboundCallsInterceptor::handleUpdate() */
+            'handleUpdate',
+        )(...);
+    }
+
     public function clearSignalQueue(): void
     {
         $this->signalQueue->clear();
@@ -290,6 +325,7 @@ public function destroy(): void
             $this->updateValidator,
             $this->prototype,
             $this->pipeline,
+            $this->queryDynamicHandler,
         );
         parent::destroy();
     }

src/Internal/Declaration/WorkflowInstance/SignalQueue.php

@@ -16,54 +16,90 @@
 /**
  * @psalm-type Consumer = callable(ValuesInterface): mixed
  *
- * @psalm-type OnSignalCallable = callable(non-empty-string $name, callable $handler, ValuesInterface $arguments): void
+ * @psalm-type OnSignalCallable = \Closure(non-empty-string $name, Consumer $handler, ValuesInterface $arguments): void
  */
 final class SignalQueue
 {
     /**
-     * @var array<string, list<ValuesInterface>>
+     * @var array<int, SignalQueueItem>
      */
     private array $queue = [];
 
     /**
-     * @var array<Consumer>
+     * @var array<non-empty-string, Consumer>
      */
     private array $consumers = [];
 
     /**
      * @var OnSignalCallable
      */
-    private $onSignal;
+    private \Closure $onSignal;
+
+    /**
+     * A fallback consumer to handle signals when no consumer is attached.
+     *
+     * @var null|\Closure(non-empty-string, ValuesInterface): mixed
+     */
+    private ?\Closure $dynamicConsumer = null;
 
     /**
      * @param non-empty-string $signal
      */
     public function push(string $signal, ValuesInterface $values): void
     {
         if (isset($this->consumers[$signal])) {
-            ($this->onSignal)($signal, $this->consumers[$signal], $values);
+            $this->consume($signal, $values, $this->consumers[$signal]);
             return;
         }
 
-        $this->queue[$signal][] = $values;
-        $this->flush($signal);
+        if ($this->dynamicConsumer !== null) {
+            $this->consumeFallback($signal, $values);
+            return;
+        }
+
+        $this->queue[] = new SignalQueueItem($signal, $values);
     }
 
     /**
      * @param OnSignalCallable $handler
      */
-    public function onSignal(callable $handler): void
+    public function onSignal(\Closure $handler): void
     {
         $this->onSignal = $handler;
     }
 
     /**
+     * @param non-empty-string $signal
      * @param Consumer $consumer
      */
     public function attach(string $signal, callable $consumer): void
     {
         $this->consumers[$signal] = $consumer; // overwrite
-        $this->flush($signal);
+
+        foreach ($this->queue as $k => $item) {
+            if ($item->name === $signal) {
+                unset($this->queue[$k]);
+                $this->consume($signal, $item->values, $consumer);
+            }
+        }
+    }
+
+    /**
+     * @param \Closure(non-empty-string, ValuesInterface): mixed $consumer
+     */
+    public function setFallback(\Closure $consumer): void
+    {
+        $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 clear(): void
@@ -72,18 +108,24 @@ public function clear(): void
     }
 
     /**
-     * @psalm-suppress UnusedVariable
+     * @param non-empty-string $signal
+     * @param Consumer $consumer
      */
-    private function flush(string $signal): void
+    private function consume(string $signal, ValuesInterface $values, callable $consumer): void
     {
-        if (!isset($this->queue[$signal], $this->consumers[$signal])) {
-            return;
-        }
+        ($this->onSignal)($signal, $consumer, $values);
+    }
 
-        while ($this->queue[$signal] !== []) {
-            $args = \array_shift($this->queue[$signal]);
+    /**
+     * @param non-empty-string $signal
+     */
+    private function consumeFallback(string $signal, ValuesInterface $values): void
+    {
+        $handler = $this->dynamicConsumer;
+        \assert($handler !== null);
 
-            ($this->onSignal)($signal, $this->consumers[$signal], $args);
-        }
+        // Wrap the fallback consumer to call interceptors
+        $consumer = static fn(ValuesInterface $values): mixed => $handler($signal, $values);
+        ($this->onSignal)($signal, $consumer, $values);
     }
 }