mongodb transport, fixes

JSON.php

@@ -0,0 +1,59 @@
+<?php
+
+namespace Enqueue\Mongodb;
+
+class JSON
+{
+    /**
+     * @param string $string
+     *
+     * @throws \InvalidArgumentException
+     *
+     * @return array
+     */
+    public static function decode($string)
+    {
+        if (!is_string($string)) {
+            throw new \InvalidArgumentException(sprintf(
+                'Accept only string argument but got: "%s"',
+                is_object($string) ? get_class($string) : gettype($string)
+            ));
+        }
+
+        // PHP7 fix - empty string and null cause syntax error
+        if (empty($string)) {
+            return null;
+        }
+
+        $decoded = 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 $decoded;
+    }
+
+    /**
+     * @param mixed $value
+     *
+     * @return string
+     */
+    public static function encode($value)
+    {
+        $encoded = json_encode($value, JSON_UNESCAPED_UNICODE);
+
+        if (JSON_ERROR_NONE !== json_last_error()) {
+            throw new \InvalidArgumentException(sprintf(
+                'Could not encode value into json. Error %s and message %s',
+                json_last_error(),
+                json_last_error_msg()
+            ));
+        }
+
+        return $encoded;
+    }
+}

MongodbConnectionFactory.php

@@ -16,7 +16,7 @@ class MongodbConnectionFactory implements PsrConnectionFactory
      * The config could be an array, string DSN or null. In case of null it will attempt to connect to Mongodb localhost with default credentials.
      *
      * $config = [
-     *   'uri' => 'mongodb://127.0.0.1/' - Mongodb connection string. see http://docs.mongodb.org/manual/reference/connection-string/
+     *   'dsn' => 'mongodb://127.0.0.1/' - Mongodb connection string. see http://docs.mongodb.org/manual/reference/connection-string/
      *   'dbname' => 'enqueue',          - database name.
      *   'collection_name' => 'enqueue'  - collection name
      *   'polling_interval' => '1000',   - How often query for new messages (milliseconds)
@@ -39,7 +39,7 @@ public function __construct($config = 'mongodb:')
             throw new \LogicException('The config must be either an array of options, a DSN string or null');
         }
         $config = array_replace([
-            'uri' => 'mongodb://127.0.0.1/',
+            'dsn' => 'mongodb://127.0.0.1/',
             'dbname' => 'enqueue',
             'collection_name' => 'enqueue',
         ], $config);
@@ -49,7 +49,7 @@ public function __construct($config = 'mongodb:')
 
     public function createContext()
     {
-        $client = new Client($this->config['uri']);
+        $client = new Client($this->config['dsn']);
 
         return new MongodbContext($client, $this->config);
     }
@@ -75,10 +75,10 @@ public static function parseDsn($dsn)
         }
         if ('mongodb:' === $dsn) {
             return [
-                'uri' => 'mongodb://127.0.0.1/',
+                'dsn' => 'mongodb://127.0.0.1/',
             ];
         }
-        $config['uri'] = $dsn;
+        $config['dsn'] = $dsn;
         if (isset($parsedUrl['path']) && '/' !== $parsedUrl['path']) {
             $pathParts = explode('/', $parsedUrl['path']);
             //DB name

MongodbConsumer.php

@@ -163,7 +163,10 @@ protected function receiveMessage()
      */
     protected function convertMessage(array $mongodbMessage)
     {
-        $message = $this->context->createMessage($mongodbMessage['body'], $mongodbMessage['properties'], $mongodbMessage['headers']);
+        $properties = JSON::decode($mongodbMessage['properties']);
+        $headers = JSON::decode($mongodbMessage['headers']);
+
+        $message = $this->context->createMessage($mongodbMessage['body'], $properties, $headers);
         $message->setId((string) $mongodbMessage['_id']);
         $message->setPriority((int) $mongodbMessage['priority']);
         $message->setRedelivered((bool) $mongodbMessage['redelivered']);

MongodbProducer.php

@@ -80,8 +80,8 @@ public function send(PsrDestination $destination, PsrMessage $message)
         $mongoMessage = [
             'published_at' => $publishedAt,
             'body' => $body,
-            'headers' => $message->getHeaders(),
-            'properties' => $message->getProperties(),
+            'headers' => JSON::encode($message->getHeaders()),
+            'properties' => JSON::encode($message->getProperties()),
             'priority' => $message->getPriority(),
             'queue' => $destination->getName(),
             'redelivered' => $message->isRedelivered(),

Tests/Functional/MongodbConsumerTest.php

@@ -4,15 +4,15 @@
 
 use Enqueue\Mongodb\MongodbContext;
 use Enqueue\Mongodb\MongodbMessage;
-use Enqueue\Mongodb\Tests\Spec\CreateMongodbContextTrait;
+use Enqueue\Test\MongodbExtensionTrait;
 use PHPUnit\Framework\TestCase;
 
 /**
  * @group functional
  */
 class MongodbConsumerTest extends TestCase
 {
-    use CreateMongodbContextTrait;
+    use MongodbExtensionTrait;
 
     /**
      * @var MongodbContext
@@ -21,7 +21,7 @@ class MongodbConsumerTest extends TestCase
 
     public function setUp()
     {
-        $this->context = $this->createMongodbContext();
+        $this->context = $this->buildMongodbContext();
     }
 
     protected function tearDown()

Tests/MongodbConnectionFactoryTest.php

@@ -22,7 +22,7 @@ public function testShouldImplementConnectionFactoryInterface()
     public function testCouldBeConstructedWithEmptyConfiguration()
     {
         $params = [
-            'uri' => 'mongodb://127.0.0.1/',
+            'dsn' => 'mongodb://127.0.0.1/',
             'dbname' => 'enqueue',
             'collection_name' => 'enqueue',
         ];
@@ -34,7 +34,7 @@ public function testCouldBeConstructedWithEmptyConfiguration()
     public function testCouldBeConstructedWithCustomConfiguration()
     {
         $params = [
-            'uri' => 'mongodb://127.0.0.3/',
+            'dsn' => 'mongodb://127.0.0.3/',
             'uriOptions' => ['testValue' => 123],
             'driverOptions' => ['testValue' => 123],
             'dbname' => 'enqueue',

Tests/Spec/MongodbContextTest.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Mongodb\Tests\Spec;
 
+use Enqueue\Test\MongodbExtensionTrait;
 use Interop\Queue\Spec\PsrContextSpec;
 
 /**
@@ -10,13 +11,13 @@
  */
 class MongodbContextTest extends PsrContextSpec
 {
-    use CreateMongodbContextTrait;
+    use MongodbExtensionTrait;
 
     /**
      * {@inheritdoc}
      */
     protected function createContext()
     {
-        return $this->createMongodbContext();
+        return $this->buildMongodbContext();
     }
 }

Tests/Spec/MongodbProducerTest.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Mongodb\Tests\Spec;
 
+use Enqueue\Test\MongodbExtensionTrait;
 use Interop\Queue\Spec\PsrProducerSpec;
 
 /**
@@ -10,13 +11,13 @@
  */
 class MongodbProducerTest extends PsrProducerSpec
 {
-    use CreateMongodbContextTrait;
+    use MongodbExtensionTrait;
 
     /**
      * {@inheritdoc}
      */
     protected function createProducer()
     {
-        return $this->createMongodbContext()->createProducer();
+        return $this->buildMongodbContext()->createProducer();
     }
 }

Tests/Spec/MongodbRequeueMessageTest.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Mongodb\Tests\Spec;
 
+use Enqueue\Test\MongodbExtensionTrait;
 use Interop\Queue\Spec\RequeueMessageSpec;
 
 /**
@@ -10,13 +11,13 @@
  */
 class MongodbRequeueMessageTest extends RequeueMessageSpec
 {
-    use CreateMongodbContextTrait;
+    use MongodbExtensionTrait;
 
     /**
      * {@inheritdoc}
      */
     protected function createContext()
     {
-        return $this->createMongodbContext();
+        return $this->buildMongodbContext();
     }
 }

Tests/Spec/MongodbSendAndReceiveDelayedMessageFromQueueTest.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Mongodb\Tests\Spec;
 
+use Enqueue\Test\MongodbExtensionTrait;
 use Interop\Queue\Spec\SendAndReceiveDelayedMessageFromQueueSpec;
 
 /**
@@ -10,13 +11,13 @@
  */
 class MongodbSendAndReceiveDelayedMessageFromQueueTest extends SendAndReceiveDelayedMessageFromQueueSpec
 {
-    use CreateMongodbContextTrait;
+    use MongodbExtensionTrait;
 
     /**
      * {@inheritdoc}
      */
     protected function createContext()
     {
-        return $this->createMongodbContext();
+        return $this->buildMongodbContext();
     }
 }

Tests/Spec/MongodbSendAndReceivePriorityMessagesFromQueueTest.php

@@ -4,6 +4,7 @@
 
 use Enqueue\Mongodb\MongodbContext;
 use Enqueue\Mongodb\MongodbMessage;
+use Enqueue\Test\MongodbExtensionTrait;
 use Interop\Queue\PsrContext;
 use Interop\Queue\Spec\SendAndReceivePriorityMessagesFromQueueSpec;
 
@@ -13,7 +14,7 @@
  */
 class MongodbSendAndReceivePriorityMessagesFromQueueTest extends SendAndReceivePriorityMessagesFromQueueSpec
 {
-    use CreateMongodbContextTrait;
+    use MongodbExtensionTrait;
 
     private $publishedAt;
 
@@ -29,7 +30,7 @@ public function setUp()
      */
     protected function createContext()
     {
-        return $this->createMongodbContext();
+        return $this->buildMongodbContext();
     }
 
     /**

Tests/Spec/MongodbSendAndReceiveTimeToLiveMessagesFromQueueTest.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Mongodb\Tests\Spec;
 
+use Enqueue\Test\MongodbExtensionTrait;
 use Interop\Queue\Spec\SendAndReceiveTimeToLiveMessagesFromQueueSpec;
 
 /**
@@ -10,13 +11,13 @@
  */
 class MongodbSendAndReceiveTimeToLiveMessagesFromQueueTest extends SendAndReceiveTimeToLiveMessagesFromQueueSpec
 {
-    use CreateMongodbContextTrait;
+    use MongodbExtensionTrait;
 
     /**
      * {@inheritdoc}
      */
     protected function createContext()
     {
-        return $this->createMongodbContext();
+        return $this->buildMongodbContext();
     }
 }

Tests/Spec/MongodbSendToAndReceiveFromQueueTest.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Mongodb\Tests\Spec;
 
+use Enqueue\Test\MongodbExtensionTrait;
 use Interop\Queue\Spec\SendToAndReceiveFromQueueSpec;
 
 /**
@@ -10,13 +11,13 @@
  */
 class MongodbSendToAndReceiveFromQueueTest extends SendToAndReceiveFromQueueSpec
 {
-    use CreateMongodbContextTrait;
+    use MongodbExtensionTrait;
 
     /**
      * {@inheritdoc}
      */
     protected function createContext()
     {
-        return $this->createMongodbContext();
+        return $this->buildMongodbContext();
     }
 }

Tests/Spec/MongodbSendToAndReceiveFromTopicTest.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Mongodb\Tests\Spec;
 
+use Enqueue\Test\MongodbExtensionTrait;
 use Interop\Queue\Spec\SendToAndReceiveFromTopicSpec;
 
 /**
@@ -10,13 +11,13 @@
  */
 class MongodbSendToAndReceiveFromTopicTest extends SendToAndReceiveFromTopicSpec
 {
-    use CreateMongodbContextTrait;
+    use MongodbExtensionTrait;
 
     /**
      * {@inheritdoc}
      */
     protected function createContext()
     {
-        return $this->createMongodbContext();
+        return $this->buildMongodbContext();
     }
 }