wamp

Author: ASKozienko

docs/transport/wamp.md

@@ -8,6 +8,7 @@ It uses internally Thruway PHP library [voryx/thruway](https://github.com/voryx/
 * [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
@@ -58,6 +59,35 @@ while (true) {
 }
 ```
 
+## 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

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

@@ -165,7 +165,7 @@ public static function getKnownConnections(): array
                 'package' => 'enqueue/mongodb',
             ];
             $map[WampConnectionFactory::class] = [
-                'schemes' => ['wamp'],
+                'schemes' => ['wamp', 'ws'],
                 'supportedSchemeExtensions' => [],
                 'package' => 'enqueue/wamp',
             ];

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/wamp/WampConnectionFactory.php

@@ -93,7 +93,7 @@ private function parseDsn(string $dsn): array
     {
         $dsn = new Dsn($dsn);
 
-        if ('wamp' !== $dsn->getSchemeProtocol()) {
+        if (false === in_array($dsn->getSchemeProtocol(), ['wamp', 'ws'], true)) {
             throw new \LogicException(sprintf(
                 'The given scheme protocol "%s" is not supported. It must be "wamp"',
                 $dsn->getSchemeProtocol()