Initial commit

src/Command.php

@@ -0,0 +1,40 @@
+<?php
+
+namespace NoccyLabs\React\CommandBus;
+
+use React\Promise\Deferred;
+use React\Promise\Promise;
+use React\Promise\PromiseInterface;
+
+/**
+ *
+ */
+class Command
+{
+    /** @var string $name The command name */
+    private string $name;
+
+    /** @var callable $handler The handler */
+    private $handler;
+
+    public function __construct(string $name, callable $handler)
+    {
+        $this->name = $name;
+        $this->handler = $handler;
+    }
+
+    public function getName(): string
+    {
+        return $this->name;
+    }
+
+    public function call(Context $context): PromiseInterface
+    {
+        return new Promise(function (callable $resolve) use ($context) {
+            $resolve(call_user_func($this->handler, $context));
+            return;
+        });
+    }
+
+}
+

src/CommandBus.php

@@ -0,0 +1,115 @@
+<?php
+
+namespace NoccyLabs\React\CommandBus;
+use Evenement\EventEmitterTrait;
+use React\Promise\PromiseInterface;
+use React\Promise\Promise;
+use React\Promise\Deferred;
+use React\Socket\ServerInterface;
+use React\Stream\DuplexStreamInterface;
+use SplObjectStorage;
+
+class CommandBus implements CommandBusInterface
+{
+    use EventEmitterTrait;
+
+    private CommandRegistry $commandRegistry;
+
+    private SplObjectStorage $connections;
+
+    private SplObjectStorage $servers;
+
+    public function __construct(CommandRegistry $commandRegistry)
+    {
+        $this->commandRegistry = $commandRegistry;
+        $this->connections = new SplObjectStorage();
+        $this->servers = new SplObjectStorage();
+    }
+
+    public function addServer(ServerInterface $server): void
+    {
+        $this->servers->attach($server);
+        $server->on('connection', $this->onServerConnection(...));
+    }
+
+    public function removeServer(ServerInterface $server): void
+    {
+        $server->close();
+        $server->removeListener('connection', $this->onServerConnection(...));
+        $this->servers->detach($server);
+    }
+
+    public function close(): void
+    {
+        foreach ($this->servers as $server) {
+            $this->removeServer($server);
+        }
+    }
+
+    private function onServerConnection(DuplexStreamInterface $client): void
+    {
+        $this->connections->attach($client);
+        $client->on('data', function ($data) use ($client) {
+            try {
+                $message = Message::fromString($data);
+                $this->onClientMessage($client, $message);
+            } catch (MessageException $e) {
+                $client->end('{"msg":"error","data":{"error":"Bad message format"}}');
+            }
+        });
+        $client->on('close', function () use ($client) {
+            $this->connections->detach($client);
+        });
+    }
+
+    private function onClientMessage(DuplexStreamInterface $client, Message $message): void
+    {
+        switch ($message->getType()) {
+            case Message::MSGTYPE_EXECUTE: // Client call to execute command
+                $data = $message->getData();
+                $context = new Context($data['command'],$data['context']);
+                $this->executeContext($context)->then(
+                    function ($result) use ($message, $client) {
+                        $client->write($message->asResult($result)->toJson()."\n");
+                    }
+                );
+                break;
+            case Message::MSGTYPE_RESULT: // Result from execution on client
+                // TODO implement me
+                break;
+                case Message::MSGTYPE_REGISTRY: // command registry actions
+                // TODO implement me
+                break;
+            default:
+                $client->end('{"msg":"error","data":{"error":"Unexpected message type"}}');
+        }
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function execute(string $command, array $context): PromiseInterface
+    {
+        $context = new Context($command, $context);
+        return $this->executeContext($context);
+    }
+
+    private function executeContext(Context $context): PromiseInterface
+    {
+        $command = $this->commandRegistry->find($context->getCommandName());
+        if (!$command) return new Promise(function (callable $resolve) use ($context) {
+            throw new \RuntimeException("Unable to resolve command: ".$context->getCommandName());
+        });
+
+        return $command->call($context);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function notify(string $event, array $data): void
+    {
+    }
+
+}
+

src/CommandBusClient.php

@@ -0,0 +1,150 @@
+<?php
+
+namespace NoccyLabs\React\CommandBus;
+use Evenement\EventEmitterTrait;
+use React\EventLoop\Loop;
+use React\Promise\PromiseInterface;
+use React\Promise\Promise;
+use React\Promise\Deferred;
+use React\Socket\ConnectorInterface;
+use React\Socket\TcpConnector;
+use React\Stream\DuplexStreamInterface;
+use Throwable;
+
+class CommandBusClient implements CommandBusInterface
+{
+    use EventEmitterTrait;
+
+    const EVENT_ERROR = 'error';
+    const EVENT_CONNECTED = 'connected';
+    const EVENT_DISCONNECTED = 'disconnected';
+
+    private ?CommandRegistry $commandRegistry = null;
+
+    private ConnectorInterface $connector;
+
+    private ?DuplexStreamInterface $connection = null;
+
+    /** @var array<string,Deferred> */
+    private array $pending = [];
+
+    private string $address;
+
+    public function __construct(?CommandRegistry $commandRegistry = null, ?ConnectorInterface $connector = null)
+    {
+        $this->commandRegistry = $commandRegistry;
+        if ($commandRegistry) {
+            $commandRegistry->on(CommandRegistry::EVENT_REGISTERED, $this->onCommandRegistered(...));
+            $commandRegistry->on(CommandRegistry::EVENT_UNREGISTERED, $this->onCommandUnregistered(...));
+        }
+        $this->connector = $connector??new TcpConnector();
+    }
+
+    public function connect(string $address): void
+    {
+        $this->address = $address;
+        $this->reconnect();
+    }
+
+    public function close(): void
+    {
+        $this->connection->close();
+        $this->connection->removeAllListeners();
+        $this->connection = null;
+    }
+
+    private function reconnect(): void
+    {
+        if ($this->connection) {
+            $this->close();
+        }
+
+        $this->connector->connect($this->address)->then(
+            function (DuplexStreamInterface $connection) {
+                $this->connection = $connection;
+                $this->emit(self::EVENT_CONNECTED);
+                $connection->on('close', function () {
+                    $this->emit(self::EVENT_DISCONNECTED);
+                });
+                $connection->on('data', function ($data) use ($connection) {
+                    try {
+                        $message = Message::fromString($data);
+                        $this->onServerMessage($message);
+                    } catch (MessageException $e) {
+                        $connection->end('{"msg":"error","data":{"error":"Bad message format"}}');
+                    }        
+                });
+            },
+            function (Throwable $e) {
+                $this->emit(self::EVENT_ERROR, [ $e->getMessage(), $e ]);
+                Loop::addTimer(5, $this->reconnect(...));
+            }
+        );
+    }
+
+    private function onServerMessage(Message $message): void
+    {
+        // fprintf(STDERR, "onServerMessage: %s\n", $message->toJson());
+        switch ($message->getType()) {
+            case Message::MSGTYPE_EXECUTE: // server call to execute command
+                // TODO implement me
+                break;
+            case Message::MSGTYPE_RESULT: // result from server
+                $uuid = $message->getUuid();
+                $result = $message->getData()['result'];
+                if (array_key_exists($uuid, $this->pending)) {
+                    $this->pending[$uuid]->resolve($result);
+                    unset($this->pending[$uuid]);
+                }
+                break;
+            case Message::MSGTYPE_REGISTRY: // command registry actions
+                // TODO implement me
+                break;
+            default:
+                $this->connection->end('{"msg":"error","data":{"error":"Unexpected message type"}}');
+        }
+    }
+
+    private function onCommandRegistered(string $command): void
+    {
+        if ($this->connection && $this->connection->isWritable()) {
+            $msg = new Message(Message::MSGTYPE_REGISTRY, [ 'register' => [ $command ]]);
+            $this->connection->write($msg->toJson());
+        }
+    }
+
+    private function onCommandUnregistered(string $command): void
+    {
+        if ($this->connection && $this->connection->isWritable()) {
+            $msg = new Message(Message::MSGTYPE_REGISTRY, [ 'unregister' => [ $command ]]);
+            $this->connection->write($msg->toJson());
+        }
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function execute(string $command, array $context): PromiseInterface
+    {
+        $deferred = new Deferred();
+
+        $message = new Message(Message::MSGTYPE_EXECUTE, [
+            'command' => $command,
+            'context' => $context
+        ]);
+        $this->pending[$message->getUuid()] = $deferred;
+
+        // fprintf(STDERR, "write: %s\n", $message->toJson());
+        $this->connection->write($message->toJson()."\n");
+
+        return $deferred->promise();
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function notify(string $event, array $data): void
+    {
+    }
+}
+

src/CommandBusException.php

@@ -0,0 +1,8 @@
+<?php
+
+namespace NoccyLabs\React\CommandBus;
+
+interface CommandBusException
+{
+}
+

src/CommandBusInterface.php

@@ -0,0 +1,33 @@
+<?php
+
+namespace NoccyLabs\React\CommandBus;
+use Evenement\EventEmitterInterface;
+use React\Promise\PromiseInterface;
+
+interface CommandBusInterface extends EventEmitterInterface
+{
+    /** @var string Emitted to distribrute notifications */
+    const EVENT_NOTIFY = "notify";
+
+    /**
+     * Execute a command on the command bus, and return a promise that will be
+     * resolved once the command has been handled.
+     *
+     * @param string $command The name of the command to execute
+     * @param array $context Data to pass in the call context
+     * @®eturn PromiseInterface A promise that will be resolved with the result
+     */
+    public function execute(string $command, array $context): PromiseInterface;
+
+    /**
+     * Notify all connected clients to emit the notify event, and also be
+     * emitted locally. These messages are unsolicited, and can not be
+     * responded to. They are fire-and-forget.
+     *
+     * @param string $event The application-specific event name
+     * @param array $data Event-specific data
+     * @return void
+     */
+    public function notify(string $event, array $data): void;
+}
+

src/CommandRegistry.php

@@ -0,0 +1,53 @@
+<?php
+
+namespace NoccyLabs\React\CommandBus;
+use Evenement\EventEmitterInterface;
+use Evenement\EventEmitterTrait;
+
+/**
+ * A collection of commands that can be executed via CommandBusInterface
+ *
+ */
+class CommandRegistry implements EventEmitterInterface
+{
+    use EventEmitterTrait;
+
+    const EVENT_REGISTERED = 'registered';
+    const EVENT_UNREGISTERED = 'unregistered';
+
+    /** @var array<string,Command> */
+    private array $commands = [];
+
+    public function register(string $command, callable $handler): void
+    {
+        $isNew = !array_key_exists($command, $this->commands);
+
+        $this->commands[$command] = new Command($command, $handler);
+
+        if ($isNew) {
+            $this->emit(self::EVENT_REGISTERED, [ $command ]);
+        }
+    }
+
+    public function unregister(string $command): void
+    {
+        if (!array_key_exists($command, $this->commands)) {
+            return;
+        }
+
+        unset($this->commands[$command]);
+
+        $this->emit(self::EVENT_UNREGISTERED, [ $command ]);
+    }
+
+    public function find(string $command): ?Command
+    {
+        return $this->commands[$command] ?? null;
+    }
+
+    public function getNames(): array
+    {
+        return array_keys($this->commands);
+    }
+}
+

src/Context.php

@@ -0,0 +1,41 @@
+<?php
+
+namespace NoccyLabs\React\CommandBus;
+
+
+/**
+ * The context contains all the data relating to calling a command. It has an
+ * unique identifier, the command to call, and the payload data.
+ *
+ */
+class Context
+{
+
+    /** @var string $commandName The command name to call */
+    private string $commandName;
+
+    /** @var array<string,mixed> The payload data */
+    private array $payload = [];
+
+    public function __construct(string $commandName, array $payload)
+    {
+        $this->commandName = $commandName;
+        $this->payload = $payload;
+    }
+
+    public function getCommandName(): string
+    {
+        return $this->commandName;
+    }
+
+    public function getPayload(): array
+    {
+        return $this->payload;
+    }
+
+    public function __get($name)
+    {
+        return $this->payload[$name] ?? null;   
+    }
+}
+

src/Message.php

@@ -0,0 +1,88 @@
+<?php
+
+namespace NoccyLabs\React\CommandBus;
+
+use JsonSerializable;
+use Symfony\Component\Uid\Uuid;
+
+/**
+ * A serializable message frame
+ *
+ */
+class Message implements JsonSerializable
+{
+    /** @var string Execute request */
+    const MSGTYPE_EXECUTE = 'execute';
+    /** @var string Execute result */
+    const MSGTYPE_RESULT = 'result';
+    /** @var string Notify event */
+    const MSGTYPE_NOTIFY = 'notify';
+    /** @var string Registry update (command list set and update) */
+    const MSGTYPE_REGISTRY = 'registry';
+
+    /** @var string $uuid The message identifier */
+    private string $uuid;
+
+    private string $messageType;
+
+    private array $messageData;
+
+    public function __construct(string $messageType, array $messageData = [], ?string $uuid = null)
+    {
+        $this->uuid = ($uuid && Uuid::isValid($uuid)) ? $uuid : (string)Uuid::v7();
+        $this->messageType = $messageType;
+        $this->messageData = $messageData;
+    }
+
+    public function getUuid(): string
+    {
+        return $this->uuid;
+    }
+
+    public function getType(): string
+    {
+        return $this->messageType;
+    }
+
+    public function getData(): array
+    {
+        return $this->messageData;
+    }
+
+    /**
+     *
+     * @param string $data The JSON-encoded message data
+     * @return Message
+     * @throws MessageException if the message can not be parsed
+     */
+    public static function fromString(string $data): Message
+    {
+        $json = @json_decode($data, true);
+        if (!$json || empty($json['msg'])) {
+            throw new MessageException("Invalid data");
+        }
+        return new Message($json['msg'], $json['data'], $json['uuid']);
+    }
+
+    public function asResult($result): Message
+    {
+        return new Message(self::MSGTYPE_RESULT, [
+            'result' => $result
+        ], $this->uuid);
+    }
+
+    public function toJson(): string
+    {
+        return json_encode($this, JSON_UNESCAPED_SLASHES);
+    }
+
+    public function jsonSerialize(): array
+    {
+        return [
+            'uuid' => $this->uuid,
+            'msg' => $this->messageType,
+            'data' => $this->messageData
+        ];
+    }
+}
+

src/MessageException.php

@@ -0,0 +1,9 @@
+<?php
+
+namespace NoccyLabs\React\CommandBus;
+use RuntimeException;
+
+class MessageException extends RuntimeException implements CommandBusException
+{
+}
+