wamp

Author: ASKozienko

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/Tests/Functional/WampConsumerTest.php

@@ -0,0 +1,65 @@
+<?php
+
+namespace Enqueue\Wamp\Tests\Functional;
+
+use Enqueue\Test\WampExtension;
+use Enqueue\Wamp\WampMessage;
+use PHPUnit\Framework\TestCase;
+use Psr\Log\NullLogger;
+use Thruway\Logging\Logger;
+
+/**
+ * @group functional
+ * @group Wamp
+ */
+class WampConsumerTest extends TestCase
+{
+    use WampExtension;
+
+    public static function setUpBeforeClass()
+    {
+        Logger::set(new NullLogger());
+    }
+
+    public function testShouldSendAndReceiveMessage()
+    {
+        $context = $this->buildWampContext();
+        $topic = $context->createTopic('topic');
+        $consumer = $context->createConsumer($topic);
+        $producer = $context->createProducer();
+        $message = $context->createMessage('the body');
+
+        // init client
+        $consumer->receive(1);
+
+        $consumer->getClient()->getLoop()->futureTick(function () use ($producer, $topic, $message) {
+            $producer->send($topic, $message);
+        });
+        
+        $receivedMessage = $consumer->receive(100);
+
+        $this->assertInstanceOf(WampMessage::class, $receivedMessage);
+        $this->assertSame('the body', $receivedMessage->getBody());
+    }
+
+    public function testShouldSendAndReceiveNoWaitMessage()
+    {
+        $context = $this->buildWampContext();
+        $topic = $context->createTopic('topic');
+        $consumer = $context->createConsumer($topic);
+        $producer = $context->createProducer();
+        $message = $context->createMessage('the body');
+
+        // init client
+        $consumer->receive(1);
+
+        $consumer->getClient()->getLoop()->futureTick(function () use ($producer, $topic, $message) {
+            $producer->send($topic, $message);
+        });
+
+        $receivedMessage = $consumer->receiveNoWait();
+
+        $this->assertInstanceOf(WampMessage::class, $receivedMessage);
+        $this->assertSame('the body', $receivedMessage->getBody());
+    }
+}

pkg/wamp/Tests/Functional/WampSubscriptionConsumerTest.php

@@ -0,0 +1,51 @@
+<?php
+
+namespace Enqueue\Wamp\Tests\Functional;
+
+use Enqueue\Test\WampExtension;
+use Enqueue\Wamp\WampMessage;
+use PHPUnit\Framework\TestCase;
+use Psr\Log\NullLogger;
+use Thruway\Logging\Logger;
+
+/**
+ * @group functional
+ * @group Wamp
+ */
+class WampSubscriptionConsumerTest extends TestCase
+{
+    use WampExtension;
+
+    public static function setUpBeforeClass()
+    {
+        Logger::set(new NullLogger());
+    }
+
+    public function testShouldSendAndReceiveMessage()
+    {
+        $context = $this->buildWampContext();
+        $topic = $context->createTopic('topic');
+        $consumer = $context->createSubscriptionConsumer();
+        $producer = $context->createProducer();
+        $message = $context->createMessage('the body');
+
+        $receivedMessage = null;
+        $consumer->subscribe($context->createConsumer($topic), function ($message) use (&$receivedMessage) {
+            $receivedMessage = $message;
+
+            return false;
+        });
+
+        // init client
+        $consumer->consume(1);
+
+        $consumer->getClient()->getLoop()->futureTick(function () use ($producer, $topic, $message) {
+            $producer->send($topic, $message);
+        });
+        
+        $consumer->consume(100);
+
+        $this->assertInstanceOf(WampMessage::class, $receivedMessage);
+        $this->assertSame('the body', $receivedMessage->getBody());
+    }
+}

pkg/wamp/Tests/Spec/WampConnectionFactoryTest.php

@@ -0,0 +1,20 @@
+<?php
+
+namespace Enqueue\Wamp\Tests\Spec;
+
+use Enqueue\Wamp\WampConnectionFactory;
+use Interop\Queue\Spec\ConnectionFactorySpec;
+
+/**
+ * @group Wamp
+ */
+class WampConnectionFactoryTest extends ConnectionFactorySpec
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function createConnectionFactory()
+    {
+        return new WampConnectionFactory();
+    }
+}

pkg/wamp/Tests/Spec/WampContextTest.php

@@ -0,0 +1,23 @@
+<?php
+
+namespace Enqueue\Wamp\Tests\Spec;
+
+use Enqueue\Test\WampExtension;
+use Interop\Queue\Spec\ContextSpec;
+
+/**
+ * @group functional
+ * @group Wamp
+ */
+class WampContextTest extends ContextSpec
+{
+    use WampExtension;
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function createContext()
+    {
+        return $this->buildWampContext();
+    }
+}

pkg/wamp/Tests/Spec/WampMessageTest.php

@@ -0,0 +1,20 @@
+<?php
+
+namespace Enqueue\Wamp\Tests\Spec;
+
+use Enqueue\Wamp\WampMessage;
+use Interop\Queue\Spec\MessageSpec;
+
+/**
+ * @group Wamp
+ */
+class WampMessageTest extends MessageSpec
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function createMessage()
+    {
+        return new WampMessage();
+    }
+}

pkg/wamp/Tests/Spec/WampProducerTest.php

@@ -0,0 +1,23 @@
+<?php
+
+namespace Enqueue\Wamp\Tests\Spec;
+
+use Enqueue\Test\WampExtension;
+use Interop\Queue\Spec\ProducerSpec;
+
+/**
+ * @group functional
+ * @group Wamp
+ */
+class WampProducerTest extends ProducerSpec
+{
+    use WampExtension;
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function createProducer()
+    {
+        return $this->buildWampContext()->createProducer();
+    }
+}

pkg/wamp/Tests/Spec/WampQueueTest.php

@@ -0,0 +1,20 @@
+<?php
+
+namespace Enqueue\Wamp\Tests\Spec;
+
+use Enqueue\Wamp\WampDestination;
+use Interop\Queue\Spec\QueueSpec;
+
+/**
+ * @group Wamp
+ */
+class WampQueueTest extends QueueSpec
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function createQueue()
+    {
+        return new WampDestination(self::EXPECTED_QUEUE_NAME);
+    }
+}

pkg/wamp/Tests/Spec/WampTopicTest.php

@@ -0,0 +1,20 @@
+<?php
+
+namespace Enqueue\Wamp\Tests\Spec;
+
+use Enqueue\Wamp\WampDestination;
+use Interop\Queue\Spec\TopicSpec;
+
+/**
+ * @group Wamp
+ */
+class WampTopicTest extends TopicSpec
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function createTopic()
+    {
+        return new WampDestination(self::EXPECTED_TOPIC_NAME);
+    }
+}

pkg/wamp/WampConsumer.php

@@ -50,6 +50,11 @@ public function getQueue(): Queue
         return $this->queue;
     }
 
+    public function getClient(): ?Client
+    {
+        return $this->client;
+    }
+
     public function receive(int $timeout = 0): ?Message
     {
         $init = false;
@@ -59,7 +64,7 @@ public function receive(int $timeout = 0): ?Message
         if (null === $this->client) {
             $init = true;
 
-            $this->client = $this->context->getClient();
+            $this->client = $this->context->getNewClient();
             $this->client->setAttemptRetry(true);
             $this->client->on('open', function (ClientSession $session) {
 
@@ -80,7 +85,10 @@ public function receive(int $timeout = 0): ?Message
         }
 
         if ($timeout > 0) {
-            $this->timer = $this->client->getLoop()->addTimer($timeout / 1000, function () {
+            $timeout = $timeout / 1000;
+            $timeout = $timeout >= 0.1 ? $timeout : 0.1;
+
+            $this->timer = $this->client->getLoop()->addTimer($timeout, function () {
                 $this->client->emit('do-stop');
             });
         }