WAMP

bin/subtree-split

@@ -66,6 +66,7 @@ remote async-event-dispatcher [email protected]:php-enqueue/async-event-dispatcher.
 remote async-command [email protected]:php-enqueue/async-command.git
 remote mongodb [email protected]:php-enqueue/mongodb.git
 remote dsn [email protected]:php-enqueue/dsn.git
+remote wamp [email protected]:php-enqueue/wamp.git
 
 split 'pkg/enqueue' enqueue
 split 'pkg/simple-client' simple-client
@@ -90,3 +91,4 @@ split 'pkg/async-event-dispatcher' async-event-dispatcher
 split 'pkg/async-command' async-command
 split 'pkg/mongodb' mongodb
 split 'pkg/dsn' dsn
+split 'pkg/wamp' wamp

composer.json

@@ -30,7 +30,9 @@
     "php-http/guzzle6-adapter": "^1.1",
     "php-http/client-common": "^1.7@dev",
     "richardfullmer/rabbitmq-management-api": "^2.0",
-    "predis/predis": "^1.1"
+    "predis/predis": "^1.1",
+    "thruway/pawl-transport": "^0.5.0",
+    "voryx/thruway": "^0.5.3"
   },
   "require-dev": {
     "phpunit/phpunit": "^5.5",
@@ -75,6 +77,7 @@
       "Enqueue\\Stomp\\": "pkg/stomp/",
       "Enqueue\\Test\\": "pkg/test/",
       "Enqueue\\Dsn\\": "pkg/dsn/",
+      "Enqueue\\Wamp\\": "pkg/wamp/",
       "Enqueue\\": "pkg/enqueue/"
     },
     "exclude-from-classmap": [

docker-compose.yml

@@ -14,6 +14,7 @@ services:
       - google-pubsub
       - rabbitmqssl
       - mongo
+      - thruway
       - localstack
     volumes:
       - './:/mqdev'
@@ -30,6 +31,7 @@ services:
       - PHPREDIS_DSN=redis+phpredis://redis
       - GPS_DSN=gps:?projectId=mqdev&emulatorHost=http://google-pubsub:8085
       - SQS_DSN=sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4576&version=latest
+      - WAMP_DSN=wamp://thruway:9090
       - REDIS_HOST=redis
       - REDIS_PORT=6379
       - AWS_SQS_KEY=key
@@ -104,6 +106,17 @@ services:
       ports:
         - "27017:27017"
 
+  thruway:
+    image: formapro/nginx-php-fpm:latest-all-exts
+    ports:
+      - '9090:9090'
+    working_dir: '/app'
+    volumes:
+      - './:/app'
+    entrypoint:
+        - '/usr/bin/php'
+        - 'docker/thruway/WsRouter.php'
+
   localstack:
     image: 'localstack/localstack:latest'
     ports:

docker/bin/test.sh

@@ -37,6 +37,7 @@ waitForService beanstalkd 11300 50
 waitForService gearmand 4730 50
 waitForService kafka 9092 50
 waitForService mongo 27017 50
+waitForService thruway 9090 50
 waitForService localstack 4576 50
 
 php docker/bin/refresh-mysql-database.php

docker/thruway/WsRouter.php

@@ -0,0 +1,14 @@
+<?php
+
+require __DIR__.'/../../vendor/autoload.php';
+
+use Thruway\Peer\Router;
+use Thruway\Transport\RatchetTransportProvider;
+
+$router = new Router();
+
+$transportProvider = new RatchetTransportProvider('0.0.0.0', 9090);
+
+$router->addTransportProvider($transportProvider);
+
+$router->start();

docs/index.md

@@ -10,6 +10,7 @@
     - [Kafka](transport/kafka.md)
     - [Stomp](transport/stomp.md)
     - [Redis](transport/redis.md)
+    - [Wamp](transport/wamp.md)
     - [Doctrine DBAL](transport/dbal.md)
     - [Filesystem](transport/filesystem.md)
     - [Null](transport/null.md)

docs/transport/wamp.md

@@ -0,0 +1,103 @@
+# Web Application Messaging Protocol (WAMP) Transport
+
+A transport for [Web Application Messaging Protocol](https://wamp-proto.org/).
+WAMP is an open standard WebSocket subprotocol.
+It uses internally Thruway PHP library [voryx/thruway](https://github.com/voryx/Thruway) 
+
+* [Installation](#installation)
+* [Start the WAMP router](#start-the-wamp-router)
+* [Create context](#create-context)
+* [Consume message](#consume-message)
+* [Subscription consumer](#subscription-consumer)
+* [Send message to topic](#send-message-to-topic)
+
+## Installation
+
+```bash
+$ composer require enqueue/wamp
+```
+
+## Start the WAMP router
+
+```bash
+$ php vendor/voryx/thruway/Examples/SimpleWsRouter.php
+```
+
+Thruway is now running on 127.0.0.1 port 9090 
+
+
+## Create context
+
+```php
+<?php
+use Enqueue\Wamp\WampConnectionFactory;
+
+$connectionFactory = new WampConnectionFactory();
+
+$context = $connectionFactory->createContext();
+
+// if you have enqueue/enqueue library installed you can use a factory to build context from DSN 
+$context = (new \Enqueue\ConnectionFactoryFactory())->create('wamp:')->createContext();
+```
+
+## Consume message:
+
+Start message consumer before send message to the topic
+
+```php
+<?php
+/** @var \Enqueue\Wamp\WampContext $context */
+
+$fooTopic = $context->createTopic('foo');
+
+$consumer = $context->createConsumer($fooQueue);
+
+while (true) {
+    if ($message = $consumer->receive()) {
+        // process a message
+    }
+}
+```
+
+## Subscription consumer
+
+```php
+<?php
+use Interop\Queue\Message;
+use Interop\Queue\Consumer;
+
+/** @var \Enqueue\Wamp\WampContext $context */
+/** @var \Enqueue\Wamp\WampDestination $fooQueue */
+/** @var \Enqueue\Wamp\WampDestination $barQueue */
+
+$fooConsumer = $context->createConsumer($fooQueue);
+$barConsumer = $context->createConsumer($barQueue);
+
+$subscriptionConsumer = $context->createSubscriptionConsumer();
+$subscriptionConsumer->subscribe($fooConsumer, function(Message $message, Consumer $consumer) {
+    // process message
+
+    return true;
+});
+$subscriptionConsumer->subscribe($barConsumer, function(Message $message, Consumer $consumer) {
+    // process message
+
+    return true;
+});
+
+$subscriptionConsumer->consume(2000); // 2 sec
+```
+
+## Send message to topic
+
+```php
+<?php
+/** @var \Enqueue\Wamp\WampContext $context */
+
+$fooTopic = $context->createTopic('foo');
+$message = $context->createMessage('Hello world!');
+
+$context->createProducer()->send($fooTopic, $message);
+```
+
+[back to index](../index.md)

phpunit.xml.dist

@@ -104,6 +104,10 @@
         <testsuite name="dsn">
             <directory>pkg/dsn/Tests</directory>
         </testsuite>
+
+        <testsuite name="wamp transport">
+            <directory>pkg/wamp/Tests</directory>
+        </testsuite>
     </testsuites>
 
     <php>

pkg/enqueue/Resources.php

@@ -16,6 +16,7 @@
 use Enqueue\Redis\RedisConnectionFactory;
 use Enqueue\Sqs\SqsConnectionFactory;
 use Enqueue\Stomp\StompConnectionFactory;
+use Enqueue\Wamp\WampConnectionFactory;
 use Interop\Queue\ConnectionFactory;
 
 final class Resources
@@ -163,6 +164,11 @@ public static function getKnownConnections(): array
                 'supportedSchemeExtensions' => [],
                 'package' => 'enqueue/mongodb',
             ];
+            $map[WampConnectionFactory::class] = [
+                'schemes' => ['wamp', 'ws'],
+                'supportedSchemeExtensions' => [],
+                'package' => 'enqueue/wamp',
+            ];
 
             self::$knownConnections = $map;
         }

pkg/enqueue/Tests/ResourcesTest.php

@@ -4,6 +4,7 @@
 
 use Enqueue\Redis\RedisConnectionFactory;
 use Enqueue\Resources;
+use Enqueue\Wamp\WampConnectionFactory;
 use Interop\Queue\ConnectionFactory;
 use PHPUnit\Framework\TestCase;
 
@@ -127,4 +128,22 @@ public function testShouldAllowGetPreviouslyRegisteredConnection()
         $this->assertArrayHasKey('package', $connectionInfo);
         $this->assertSame('foo/bar', $connectionInfo['package']);
     }
+
+    public function testShouldHaveRegisteredWampConfiguration()
+    {
+        $availableConnections = Resources::getKnownConnections();
+
+        $this->assertInternalType('array', $availableConnections);
+        $this->assertArrayHasKey(WampConnectionFactory::class, $availableConnections);
+
+        $connectionInfo = $availableConnections[WampConnectionFactory::class];
+        $this->assertArrayHasKey('schemes', $connectionInfo);
+        $this->assertSame(['wamp', 'ws'], $connectionInfo['schemes']);
+
+        $this->assertArrayHasKey('supportedSchemeExtensions', $connectionInfo);
+        $this->assertSame([], $connectionInfo['supportedSchemeExtensions']);
+
+        $this->assertArrayHasKey('package', $connectionInfo);
+        $this->assertSame('enqueue/wamp', $connectionInfo['package']);
+    }
 }

pkg/test/WampExtension.php

@@ -0,0 +1,18 @@
+<?php
+
+namespace Enqueue\Test;
+
+use Enqueue\Wamp\WampConnectionFactory;
+use Enqueue\Wamp\WampContext;
+
+trait WampExtension
+{
+    private function buildWampContext(): WampContext
+    {
+        if (false == $dsn = getenv('WAMP_DSN')) {
+            throw new \PHPUnit_Framework_SkippedTestError('Functional tests are not allowed in this environment');
+        }
+
+        return (new WampConnectionFactory($dsn))->createContext();
+    }
+}

pkg/wamp/.gitignore

@@ -0,0 +1,6 @@
+*~
+/composer.lock
+/composer.phar
+/phpunit.xml
+/vendor/
+/.idea/

pkg/wamp/.travis.yml

@@ -0,0 +1,21 @@
+sudo: false
+
+git:
+    depth: 10
+
+language: php
+
+php:
+    - '7.1'
+    - '7.2'
+
+cache:
+    directories:
+        - $HOME/.composer/cache
+
+install:
+    - composer self-update
+    - composer install
+
+script:
+    - vendor/bin/phpunit --exclude-group=functional

pkg/wamp/JsonSerializer.php

@@ -0,0 +1,41 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Enqueue\Wamp;
+
+class JsonSerializer implements Serializer
+{
+    public function toString(WampMessage $message): string
+    {
+        $json = json_encode([
+            'body' => $message->getBody(),
+            'properties' => $message->getProperties(),
+            'headers' => $message->getHeaders(),
+        ]);
+
+        if (JSON_ERROR_NONE !== json_last_error()) {
+            throw new \InvalidArgumentException(sprintf(
+                'The malformed json given. Error %s and message %s',
+                json_last_error(),
+                json_last_error_msg()
+            ));
+        }
+
+        return $json;
+    }
+
+    public function toMessage(string $string): WampMessage
+    {
+        $data = json_decode($string, true);
+        if (JSON_ERROR_NONE !== json_last_error()) {
+            throw new \InvalidArgumentException(sprintf(
+                'The malformed json given. Error %s and message %s',
+                json_last_error(),
+                json_last_error_msg()
+            ));
+        }
+
+        return new WampMessage($data['body'], $data['properties'], $data['headers']);
+    }
+}

pkg/wamp/LICENSE

@@ -0,0 +1,20 @@
+The MIT License (MIT)
+Copyright (c) 2018 Forma-Pro
+
+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.