Mongodb transport by turboboy88 · Pull Request #430 · php-enqueue/enqueue-dev · GitHub

Mongodb transport (ref #284)

rename to MONGO_DSN

constructor must accept one argument, an array.

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

no need for catch

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

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

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

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

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

let's not make it public for now.

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

I dont think we needed, please remove

could be remove as you required it in enqueue/mongodb

should be mongodb://mongo/

add an empty line after the service definition.

please split into several lines.

(string) $mongodbMessage['_id']

<?php

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

no need to json encode

^7.1

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

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

hardcode string mongodb:

same here

rename the var to pathParts

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

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

there is no need for isset()

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

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

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

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.

let's try capped collection plus tailable cursor approach.

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

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

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

                     - RDKAFKA_PORT=9092
                     - PUBSUB_EMULATOR_HOST=http://google-pubsub:8085
                     - GCLOUD_PROJECT=mqdev
+                    - MONGO_CONNECTION_STRING=mongodb://127.0.0.1/

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

+                          $factory = new Definition(MongodbConnectionFactory::class);
+                          $factory->setArguments([$config]);
+                      } else {
+                          throw new \LogicException('Set "dsn" options when you want ot use 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/',

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

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

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

Mongodb transport by turboboy88 · Pull Request #430 · php-enqueue/enqueue-dev · GitHub

Uh oh!

Mongodb transport - #430

Mongodb transport#430makasim merged 25 commits intophp-enqueue:masterphp-enqueue/enqueue-dev:masterfrom turboboy88:mongodb_transportturboboy88/enqueue-dev:mongodb_transportCopy head branch name to clipboard

Conversation

turboboy88 commented Apr 26, 2018 • edited by makasim Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

makasim Apr 27, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Mongodb transport#430
makasim merged 25 commits into
php-enqueue:masterfrom
turboboy88:mongodb_transport

turboboy88 commented Apr 26, 2018 •
edited by makasim
Loading

makasim Apr 27, 2018 •
edited
Loading