develop

Author: makasim

bin/test

@@ -36,6 +36,7 @@ waitForService redis 6379 50
 waitForService beanstalkd 11300 50
 waitForService gearmand 4730 50
 waitForService kafka 9092 50
+waitForService mongo 27017 50
 
 php pkg/job-queue/Tests/Functional/app/console doctrine:database:create --if-not-exists
 php pkg/job-queue/Tests/Functional/app/console doctrine:schema:update --force

composer.json

@@ -145,7 +145,7 @@
     {
       "type": "path",
       "url": "pkg/async-event-dispatcher"
-    }
+    },
     {
       "type": "path",
       "url": "pkg/mongodb"

docker-compose.yml

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

pkg/mongodb/MongodbConnectionFactory.php

@@ -12,13 +12,23 @@ class MongodbConnectionFactory implements PsrConnectionFactory
      */
     private $config;
 
-    public function __construct(array $params)
+    public function __construct($config = 'mongodb:')
     {
-        $this->config = array_replace([
+        if (empty($config)) {
+            $config = $this->parseDsn('mongodb:');
+        } elseif (is_string($config)) {
+            $config = $this->parseDsn($config);
+        } elseif (is_array($config)) {
+        } else {
+            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/',
             'uriOptions' => [],
             'driverOptions' => [],
-        ], $params);
+        ], $config);
+
+        $this->config = $config;
     }
 
     public function createContext()
@@ -27,4 +37,32 @@ public function createContext()
 
         return new MongodbContext($client, $this->config);
     }
+
+    private function parseDsn($dsn)
+    {
+        if (false === parse_url($dsn)) {
+            throw new \LogicException(sprintf('Failed to parse DSN "%s"', $dsn));
+        }
+
+        $schema = parse_url($dsn, PHP_URL_SCHEME);
+        if (empty($schema)) {
+            throw new \LogicException('Schema is empty');
+        }
+
+        $supported = [
+            'mongodb' => true,
+        ];
+
+        if (false == isset($supported[$schema])) {
+            throw new \LogicException(sprintf(
+                'The given DSN schema "%s" is not supported. There are supported schemes: "%s".',
+                $schema,
+                implode('", "', array_keys($supported))
+            ));
+        }
+
+        return [
+            'uri' => $schema.':' === $dsn ? $schema.'://127.0.0.1/' : $dsn,
+        ];
+    }
 }

pkg/mongodb/MongodbConsumer.php

@@ -136,15 +136,14 @@ protected function receiveMessage()
         $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']]);
+            ['sort' => ['priority' => -1, 'published_at' => 1], 'typeMap' => ['root' => 'array', 'document' => 'array']]);
 
         if (!$message) {
             return null;
         }
-
-        $convertedMessage = $this->convertMessage($message);
-
-        return $convertedMessage;
+        if (empty($message['time_to_live']) || $message['time_to_live'] > time()) {
+            return $this->convertMessage($message);
+        }
     }
 
     /**

pkg/mongodb/MongodbProducer.php

@@ -2,7 +2,6 @@
 
 namespace Enqueue\Mongodb;
 
-use Enqueue\Util\JSON;
 use Interop\Queue\Exception;
 use Interop\Queue\InvalidDestinationException;
 use Interop\Queue\InvalidMessageException;
@@ -81,10 +80,11 @@ public function send(PsrDestination $destination, PsrMessage $message)
         $mongoMessage = [
             'published_at' => $publishedAt,
             'body' => $body,
-            'headers' => JSON::encode($message->getHeaders()),
-            'properties' => JSON::encode($message->getProperties()),
+            'headers' => $message->getHeaders(),
+            'properties' => $message->getProperties(),
             'priority' => $message->getPriority(),
             'queue' => $destination->getQueueName(),
+            'redelivered' => $message->isRedelivered(),
         ];
 
         $delay = $message->getDeliveryDelay();

pkg/mongodb/Tests/Functional/MongodbConsumerTest.php

@@ -75,7 +75,7 @@ public function testShouldOrderMessagesWithSamePriorityByPublishedAtDate()
         $olderTime = $time - 10000;
 
         $expectedPriority5Body = __CLASS__.'_priority5_'.$time;
-        $expectedPriority5BodyOlderTime = __CLASS__.'_priority5_'.$olderTime;
+        $expectedPriority5BodyOlderTime = __CLASS__.'_priority5Old_'.$olderTime;
 
         $producer = $context->createProducer();
 

pkg/mongodb/Tests/MongodbConnectionFactoryTest.php

@@ -18,21 +18,27 @@ public function testShouldImplementConnectionFactoryInterface()
 
     public function testCouldBeConstructedWithEmptyConfiguration()
     {
+        $params = [
+            'uri' => 'mongodb://127.0.0.1/',
+            'uriOptions' => [],
+            'driverOptions' => [],
+        ];
+
         $factory = new MongodbConnectionFactory();
-        $this->assertAttributeEquals('mongodb://127.0.0.1/', 'uri', $factory);
-        $this->assertAttributeEquals([], 'config', $factory);
-        $this->assertAttributeEquals([], 'uriOptions', $factory);
-        $this->assertAttributeEquals([], 'driverOptions', $factory);
+        $this->assertAttributeEquals($params, 'config', $factory);
     }
 
     public function testCouldBeConstructedWithCustomConfiguration()
     {
-        $factory = new MongodbConnectionFactory('mongodb://127.0.0.3/', ['testValue' => 123], ['testValue' => 123], ['testValue' => 123]);
+        $params = [
+            'uri' => 'mongodb://127.0.0.3/',
+            'uriOptions' => ['testValue' => 123],
+            'driverOptions' => ['testValue' => 123],
+        ];
+
+        $factory = new MongodbConnectionFactory($params);
 
-        $this->assertAttributeEquals('mongodb://127.0.0.3/', 'uri', $factory);
-        $this->assertAttributeEquals(['testValue' => 123], 'config', $factory);
-        $this->assertAttributeEquals(['testValue' => 123], 'uriOptions', $factory);
-        $this->assertAttributeEquals(['testValue' => 123], 'driverOptions', $factory);
+        $this->assertAttributeEquals($params, 'config', $factory);
     }
 
     public function testShouldCreateContext()

pkg/mongodb/Tests/Spec/CreateMongodbContextTrait.php

@@ -11,13 +11,11 @@ protected function createMongodbContext()
         if (false == $env = getenv('MONGO_DSN')) {
             $this->markTestSkipped('The MONGO_DSN env is not available. Skip tests');
         }
-
-        $factory = new MongodbConnectionFactory($env);
+        $params = ['uri' => $env];
+        $factory = new MongodbConnectionFactory($params);
 
         $context = $factory->createContext();
 
-        //$context->getClient()->dropDatabase('enqueue');
-
         return $context;
     }
 }