wamp

Author: ASKozienko

pkg/wamp/Tests/Functional/WampConsumerTest.php

@@ -35,7 +35,7 @@ public function testShouldSendAndReceiveMessage()
         $consumer->getClient()->getLoop()->futureTick(function () use ($producer, $topic, $message) {
             $producer->send($topic, $message);
         });
-        
+
         $receivedMessage = $consumer->receive(100);
 
         $this->assertInstanceOf(WampMessage::class, $receivedMessage);

pkg/wamp/Tests/Functional/WampSubscriptionConsumerTest.php

@@ -42,7 +42,7 @@ public function testShouldSendAndReceiveMessage()
         $consumer->getClient()->getLoop()->futureTick(function () use ($producer, $topic, $message) {
             $producer->send($topic, $message);
         });
-        
+
         $consumer->consume(100);
 
         $this->assertInstanceOf(WampMessage::class, $receivedMessage);

pkg/wamp/Tests/Spec/JsonSerializerTest.php

@@ -2,9 +2,9 @@
 
 namespace Enqueue\Wamp\Tests\Spec;
 
+use Enqueue\Test\ClassExtensionTrait;
 use Enqueue\Wamp\JsonSerializer;
 use Enqueue\Wamp\Serializer;
-use Enqueue\Test\ClassExtensionTrait;
 use Enqueue\Wamp\WampMessage;
 use PHPUnit\Framework\TestCase;
 
@@ -40,7 +40,7 @@ public function testThrowIfFailedToEncodeMessageToJson()
     {
         $serializer = new JsonSerializer();
 
-        $resource = fopen(__FILE__, 'r');
+        $resource = fopen(__FILE__, 'rb');
 
         //guard
         $this->assertInternalType('resource', $resource);

pkg/wamp/WampConsumer.php

@@ -67,7 +67,6 @@ public function receive(int $timeout = 0): ?Message
             $this->client = $this->context->getNewClient();
             $this->client->setAttemptRetry(true);
             $this->client->on('open', function (ClientSession $session) {
-
                 $session->subscribe($this->queue->getQueueName(), function ($args) {
                     $this->message = $this->context->getSerializer()->toMessage($args[0]);
 

pkg/wamp/WampProducer.php

@@ -51,7 +51,7 @@ public function __construct(WampContext $context)
      * {@inheritdoc}
      *
      * @param WampDestination $destination
-     * @param WampMessage $message
+     * @param WampMessage     $message
      */
     public function send(Destination $destination, Message $message): void
     {
@@ -86,7 +86,6 @@ public function send(Destination $destination, Message $message): void
             });
 
             $this->client->on('do-send', function (WampDestination $destination, WampMessage $message) {
-
                 $onFinish = function () {
                     $this->client->emit('do-stop');
                 };
@@ -113,25 +112,6 @@ public function send(Destination $destination, Message $message): void
         $this->client->getLoop()->run();
     }
 
-    private function doSendMessageIfPossible()
-    {
-        if (null === $this->session) {
-            return;
-        }
-
-        if (null === $this->message) {
-            return;
-        }
-
-        $message = $this->message;
-        $destination = $this->destination;
-
-        $this->message = null;
-        $this->destination = null;
-
-        $this->client->emit('do-send', [$destination, $message]);
-    }
-
     /**
      * {@inheritdoc}
      *
@@ -188,4 +168,23 @@ public function getTimeToLive(): ?int
     {
         return null;
     }
+
+    private function doSendMessageIfPossible()
+    {
+        if (null === $this->session) {
+            return;
+        }
+
+        if (null === $this->message) {
+            return;
+        }
+
+        $message = $this->message;
+        $destination = $this->destination;
+
+        $this->message = null;
+        $this->destination = null;
+
+        $this->client->emit('do-send', [$destination, $message]);
+    }
 }

pkg/wamp/WampSubscriptionConsumer.php

@@ -59,14 +59,13 @@ public function consume(int $timeout = 0): void
             $this->client = $this->context->getNewClient();
             $this->client->setAttemptRetry(true);
             $this->client->on('open', function (ClientSession $session) {
-
                 foreach ($this->subscribers as $queue => $subscriber) {
                     $session->subscribe($queue, function ($args) use ($subscriber) {
                         $message = $this->context->getSerializer()->toMessage($args[0]);
 
                         /**
                          * @var WampConsumer $consumer
-                         * @var callable $callback
+                         * @var callable     $callback
                          */
                         list($consumer, $callback) = $subscriber;