Mongodb transport #430

turboboy88 · 2018-04-26T09:22:04Z

Mongodb transport (ref #284)

makasim · 2018-04-26T09:23:42Z

docker-compose.yml

@@ -44,6 +45,7 @@ services:
      - RDKAFKA_PORT=9092
      - PUBSUB_EMULATOR_HOST=http://google-pubsub:8085
      - GCLOUD_PROJECT=mqdev
+      - MONGO_CONNECTION_STRING=mongodb://127.0.0.1/


rename to MONGO_DSN

makasim · 2018-04-26T09:25:16Z

pkg/mongodb/MongodbConnectionFactory.php

+     */
+    private $driverOptions;
+
+    public function __construct($uri = 'mongodb://127.0.0.1/', array $config = [], array $uriOptions = [], array $driverOptions = [])


constructor must accept one argument, an array.

t should have uri, uriOptions, driverOptions keys and other options

makasim · 2018-04-26T09:27:55Z

pkg/mongodb/MongodbConsumer.php

+            }
+
+            return $convertedMessage;
+        } catch (\Exception $e) {


no need for catch

makasim · 2018-04-26T09:28:52Z

pkg/mongodb/MongodbConsumer.php

+        try {
+            $now = time();
+            $collection = $this->context->getCollection();
+            $message = $collection->findOne(['$or' => [['delayed_until' => ['$exists' => false]], ['delayed_until' => ['$lte' => $now]]]], ['sort' => ['priority' => -1]]);


the line is too long, could you split it into several lines??

https://docs.mongodb.com/manual/reference/method/db.collection.findOneAndDelete/

makasim · 2018-04-26T09:32:06Z

pkg/mongodb/MongodbConsumer.php

+            if (!$message) {
+                return null;
+            }
+            $mongodbMessage = $message->getArrayCopy();


https://github.com/makasim/yadm/blob/master/src/Storage.php#L207

makasim · 2018-04-26T09:35:26Z

pkg/mongodb/MongodbConsumer.php

+        }
+
+        if ($mongodbMessage['properties']) {
+            $message->setProperties(JSON::decode($mongodbMessage['properties']));


IMO there is no need to encode properties nor headers to json, we could store an array in mongodb.

makasim · 2018-04-26T09:36:22Z

pkg/mongodb/MongodbConsumer.php

+     */
+    protected function convertMessage(array $mongodbMessage)
+    {
+        $message = $this->context->createMessage();


the createMessage method accept body, headers, and properties as arguments. pass them

makasim · 2018-04-26T09:37:28Z

pkg/mongodb/MongodbContext.php

+    /**
+     * @return array
+     */
+    public function getConfig()


let's not make it public for now.

makasim · 2018-04-26T09:39:13Z

pkg/mongodb/MongodbProducer.php

+            $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);


The transport has failed to send the message due to some internal error.

makasim · 2018-04-26T09:40:21Z

pkg/mongodb/composer.json

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


I dont think we needed, please remove

makasim · 2018-04-26T15:02:59Z

composer.json

@@ -41,7 +42,8 @@
    "symfony/console": "^2.8|^3|^4",
    "friendsofphp/php-cs-fixer": "^2",
    "empi89/php-amqp-stubs": "*@dev",
-    "php-http/client-common": "^1.7@dev"
+    "php-http/client-common": "^1.7@dev",
+    "mongodb/mongodb": "^1.3"


could be remove as you required it in enqueue/mongodb

makasim · 2018-04-26T15:03:49Z

docker-compose.yml

@@ -44,6 +45,7 @@ services:
      - RDKAFKA_PORT=9092
      - PUBSUB_EMULATOR_HOST=http://google-pubsub:8085
      - GCLOUD_PROJECT=mqdev
+      - MONGO_DSN=mongodb://127.0.0.1/


should be mongodb://mongo/

makasim · 2018-04-26T15:04:17Z

docker-compose.yml

+  mongo:
+      image: mongo
+      ports:
+        - "27017:27017"


add an empty line after the service definition.

makasim · 2018-04-26T15:07:46Z

pkg/mongodb/MongodbConsumer.php

+    {
+        $now = time();
+        $collection = $this->context->getCollection();
+        $message = $collection->findOneAndDelete(['$or' => [['delayed_until' => ['$exists' => false]], ['delayed_until' => ['$lte' => $now]]]],


please split into several lines.

makasim · 2018-04-26T15:08:11Z

pkg/mongodb/MongodbConsumer.php

+    protected function convertMessage(array $mongodbMessage)
+    {
+        $message = $this->context->createMessage($mongodbMessage['body'], $mongodbMessage['properties'], $mongodbMessage['headers']);
+        $message->setId($mongodbMessage['_id']->__toString());


(string) $mongodbMessage['_id']

makasim · 2018-04-26T15:09:08Z

pkg/mongodb/MongodbContext.php

+
+    public function getCollection()
+    {
+        return $this->client->selectDatabase($this->config['dbname'])->selectCollection($this->config['collection_name']);


<?php return $this->client ->selectDatabase($this->config['dbname']) ->selectCollection($this->config['collection_name']) ;

makasim · 2018-04-26T15:09:58Z

pkg/mongodb/MongodbProducer.php

+        $mongoMessage = [
+            'published_at' => $publishedAt,
+            'body' => $body,
+            'headers' => JSON::encode($message->getHeaders()),


no need to json encode

makasim · 2018-04-26T15:10:59Z

pkg/mongodb/composer.json

+  "homepage": "https://enqueue.forma-pro.com/",
+  "license": "MIT",
+  "require": {
+    "php": ">=5.6",


makasim · 2018-04-26T15:20:38Z

pkg/mongodb/MongodbConnectionFactory.php

+     */
+    private $config;
+
+    public function __construct($config = 'mongodb:')


could you please add a docblcok with the information on possible configuration options

makasim · 2018-04-26T15:45:53Z

pkg/mongodb/Tests/Spec/CreateMongodbContextTrait.php

+            $this->markTestSkipped('The MONGO_DSN env is not available. Skip tests');
+        }
+        $params = ['uri' => $env];
+        $factory = new MongodbConnectionFactory($params);


<?php $factory = new MongodbConnectionFactory(['uri' => $env]);

makasim · 2018-04-27T10:25:36Z

pkg/mongodb/MongodbConnectionFactory.php

+            ));
+        }
+
+        if ($parsedUrl['scheme'].':' === $dsn) {


hardcode string mongodb:

makasim · 2018-04-27T10:25:44Z

pkg/mongodb/MongodbConnectionFactory.php

+
+        if ($parsedUrl['scheme'].':' === $dsn) {
+            return [
+                'uri' => $parsedUrl['scheme'].'://127.0.0.1/',


makasim · 2018-04-27T10:26:32Z

pkg/mongodb/MongodbConnectionFactory.php

+        $config['uri'] = $parsedUrl['scheme'].'://'.$parsedUrl['host'];
+
+        if (isset($parsedUrl['path']) && '/' !== $parsedUrl['path']) {
+            $pathArr = explode('/', $parsedUrl['path']);


rename the var to pathParts

makasim · 2018-04-27T10:27:33Z

pkg/mongodb/MongodbConnectionFactory.php

+            ];
+        }
+
+        $config['uri'] = $parsedUrl['scheme'].'://'.$parsedUrl['host'];


this is not correct, the DSN might contain user, pasword, port, other options in it,

I think it would be good to store the orignal url as is is

makasim · 2018-04-27T11:17:25Z

pkg/mongodb/MongodbConnectionFactory.php

+        }
+
+        if (isset($parsedUrl['query'])) {
+            parse_str($parsedUrl['query'], $queryParts);


use of undefined var, define $queryParts = null just before this line

makasim · 2018-04-27T11:17:50Z

pkg/mongodb/MongodbConnectionFactory.php

+            parse_str($parsedUrl['query'], $queryParts);
+
+            //get enqueue attributes values
+            if (isset($queryParts['polling_interval']) && !empty($queryParts['polling_interval'])) {


there is no need for isset()

makasim · 2018-04-27T15:24:54Z

pkg/mongodb/Symfony/MongodbTransportFactory.php

+            $factory = new Definition(MongodbConnectionFactory::class);
+            $factory->setArguments([$config]);
+        } else {
+            throw new \LogicException('Set "dsn" options when you want ot use Mongodb.');


instead of this add required to ->scalarNode('dsn')

makasim · 2018-04-27T15:25:44Z

pkg/mongodb/Symfony/MongodbTransportFactory.php

+    {
+        if (false == empty($config['dsn'])) {
+            $factory = new Definition(MongodbConnectionFactory::class);
+            $factory->setArguments([$config]);


as far as I remember Definition takes an array of arguments as a second constructor argument.

makasim · 2018-04-27T15:52:50Z

pkg/mongodb/MongodbConnectionFactory.php

+            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/',


I think we should add dbname and a collection here as well

makasim · 2018-04-27T15:56:36Z

pkg/mongodb/MongodbProducer.php

+            'headers' => $message->getHeaders(),
+            'properties' => $message->getProperties(),
+            'priority' => $message->getPriority(),
+            'queue' => $destination->getQueueName(),


The use of getQueueName method is a bit misleading as we can send a message to the topic too.

Could you please add a neutral getName method to the destination and use it everywhere?

Added getName() alias for getQueueName() method. getQueueName() method should be implemented from PsrQueue interface.

makasim · 2018-04-27T16:00:34Z

pkg/mongodb/MongodbConsumer.php

+        $timeout /= 1000;
+        $startAt = microtime(true);
+
+        while (true) {


let's try capped collection plus tailable cursor approach.

https://gist.github.com/ASKozienko/cf45b09c68774ed83dea67b6d02d5846

makasim · 2018-04-28T10:23:07Z

pkg/mongodb/MongodbProducer.php

+        InvalidDestinationException::assertDestinationInstanceOf($destination, MongodbDestination::class);
+        InvalidMessageException::assertMessageInstanceOf($message, MongodbMessage::class);
+
+        if (null !== $this->priority && 0 === $message->getPriority()) {


should be if (null !== $this->priority && null === $message->getPriority()) {

and set priority in the message to null by default (just unset).

Mongodb transport

makasim requested changes Apr 26, 2018

View reviewed changes

turboboy88 changed the title ~~initial commit~~ Mongodb transport Apr 26, 2018

makasim reviewed Apr 26, 2018

View reviewed changes

docker-compose.yml Outdated

mongo:

image: mongo

ports:

- "27017:27017"

Copy link

Member

makasim Apr 26, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

add an empty line after the service definition.

makasim reviewed Apr 26, 2018

View reviewed changes

makasim reviewed Apr 27, 2018

View reviewed changes

pkg/mongodb/MongodbConnectionFactory.php Outdated

));

}

if ($parsedUrl['scheme'].':' === $dsn) {

Copy link

Member

makasim Apr 27, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hardcode string mongodb:

makasim reviewed Apr 27, 2018

View reviewed changes

makasim added the enhancement label Apr 27, 2018

makasim reviewed Apr 27, 2018

View reviewed changes

turboboy88 force-pushed the mongodb_transport branch from 9f620b5 to 85e4b9d Compare April 27, 2018 11:31

Mongo db transport

07ed3ba

turboboy88 force-pushed the mongodb_transport branch from 85e4b9d to 07ed3ba Compare April 27, 2018 11:35

turboboy88 added 2 commits April 27, 2018 16:03

client

4346e9f

smfony transport

83d4b85

makasim reviewed Apr 27, 2018

View reviewed changes

fix

2e8f931

makasim reviewed Apr 27, 2018

View reviewed changes

develop

7e9f62d

makasim reviewed Apr 28, 2018

View reviewed changes

develop

9cbbce8

makasim mentioned this pull request May 3, 2018

A list of nice to add transports. Feel free to make a proposition. #284

Open

21 tasks

makasim added 19 commits May 3, 2018 12:13

Update .travis.yml

947e00f

Merge branch 'master' into mongodb_transport

c17e79f

update enqueue/dev docker image

a7ea2f1

debug travis.

e924c23

use $HOME

e8de54a

add group

f76e48d

add license, readme, phpunit cfg

aa3178f

mongolib ^1.2, ext-mongodb ^1.3

f460e6c

require php7.0

41f2b47

[doc][skip ci] add docs.

a9fd835

[symfony] add mongodb to default transport and dsn_to_function

368b608

run mongodb tests.

0f8d1dc

fix tests.

41254f9

fix tests.

f011a8b

fix tests

b128a65

fix test.

e15e92f

Add mongodb transport.

3b8c997

mongodb transport, fixes

1b23270

revert sf version check

c407d24

makasim merged commit ccaaeb1 into php-enqueue:master May 3, 2018

ASKozienko pushed a commit that referenced this pull request Nov 2, 2018

Merge pull request #430 from turboboy88/mongodb_transport

56e5c97

Mongodb transport

Mongodb transport #430

Mongodb transport #430

Conversation

turboboy88 commented Apr 26, 2018 • edited by makasim Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

makasim Apr 27, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

turboboy88 commented Apr 26, 2018 •

edited by makasim

Loading

makasim Apr 27, 2018 •

edited

Loading