develop

Author: makasim

docker-compose.yml

@@ -45,7 +45,7 @@ services:
       - RDKAFKA_PORT=9092
       - PUBSUB_EMULATOR_HOST=http://google-pubsub:8085
       - GCLOUD_PROJECT=mqdev
-      - MONGO_CONNECTION_STRING=mongodb://127.0.0.1/
+      - MONGO_DSN=mongodb://127.0.0.1/
 
   rabbitmq:
     image: 'enqueue/rabbitmq:latest'

pkg/mongodb/MongodbConnectionFactory.php

@@ -7,37 +7,23 @@
 
 class MongodbConnectionFactory implements PsrConnectionFactory
 {
-    /**
-     * @var string
-     */
-    private $uri;
-
     /**
      * @var array
      */
     private $config;
 
-    /**
-     * @var array
-     */
-    private $uriOptions;
-
-    /**
-     * @var array
-     */
-    private $driverOptions;
-
-    public function __construct($uri = 'mongodb://127.0.0.1/', array $config = [], array $uriOptions = [], array $driverOptions = [])
+    public function __construct(array $params)
     {
-        $this->uri = $uri;
-        $this->config = $config;
-        $this->uriOptions = $uriOptions;
-        $this->driverOptions = $driverOptions;
+        $this->config = array_replace([
+            'uri' => 'mongodb://127.0.0.1/',
+            'uriOptions' => [],
+            'driverOptions' => [],
+        ], $params);
     }
 
     public function createContext()
     {
-        $client = new Client($this->uri, $this->uriOptions, $this->driverOptions);
+        $client = new Client($this->config['uri'], $this->config['uriOptions'], $this->config['driverOptions']);
 
         return new MongodbContext($client, $this->config);
     }

pkg/mongodb/MongodbConsumer.php

@@ -2,7 +2,6 @@
 
 namespace Enqueue\Mongodb;
 
-use Enqueue\Util\JSON;
 use Interop\Queue\InvalidMessageException;
 use Interop\Queue\PsrConsumer;
 use Interop\Queue\PsrMessage;
@@ -134,24 +133,18 @@ public function reject(PsrMessage $message, $requeue = false)
      */
     protected function receiveMessage()
     {
-        try {
-            $now = time();
-            $collection = $this->context->getCollection();
-            $message = $collection->findOne(['$or' => [['delayed_until' => ['$exists' => false]], ['delayed_until' => ['$lte' => $now]]]], ['sort' => ['priority' => -1]]);
-            if (!$message) {
-                return null;
-            }
-            $mongodbMessage = $message->getArrayCopy();
-            $convertedMessage = $this->convertMessage($mongodbMessage);
-            $affected = $collection->deleteOne(['_id' => $mongodbMessage['_id']]);
-            if (1 !== $affected->getDeletedCount()) {
-                throw new \LogicException(sprintf('Expected record was removed but it is not. id: "%s"', $mongodbMessage['_id']->__toString()));
-            }
+        $now = time();
+        $collection = $this->context->getCollection();
+        $message = $collection->findOneAndDelete(['$or' => [['delayed_until' => ['$exists' => false]], ['delayed_until' => ['$lte' => $now]]]],
+            ['sort' => ['priority' => -1], 'typeMap' => ['root' => 'array', 'document' => 'array', 'array' => 'array']]);
 
-            return $convertedMessage;
-        } catch (\Exception $e) {
-            throw $e;
+        if (!$message) {
+            return null;
         }
+
+        $convertedMessage = $this->convertMessage($message);
+
+        return $convertedMessage;
     }
 
     /**
@@ -161,21 +154,12 @@ protected function receiveMessage()
      */
     protected function convertMessage(array $mongodbMessage)
     {
-        $message = $this->context->createMessage();
+        $message = $this->context->createMessage($mongodbMessage['body'], $mongodbMessage['properties'], $mongodbMessage['headers']);
         $message->setId($mongodbMessage['_id']->__toString());
-        $message->setBody($mongodbMessage['body']);
         $message->setPriority((int) $mongodbMessage['priority']);
         $message->setRedelivered((bool) $mongodbMessage['redelivered']);
         $message->setPublishedAt((int) $mongodbMessage['published_at']);
 
-        if ($mongodbMessage['headers']) {
-            $message->setHeaders(JSON::decode($mongodbMessage['headers']));
-        }
-
-        if ($mongodbMessage['properties']) {
-            $message->setProperties(JSON::decode($mongodbMessage['properties']));
-        }
-
         return $message;
     }
 }

pkg/mongodb/MongodbProducer.php

@@ -123,7 +123,7 @@ public function send(PsrDestination $destination, PsrMessage $message)
             $collection = $this->context->getCollection();
             $collection->insertOne($mongoMessage);
         } catch (\Exception $e) {
-            throw new Exception('The transport fails to send the message due to some internal error.', null, $e);
+            throw new Exception('The transport has failed to send the message due to some internal error.', null, $e);
         }
     }
 

pkg/mongodb/Tests/Spec/CreateMongodbContextTrait.php

@@ -8,8 +8,8 @@ trait CreateMongodbContextTrait
 {
     protected function createMongodbContext()
     {
-        if (false == $env = getenv('MONGO_CONNECTION_STRING')) {
-            $this->markTestSkipped('The MONGO_CONNECTION_STRING env is not available. Skip tests');
+        if (false == $env = getenv('MONGO_DSN')) {
+            $this->markTestSkipped('The MONGO_DSN env is not available. Skip tests');
         }
 
         $factory = new MongodbConnectionFactory($env);

pkg/mongodb/composer.json

@@ -12,7 +12,6 @@
   "require": {
     "php": ">=5.6",
     "queue-interop/queue-interop": "^0.6@dev|^1.0.0-alpha1",
-    "php-http/client-common": "^1.7@dev",
     "mongodb/mongodb": "^1.3"
   },
   "require-dev": {