Adding event dispatcher

.phan/config.php

@@ -375,6 +375,7 @@
         'vendor/promphp/prometheus_client_php/src',
         'vendor/google/protobuf/src',
         'vendor/nyholm/dsn/src',
+        'vendor/cloudevents/sdk-php/src',
     ],
 
     // A list of individual files to include in analysis

composer.json

@@ -8,6 +8,7 @@
     "license": "Apache-2.0",
     "require": {
         "php": "^7.4 || ^8.0",
+        "cloudevents/sdk-php": "^v1.0.1",
         "ext-json": "*",
         "google/protobuf": "^3.3.0",
         "grpc/grpc": "^1.30",

deptrac.yaml

@@ -92,6 +92,10 @@ deptrac:
           collectors:
               - type: className
                 regex: ^Swoole\\*
+      -   name: CloudEvents
+          collectors:
+              - type: className
+                regex: ^CloudEvents\\*
 
     ruleset:
       Context:
@@ -101,6 +105,7 @@ deptrac:
           - Context
           - SemConv
           - PsrLog
+          - CloudEvents
       SDK:
           - +API
           - PsrHttp

src/API/Common/Event/Dispatcher.php

@@ -0,0 +1,63 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\API\Common\Event;
+
+use CloudEvents\V1\CloudEventInterface;
+use OpenTelemetry\Context\Context;
+use OpenTelemetry\Context\ContextKey;
+
+class Dispatcher implements DispatcherInterface
+{
+    private static ?ContextKey $key = null;
+    /** @var array<string, array<int, array<callable>>> */
+    private array $listeners = [];
+
+    public static function getInstance(): self
+    {
+        $key = self::getConstantKeyInstance();
+
+        return Context::getCurrent()->get($key) ?? self::createInstance();
+    }
+
+    public function dispatch(CloudEventInterface $event): void
+    {
+        $this->dispatchEvent($this->getListenersForEvent($event->getType()), $event);
+    }
+
+    public function listen(string $type, callable $listener, int $priority = 0): void
+    {
+        $this->listeners[$type][$priority][] = $listener;
+        ksort($this->listeners[$type]);
+    }
+
+    private function getListenersForEvent(string $key): iterable
+    {
+        foreach ($this->listeners[$key] as $listeners) {
+            foreach ($listeners as $listener) {
+                yield $listener;
+            }
+        }
+    }
+
+    private function dispatchEvent(iterable $listeners, CloudEventInterface $event): void
+    {
+        foreach ($listeners as $listener) {
+            $listener($event);
+        }
+    }
+
+    private static function getConstantKeyInstance(): ContextKey
+    {
+        return self::$key ??= new ContextKey(self::class);
+    }
+
+    private static function createInstance(): self
+    {
+        $dispatcher = new self();
+        Context::getCurrent()->with(self::getConstantKeyInstance(), $dispatcher)->activate();
+
+        return $dispatcher;
+    }
+}

src/API/Common/Event/DispatcherInterface.php

@@ -0,0 +1,13 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\API\Common\Event;
+
+use CloudEvents\V1\CloudEventInterface;
+
+interface DispatcherInterface
+{
+    public function dispatch(CloudEventInterface $event): void;
+    public function listen(string $type, callable $listener, int $priority = 0): void;
+}

src/API/Common/Event/EmitsEventsTrait.php

@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\API\Common\Event;
+
+use CloudEvents\V1\CloudEventInterface;
+
+trait EmitsEventsTrait
+{
+    protected static function emit(CloudEventInterface $event): void
+    {
+        Dispatcher::getInstance()->dispatch($event);
+    }
+}

src/API/composer.json

@@ -12,6 +12,7 @@
     ],
     "require": {
         "php": "^7.4 || ^8.0",
+        "cloudevents/sdk-php": "^v1.0.1",
         "open-telemetry/context": "self.version",
         "open-telemetry/sem-conv": "self.version",
         "psr/log": "^1.1|^2.0|^3.0"

tests/Benchmark/EventBench.php

@@ -0,0 +1,78 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\Tests\Benchmark;
+
+use CloudEvents\V1\CloudEvent;
+use CloudEvents\V1\CloudEventInterface;
+use Generator;
+use OpenTelemetry\API\Common\Event\Dispatcher;
+
+class EventBench
+{
+    private Dispatcher $dispatcher;
+    private $listener;
+    private CloudEventInterface $event;
+
+    public function __construct()
+    {
+        $this->dispatcher = Dispatcher::getInstance();
+        $this->listener = function () {
+        };
+        $this->event = new CloudEvent(uniqid(), self::class, 'foo');
+    }
+
+    public function addEvents(): void
+    {
+        for ($i=0; $i<10; $i++) {
+            $this->dispatcher->listen('event_' . $i, $this->listener);
+        }
+        $this->dispatcher->listen($this->event->getType(), $this->listener);
+    }
+
+    /**
+     * @ParamProviders("provideListenerCounts")
+     * @Revs(1000)
+     * @Iterations(10)
+     * @OutputTimeUnit("microseconds")
+     */
+    public function benchAddListeners(array $params): void
+    {
+        for ($i=0; $i<$params[0]; $i++) {
+            $this->dispatcher->listen('event_' . $i, $this->listener);
+        }
+    }
+
+    /**
+     * @ParamProviders("provideListenerCounts")
+     * @Revs(1000)
+     * @Iterations(10)
+     * @OutputTimeUnit("microseconds")
+     */
+    public function benchAddListenersForSameEvent(array $params): void
+    {
+        for ($i=0; $i<$params[0]; $i++) {
+            $this->dispatcher->listen('event', $this->listener);
+        }
+    }
+
+    /**
+     * @BeforeMethods("addEventsToListener")
+     * @Revs(1000)
+     * @Iterations(10)
+     * @OutputTimeUnit("microseconds")
+     */
+    public function benchDispatchEvent(): void
+    {
+        $this->dispatcher->dispatch($this->event);
+    }
+
+    public function provideListenerCounts(): Generator
+    {
+        yield [1];
+        yield [4];
+        yield [16];
+        yield [256];
+    }
+}

tests/Unit/API/Common/Event/DispatcherTest.php

@@ -0,0 +1,115 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\Tests\Unit\API\Common\Event;
+
+use CloudEvents\V1\CloudEventInterface;
+use OpenTelemetry\API\Common\Event\Dispatcher;
+use OpenTelemetry\Context\Context;
+use OpenTelemetry\Context\ContextKey;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * @covers \OpenTelemetry\API\Common\Event\Dispatcher
+ */
+class DispatcherTest extends TestCase
+{
+    private Dispatcher $dispatcher;
+    private CloudEventInterface $event;
+    private \ReflectionMethod $method;
+
+    public function setUp(): void
+    {
+        $this->dispatcher = new Dispatcher();
+
+        $reflection = new \ReflectionClass($this->dispatcher);
+        $this->method = $reflection->getMethod('getListenersForEvent');
+        $this->method->setAccessible(true);
+
+        $this->event = $this->createMock(CloudEventInterface::class);
+        $this->event->method('getType')->willReturn('foo');
+    }
+
+    public function test_get_instance(): void
+    {
+        $dispatcher = Dispatcher::getInstance();
+        $this->assertInstanceOf(Dispatcher::class, $dispatcher);
+        $this->assertSame($dispatcher, Dispatcher::getInstance());
+    }
+
+    public function test_get_instance_from_parent_context(): void
+    {
+        $dispatcher = Dispatcher::getInstance();
+        $this->assertInstanceOf(Dispatcher::class, $dispatcher);
+        $parent = Context::getCurrent()->with(new ContextKey('foo'), 'bar');
+        $parent->activate();
+        $this->assertSame($dispatcher, Dispatcher::getInstance());
+    }
+
+    public function test_add_listener(): void
+    {
+        $listenerFunction = function () {
+        };
+        $this->dispatcher->listen($this->event->getType(), $listenerFunction);
+        $listeners = [...$this->method->invokeArgs($this->dispatcher, [$this->event->getType()])];
+        $this->assertCount(1, $listeners);
+        $this->assertSame($listenerFunction, $listeners[0]);
+    }
+
+    public function test_dispatch_event(): void
+    {
+        $handler = function ($receivedEvent) {
+            $this->assertSame($this->event, $receivedEvent);
+        };
+        $this->dispatcher->listen($this->event->getType(), $handler);
+        $this->dispatcher->dispatch($this->event);
+    }
+
+    public function test_add_multiple_listeners_with_same_priority(): void
+    {
+        $listenerOne = function (CloudEventInterface $event) {
+        };
+        $listenerTwo = function (CloudEventInterface $event) {
+        };
+        $this->dispatcher->listen($this->event->getType(), $listenerOne);
+        $this->dispatcher->listen($this->event->getType(), $listenerTwo);
+        $listeners = [...$this->method->invokeArgs($this->dispatcher, [$this->event->getType()])];
+        $this->assertCount(2, $listeners);
+        $this->assertSame($listenerOne, $listeners[0]);
+        $this->assertSame($listenerTwo, $listeners[1]);
+    }
+
+    public function test_listener_priority(): void
+    {
+        $listenerOne = function () {
+        };
+        $listenerTwo = function () {
+        };
+        $listenerThree = function () {
+        };
+        $listenerFour = function () {
+        };
+        $this->dispatcher->listen($this->event->getType(), $listenerOne, 1);
+        $this->dispatcher->listen($this->event->getType(), $listenerTwo, -1);
+        $this->dispatcher->listen($this->event->getType(), $listenerThree, 0);
+        $this->dispatcher->listen($this->event->getType(), $listenerFour, 1);
+        $listeners = [...$this->method->invokeArgs($this->dispatcher, [$this->event->getType()])];
+        $this->assertCount(4, $listeners);
+        $this->assertSame($listenerTwo, $listeners[0]);
+        $this->assertSame($listenerThree, $listeners[1]);
+        $this->assertSame($listenerOne, $listeners[2]);
+        $this->assertSame($listenerFour, $listeners[3]);
+    }
+
+    public function test_add_listener_to_multiple_events(): void
+    {
+        $event = $this->createMock(CloudEventInterface::class);
+        $event->method('getType')->willReturn('bar');
+        $listener = function () {
+        };
+        $this->dispatcher->listen($this->event->getType(), $listener);
+        $this->dispatcher->listen($event->getType(), $listener);
+        $this->assertSame([$listener], [...$this->method->invokeArgs($this->dispatcher, [$this->event->getType()])]);
+    }
+}

tests/Unit/API/Common/Event/EmitsEventsTraitTest.php

@@ -0,0 +1,42 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\Tests\Unit\API\Common\Event;
+
+use CloudEvents\V1\CloudEventInterface;
+use OpenTelemetry\API\Common\Event\Dispatcher;
+use OpenTelemetry\API\Common\Event\EmitsEventsTrait;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * @covers \OpenTelemetry\API\Common\Event\EmitsEventsTrait
+ */
+class EmitsEventsTraitTest extends TestCase
+{
+    public function test_emits_event(): void
+    {
+        $event = $this->createMock(CloudEventInterface::class);
+        $event->method('getType')->willReturn('bar');
+        $called = false;
+        $class = $this->createInstance();
+        Dispatcher::getInstance()->listen($event->getType(), function () use (&$called) {
+            $this->assertTrue(true, 'listener was called');
+            $called = true;
+        });
+        $class->run('emit', $event);
+        $this->assertTrue($called);
+    }
+
+    private function createInstance(): object
+    {
+        return new class() {
+            use EmitsEventsTrait;
+            //accessor for protected trait methods
+            public function run(string $method, $param): void
+            {
+                $this->{$method}($param);
+            }
+        };
+    }
+}