event dispatcher

.gitignore

@@ -22,3 +22,4 @@ composer.lock
 phpunit.phar
 .phpunit.result.cache
 
+*.local.php

benchmark/PublisherRedisBench.php

@@ -42,18 +42,17 @@ public function benchAsWhile(): void
 
         // range
         foreach (range(1, 100) as $item) {
-            $this->publisher
-                ->push(
-                    $schema,
-                    new QueueTask(
-                        target: QueueHandlerStub::class,
-                        arguments: [
-                            'id' => $item,
-                            'name' => 'bench range',
-                        ],
-                    ),
-                    QueueContext::make($schema)
-                );
+            $this->publisher->push(
+                $schema,
+                new QueueTask(
+                    target: QueueHandlerStub::class,
+                    arguments: [
+                        'id' => $item,
+                        'name' => 'bench range',
+                    ],
+                ),
+                QueueContext::make($schema)
+            );
         }
     }
 
@@ -72,11 +71,10 @@ public function benchAsBatch(): void
             );
         }
 
-        $this->publisher
-            ->pushBatch(
-                $schema,
-                $batch,
-                QueueContext::make($schema)
-            );
+        $this->publisher->pushBatch(
+            $schema,
+            $batch,
+            QueueContext::make($schema)
+        );
     }
 }

benchmark/PublisherValkeyBench.php

@@ -42,18 +42,17 @@ public function benchAsWhile(): void
 
         // range
         foreach (range(1, 100) as $item) {
-            $this->publisher
-                ->push(
-                    $schema,
-                    new QueueTask(
-                        target: QueueHandlerStub::class,
-                        arguments: [
-                            'id' => $item,
-                            'name' => 'bench range',
-                        ],
-                    ),
-                    QueueContext::make($schema)
-                );
+            $this->publisher->push(
+                $schema,
+                new QueueTask(
+                    target: QueueHandlerStub::class,
+                    arguments: [
+                        'id' => $item,
+                        'name' => 'bench range',
+                    ],
+                ),
+                QueueContext::make($schema)
+            );
         }
     }
 
@@ -72,11 +71,10 @@ public function benchAsBatch(): void
             );
         }
 
-        $this->publisher
-            ->pushBatch(
-                $schema,
-                $batch,
-                QueueContext::make($schema)
-            );
+        $this->publisher->pushBatch(
+            $schema,
+            $batch,
+            QueueContext::make($schema)
+        );
     }
 }

composer.json

@@ -18,7 +18,8 @@
     "php": "^8.3",
     "ext-redis": "*",
     "amphp/redis": "^2.0",
-    "kuaukutsu/queue-core": "^0.5.3"
+    "kuaukutsu/queue-core": "^0.5.3",
+    "symfony/console": "^7.3"
   },
   "require-dev": {
     "ext-pcntl": "*",

src/Builder.php

@@ -6,13 +6,16 @@
 
 use Closure;
 use Override;
+use Throwable;
 use Amp\Redis\RedisConfig;
 use Amp\Redis\RedisException;
 use kuaukutsu\queue\core\handler\FactoryInterface;
 use kuaukutsu\queue\core\handler\HandlerInterface;
 use kuaukutsu\queue\core\handler\Pipeline;
 use kuaukutsu\queue\core\interceptor\InterceptorInterface;
 use kuaukutsu\queue\core\BuilderInterface;
+use kuaukutsu\poc\queue\stream\event\EventDispatcher;
+use kuaukutsu\poc\queue\stream\event\EventSubscriberInterface;
 
 use function Amp\Redis\createRedisClient;
 
@@ -27,6 +30,14 @@ final class Builder implements BuilderInterface
 
     private HandlerInterface $handler;
 
+    /**
+     * @var EventSubscriberInterface[]
+     */
+    private array $eventSubscribers = [];
+
+    /**
+     * @var ?Closure(?string, Throwable):void
+     */
     private ?Closure $catch = null;
 
     /**
@@ -59,6 +70,13 @@ public function withStreamOptions(StreamOptions $options): self
         return $clone;
     }
 
+    public function withSubscribers(EventSubscriberInterface ...$subscribers): self
+    {
+        $clone = clone $this;
+        $clone->eventSubscribers = $subscribers;
+        return $clone;
+    }
+
     #[Override]
     public function withCatch(Closure $catch): BuilderInterface
     {
@@ -84,6 +102,12 @@ public function buildPublisher(): Publisher
     #[Override]
     public function buildConsumer(): Consumer
     {
-        return new Consumer(createRedisClient($this->config), $this->options, $this->handler, $this->catch);
+        return new Consumer(
+            createRedisClient($this->config),
+            $this->options,
+            new EventDispatcher($this->eventSubscribers),
+            $this->handler,
+            $this->catch,
+        );
     }
 }

src/Consumer.php

@@ -6,19 +6,21 @@
 
 use Closure;
 use Override;
+use Throwable;
 use Amp\Redis\RedisClient;
 use Revolt\EventLoop;
 use kuaukutsu\queue\core\exception\QueueConsumeException;
 use kuaukutsu\queue\core\handler\HandlerInterface;
 use kuaukutsu\queue\core\ConsumerInterface;
 use kuaukutsu\queue\core\SchemaInterface;
-use kuaukutsu\poc\queue\stream\internal\Context;
+use kuaukutsu\poc\queue\stream\event\EventDispatcher;
 use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup;
 use kuaukutsu\poc\queue\stream\internal\stream\RedisString;
 use kuaukutsu\poc\queue\stream\internal\workflow\TaskHandler;
 use kuaukutsu\poc\queue\stream\internal\workflow\WorkflowCatch;
 use kuaukutsu\poc\queue\stream\internal\workflow\WorkflowClaim;
 use kuaukutsu\poc\queue\stream\internal\workflow\WorkflowMain;
+use kuaukutsu\poc\queue\stream\internal\Context;
 
 /**
  * @api
@@ -29,9 +31,13 @@ final class Consumer implements ConsumerInterface
 
     private readonly TaskHandler $handler;
 
+    /**
+     * @param ?Closure(?string, Throwable):void $catch
+     */
     public function __construct(
         private readonly RedisClient $redis,
         private readonly StreamOptions $options,
+        private readonly EventDispatcher $eventDispatcher,
         HandlerInterface $handler,
         ?Closure $catch = null,
     ) {
@@ -47,7 +53,12 @@ public function consume(SchemaInterface $schema): void
         $stream = new RedisStreamGroup($this->redis, $this->options, $schema);
         $stream->create();
 
-        $this->ctx = new Context($schema, $stream, new RedisString($this->redis));
+        $this->ctx = new Context(
+            $schema,
+            $stream,
+            new RedisString($this->redis),
+            $this->eventDispatcher,
+        );
 
         EventLoop::queue(
             (new WorkflowMain($this->handler, $stream))(...),

src/event/CallbackEvent.php

@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace kuaukutsu\poc\queue\stream\event;
+
+use Override;
+
+final readonly class CallbackEvent implements EventInterface
+{
+    public function __construct(public string $id)
+    {
+    }
+
+    #[Override]
+    public function getMessage(): string
+    {
+        return sprintf('id: %s', $this->id);
+    }
+}

src/event/Event.php

@@ -0,0 +1,24 @@
+<?php
+
+declare(strict_types=1);
+
+namespace kuaukutsu\poc\queue\stream\event;
+
+enum Event: string
+{
+    case TimeoutCancellation = 'cancellation by timeout';
+
+    case MessageHandleError = 'error processing message';
+
+    case MessageCorruptedError = 'error unmarshaling message';
+
+    case MessageTimeoutCancellation = 'cancellation by timeout while processing message';
+
+    case MessageAck = 'message ACK';
+
+    case CallbackDeferred = 'callback registration';
+
+    case CallbackDone = 'callback done';
+
+    case RuntimeException = 'runtime exception';
+}

src/event/EventDispatcher.php

@@ -0,0 +1,45 @@
+<?php
+
+declare(strict_types=1);
+
+namespace kuaukutsu\poc\queue\stream\event;
+
+use Throwable;
+
+/**
+ * @psalm-internal kuaukutsu\poc\queue\stream
+ */
+final readonly class EventDispatcher
+{
+    /**
+     * @var array<string, list<callable(Event $name, EventInterface $event):void>>
+     */
+    private array $eventHandlers;
+
+    /**
+     * @param EventSubscriberInterface[] $eventSubscribers
+     */
+    public function __construct(array $eventSubscribers)
+    {
+        $subscriptions = [];
+        foreach ($eventSubscribers as $subscriber) {
+            foreach ($subscriber->subscriptions() as $name => $callback) {
+                $subscriptions[$name][] = $callback;
+            }
+        }
+
+        $this->eventHandlers = $subscriptions;
+    }
+
+    public function trigger(Event $name, EventInterface $event): void
+    {
+        if (array_key_exists($name->name, $this->eventHandlers)) {
+            foreach ($this->eventHandlers[$name->name] as $subscriberCallback) {
+                try {
+                    $subscriberCallback($name, $event);
+                } catch (Throwable) {
+                }
+            }
+        }
+    }
+}

src/event/EventInterface.php

@@ -0,0 +1,10 @@
+<?php
+
+declare(strict_types=1);
+
+namespace kuaukutsu\poc\queue\stream\event;
+
+interface EventInterface
+{
+    public function getMessage(): string;
+}

src/event/EventPublisherInterface.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace kuaukutsu\poc\queue\stream\event;
+
+interface EventPublisherInterface
+{
+    public function on(EventSubscriberInterface $subscriber): void;
+
+    public function off(EventSubscriberInterface $subscriber): void;
+}

src/event/EventSubscriberInterface.php

@@ -0,0 +1,13 @@
+<?php
+
+declare(strict_types=1);
+
+namespace kuaukutsu\poc\queue\stream\event;
+
+interface EventSubscriberInterface
+{
+    /**
+     * @return array<string, callable(Event $name, EventInterface $event):void>
+     */
+    public function subscriptions(): array;
+}

src/event/MessageAckEvent.php

@@ -0,0 +1,23 @@
+<?php
+
+declare(strict_types=1);
+
+namespace kuaukutsu\poc\queue\stream\event;
+
+use Override;
+
+final readonly class MessageAckEvent implements EventInterface
+{
+    /**
+     * @param non-empty-string[] $identityList
+     */
+    public function __construct(public array $identityList)
+    {
+    }
+
+    #[Override]
+    public function getMessage(): string
+    {
+        return sprintf('COUNT: %d, %s', count($this->identityList), implode(', ', $this->identityList));
+    }
+}