pre-release

.gitignore

@@ -0,0 +1,3 @@
+/vendor/
+.idea
+composer.lock

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2024 Mikhail Gerasimov
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,24 @@
+Symfony Messenger WebSocket Transport
+=====================================
+
+This package provides a WebSocket transport for
+the [Symfony Messenger](https://symfony.com/doc/current/components/messenger.html) component, enabling real-time
+communication between Symfony Messenger and external services via WebSocket connections.
+
+Installation
+------------
+
+Install the package via Composer:
+
+```bash
+composer require gri3li/symfony-messenger-websocket-transport
+```
+
+Usage
+-----
+
+Since all messages will be serialized and deserialized as instances of StdClass, you will most likely need to provide
+custom implementations of the interfaces:
+
+- `SendersLocatorInterface`: Defines which sender will be used for dispatching a message.
+- `HandlersLocatorInterface`: Defines which handler will process the message.

composer.json

@@ -0,0 +1,37 @@
+{
+  "name": "gri3li/symfony-messenger-websocket-transport",
+  "description": "WebSocket transport for Symfony Messenger",
+  "keywords": ["symfony-messenger", "symfony-messenger-transport"],
+  "type": "library",
+  "license": "MIT",
+  "minimum-stability": "dev",
+  "authors": [
+    {
+      "name": "Mikhail Gerasimov",
+      "email": "[email protected]"
+    }
+  ],
+  "require": {
+    "php": ">=8.2",
+    "symfony/messenger": "^6.4|^7.0",
+    "symfony/serializer": "^6.4|^7.0",
+    "gri3li/symfony-messenger-serializer-plain": "dev-main",
+    "amphp/websocket-client": "^2.0"
+  },
+  "autoload": {
+    "psr-4": { "Gri3li\\SymfonyMessengerWebSocketTransport\\": "src" }
+  },
+  "autoload-dev": {
+    "psr-4": {
+      "Gri3li\\SymfonyMessengerWebSocketTransport\\Tests\\": "tests/"
+    }
+  },
+  "scripts": {
+    "cs": "./vendor/bin/phpcs --standard=PSR12 src tests",
+    "unit": "./vendor/bin/phpunit ./tests"
+  },
+  "require-dev": {
+    "squizlabs/php_codesniffer": "^3.10",
+    "phpunit/phpunit": "^11.3"
+  }
+}

src/WebSocketReceiver.php

@@ -0,0 +1,40 @@
+<?php
+
+namespace Gri3li\SymfonyMessengerWebSocketTransport;
+
+use Amp\Websocket\Client\WebsocketConnection;
+use Amp\Websocket\WebsocketMessage;
+use Gri3li\SymfonyMessengerSerializerPlain\PlainSerializer;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Exception\TransportException;
+use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
+use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
+
+readonly class WebSocketReceiver implements ReceiverInterface
+{
+    public function __construct(
+        private WebsocketConnection $connection,
+        private SerializerInterface $serializer = new PlainSerializer(),
+    ) {
+    }
+
+    public function get(): iterable
+    {
+        try {
+            /** @var WebsocketMessage $message */
+            foreach ($this->connection->receive() as $message) {
+                yield $this->serializer->decode(['body' => (string)$message]);
+            }
+        } catch (\Throwable $throwable) {
+            throw new TransportException($throwable->getMessage(), 0, $throwable);
+        }
+    }
+
+    public function ack(Envelope $envelope): void
+    {
+    }
+
+    public function reject(Envelope $envelope): void
+    {
+    }
+}

src/WebSocketSender.php

@@ -0,0 +1,31 @@
+<?php
+
+namespace Gri3li\SymfonyMessengerWebSocketTransport;
+
+use Amp\Websocket\Client\WebsocketConnection;
+use Gri3li\SymfonyMessengerSerializerPlain\PlainSerializer;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Exception\TransportException;
+use Symfony\Component\Messenger\Transport\Sender\SenderInterface;
+use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
+
+readonly class WebSocketSender implements SenderInterface
+{
+    public function __construct(
+        private WebsocketConnection $connection,
+        private SerializerInterface $serializer = new PlainSerializer(),
+    ) {
+    }
+
+    public function send(Envelope $envelope): Envelope
+    {
+        try {
+            ['body' => $data] = $this->serializer->encode($envelope);
+            $this->connection->sendText($data);
+        } catch (\Throwable $throwable) {
+            throw new TransportException($throwable->getMessage(), 0, $throwable);
+        }
+
+        return $envelope;
+    }
+}

src/WebSocketTransport.php

@@ -0,0 +1,51 @@
+<?php
+
+namespace Gri3li\SymfonyMessengerWebSocketTransport;
+
+use Amp\Websocket\Client\WebsocketConnection;
+use Gri3li\SymfonyMessengerSerializerPlain\PlainSerializer;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
+use Symfony\Component\Messenger\Transport\TransportInterface;
+
+class WebSocketTransport implements TransportInterface
+{
+    private WebSocketReceiver $receiver;
+    private WebSocketSender $sender;
+
+    public function __construct(
+        private readonly WebsocketConnection $connection,
+        private readonly SerializerInterface $serializer = new PlainSerializer(),
+    ) {
+    }
+
+    public function send(Envelope $envelope): Envelope
+    {
+        return $this->getSender()->send($envelope);
+    }
+
+    public function get(): iterable
+    {
+        yield from $this->getReceiver()->get();
+    }
+
+    public function ack(Envelope $envelope): void
+    {
+        $this->getReceiver()->ack($envelope);
+    }
+
+    public function reject(Envelope $envelope): void
+    {
+        $this->getReceiver()->reject($envelope);
+    }
+
+    private function getReceiver(): WebSocketReceiver
+    {
+        return $this->receiver ??= new WebSocketReceiver($this->connection, $this->serializer);
+    }
+
+    private function getSender(): WebSocketSender
+    {
+        return $this->sender ??= new WebSocketSender($this->connection, $this->serializer);
+    }
+}

src/WebSocketTransportFactory.php

@@ -0,0 +1,32 @@
+<?php
+
+namespace Gri3li\SymfonyMessengerWebSocketTransport;
+
+use Amp\Websocket\Client;
+use Symfony\Component\Messenger\Exception\TransportException;
+use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
+use Symfony\Component\Messenger\Transport\TransportFactoryInterface;
+use Symfony\Component\Messenger\Transport\TransportInterface;
+
+class WebSocketTransportFactory implements TransportFactoryInterface
+{
+    public function createTransport(
+        #[\SensitiveParameter] string $dsn,
+        array $options,
+        SerializerInterface $serializer
+    ): TransportInterface {
+        unset($options['transport_name']);
+        try {
+            $connection = Client\connect($dsn);
+        } catch (\Throwable $throwable) {
+            throw new TransportException($throwable->getMessage(), 0, $throwable);
+        }
+
+        return new WebSocketTransport($connection, $serializer);
+    }
+
+    public function supports(#[\SensitiveParameter] string $dsn, array $options): bool
+    {
+        return str_starts_with($dsn, 'ws://') || str_starts_with($dsn, 'wss://');
+    }
+}

tests/WebSocketTransportTest.php

@@ -0,0 +1,50 @@
+<?php
+
+namespace Gri3li\SymfonyMessengerWebSocketTransport\Tests;
+
+use Amp\Websocket\Client;
+use Gri3li\SymfonyMessengerWebSocketTransport\WebSocketTransport;
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
+use Symfony\Component\Messenger\Exception\TransportException;
+
+class WebSocketTransportTest extends TestCase
+{
+    private const SEND_COUNT = 10;
+
+    public function testEcho(): void
+    {
+        $connection = Client\connect('wss://echo.websocket.in');
+        $transport = new WebSocketTransport($connection);
+        $asserts = [];
+        $i = 0;
+        while ($i < self::SEND_COUNT) {
+            $message = ['number' => $i];
+            $asserts[$i]['expect'] = $transport->send(new Envelope((object)$message));
+            $i++;
+        }
+        $j = 0;
+        while (true) {
+            try {
+                foreach ($transport->get() as $item) {
+                    if (is_object($item)) {
+                        $asserts[$j]['actual'] = $item;
+                        $j++;
+                    }
+                }
+            } catch (TransportException $exception) {
+                // Skipping exceptions caused by debug messages from websocket echo server
+                if (!$exception->getPrevious() instanceof MessageDecodingFailedException) {
+                    throw $exception;
+                }
+            }
+            if ($j >= self::SEND_COUNT) {
+                break;
+            }
+        }
+        foreach ($asserts as $assert) {
+            $this->assertEquals($assert['expect'], $assert['actual']);
+        }
+    }
+}