Skip to content

Commit 07ed3ba

Browse files
turboboy88turboboy88
committed
Mongo db transport
1 parent 9adbf01 commit 07ed3ba

33 files changed

+2036
-1
lines changed

‎bin/test

+1
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@ waitForService redis 6379 50
3636
waitForService beanstalkd 11300 50
3737
waitForService gearmand 4730 50
3838
waitForService kafka 9092 50
39+
waitForService mongo 27017 50
3940

4041
php pkg/job-queue/Tests/Functional/app/console doctrine:database:create --if-not-exists
4142
php pkg/job-queue/Tests/Functional/app/console doctrine:schema:update --force

‎composer.json

+5
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
"enqueue/fs": "*@dev",
1717
"enqueue/null": "*@dev",
1818
"enqueue/dbal": "*@dev",
19+
"enqueue/mongodb": "*@dev",
1920
"enqueue/sqs": "*@dev",
2021
"enqueue/pheanstalk": "*@dev",
2122
"enqueue/gearman": "*@dev",
@@ -143,6 +144,10 @@
143144
{
144145
"type": "path",
145146
"url": "pkg/async-event-dispatcher"
147+
},
148+
{
149+
"type": "path",
150+
"url": "pkg/mongodb"
146151
}
147152
]
148153
}

‎docker-compose.yml

+8-1
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ services:
1313
- zookeeper
1414
- google-pubsub
1515
- rabbitmqssl
16+
- mongo
1617
volumes:
1718
- './:/mqdev'
1819
environment:
@@ -24,7 +25,7 @@ services:
2425
- RABBITMQ_PASSWORD=guest
2526
- RABBITMQ_VHOST=mqdev
2627
- RABBITMQ_AMQP__PORT=5672
27-
- RABBITMQ_STOMP_PORT=61613
28+
- RABBITMQ_STOMP_PORT=61613
2829
- DOCTRINE_DRIVER=pdo_mysql
2930
- DOCTRINE_HOST=mysql
3031
- DOCTRINE_PORT=3306
@@ -44,6 +45,7 @@ services:
4445
- RDKAFKA_PORT=9092
4546
- PUBSUB_EMULATOR_HOST=http://google-pubsub:8085
4647
- GCLOUD_PROJECT=mqdev
48+
- MONGO_DSN=mongodb://mongo
4749

4850
rabbitmq:
4951
image: 'enqueue/rabbitmq:latest'
@@ -102,6 +104,11 @@ services:
102104
image: 'google/cloud-sdk:latest'
103105
entrypoint: 'gcloud beta emulators pubsub start --host-port=0.0.0.0:8085'
104106

107+
mongo:
108+
image: mongo
109+
ports:
110+
- "27017:27017"
111+
105112
volumes:
106113
mysql-data:
107114
driver: local

‎pkg/mongodb/.gitignore

+6
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
*~
2+
/composer.lock
3+
/composer.phar
4+
/phpunit.xml
5+
/vendor/
6+
/.idea/

‎pkg/mongodb/MongodbConnectionFactory.php

+101
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
<?php
2+
3+
namespace Enqueue\Mongodb;
4+
5+
use Interop\Queue\PsrConnectionFactory;
6+
use MongoDB\Client;
7+
8+
class MongodbConnectionFactory implements PsrConnectionFactory
9+
{
10+
/**
11+
* @var array
12+
*/
13+
private $config;
14+
15+
/**
16+
* 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.
17+
*
18+
* $config = [
19+
* 'uri' => 'mongodb://127.0.0.1/' - Mongodb connection string. see http://docs.mongodb.org/manual/reference/connection-string/
20+
* 'dbname' => 'enqueue', - database name.
21+
* 'collection_name' => 'enqueue' - collection name
22+
* 'polling_interval' => '1000', - How often query for new messages (milliseconds)
23+
* ]
24+
*
25+
* or
26+
*
27+
* mongodb://127.0.0.1:27017/dbname/collection_name?polling_interval=1000
28+
*
29+
* @param array|string|null $config
30+
*/
31+
public function __construct($config = 'mongodb:')
32+
{
33+
if (empty($config)) {
34+
$config = $this->parseDsn('mongodb:');
35+
} elseif (is_string($config)) {
36+
$config = $this->parseDsn($config);
37+
} elseif (is_array($config)) {
38+
} else {
39+
throw new \LogicException('The config must be either an array of options, a DSN string or null');
40+
}
41+
$config = array_replace([
42+
'uri' => 'mongodb://127.0.0.1/',
43+
], $config);
44+
45+
$this->config = $config;
46+
}
47+
48+
public function createContext()
49+
{
50+
$client = new Client($this->config['uri']);
51+
52+
return new MongodbContext($client, $this->config);
53+
}
54+
55+
public static function parseDsn($dsn)
56+
{
57+
$parsedUrl = parse_url($dsn);
58+
if (false === $parsedUrl) {
59+
throw new \LogicException(sprintf('Failed to parse DSN "%s"', $dsn));
60+
}
61+
if (empty($parsedUrl['scheme'])) {
62+
throw new \LogicException('Schema is empty');
63+
}
64+
$supported = [
65+
'mongodb' => true,
66+
];
67+
if (false == isset($parsedUrl['scheme'])) {
68+
throw new \LogicException(sprintf(
69+
'The given DSN schema "%s" is not supported. There are supported schemes: "%s".',
70+
$parsedUrl['scheme'],
71+
implode('", "', array_keys($supported))
72+
));
73+
}
74+
if ('mongodb:' === $dsn) {
75+
return [
76+
'uri' => 'mongodb://127.0.0.1/',
77+
];
78+
}
79+
$config['uri'] = $dsn;
80+
if (isset($parsedUrl['path']) && '/' !== $parsedUrl['path']) {
81+
$pathParts = explode('/', $parsedUrl['path']);
82+
//DB name
83+
if ($pathParts[1]) {
84+
$config['dbname'] = $pathParts[1];
85+
}
86+
}
87+
if (isset($parsedUrl['query'])) {
88+
$queryParts = null;
89+
parse_str($parsedUrl['query'], $queryParts);
90+
//get enqueue attributes values
91+
if (!empty($queryParts['polling_interval'])) {
92+
$config['polling_interval'] = $queryParts['polling_interval'];
93+
}
94+
if (!empty($queryParts['enqueue_collection'])) {
95+
$config['collection_name'] = $queryParts['enqueue_collection'];
96+
}
97+
}
98+
99+
return $config;
100+
}
101+
}

‎pkg/mongodb/MongodbConsumer.php

+174
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,174 @@
1+
<?php
2+
3+
namespace Enqueue\Mongodb;
4+
5+
use Interop\Queue\InvalidMessageException;
6+
use Interop\Queue\PsrConsumer;
7+
use Interop\Queue\PsrMessage;
8+
9+
class MongodbConsumer implements PsrConsumer
10+
{
11+
/**
12+
* @var MongodbContext
13+
*/
14+
private $context;
15+
16+
/**
17+
* @var MongodbDestination
18+
*/
19+
private $queue;
20+
21+
/**
22+
* @var int microseconds
23+
*/
24+
private $pollingInterval = 1000000;
25+
26+
/**
27+
* @param MongodbContext $context
28+
* @param MongodbDestination $queue
29+
*/
30+
public function __construct(MongodbContext $context, MongodbDestination $queue)
31+
{
32+
$this->context = $context;
33+
$this->queue = $queue;
34+
}
35+
36+
/**
37+
* Set polling interval in milliseconds.
38+
*
39+
* @param int $msec
40+
*/
41+
public function setPollingInterval($msec)
42+
{
43+
$this->pollingInterval = $msec * 1000;
44+
}
45+
46+
/**
47+
* Get polling interval in milliseconds.
48+
*
49+
* @return int
50+
*/
51+
public function getPollingInterval()
52+
{
53+
return (int) $this->pollingInterval / 1000;
54+
}
55+
56+
/**
57+
* {@inheritdoc}
58+
*
59+
* @return MongodbDestination
60+
*/
61+
public function getQueue()
62+
{
63+
return $this->queue;
64+
}
65+
66+
/**
67+
* {@inheritdoc}
68+
*
69+
* @return MongodbMessage|null
70+
*/
71+
public function receive($timeout = 0)
72+
{
73+
$timeout /= 1000;
74+
$startAt = microtime(true);
75+
76+
while (true) {
77+
$message = $this->receiveMessage();
78+
79+
if ($message) {
80+
return $message;
81+
}
82+
83+
if ($timeout && (microtime(true) - $startAt) >= $timeout) {
84+
return;
85+
}
86+
87+
usleep($this->pollingInterval);
88+
89+
if ($timeout && (microtime(true) - $startAt) >= $timeout) {
90+
return;
91+
}
92+
}
93+
}
94+
95+
/**
96+
* {@inheritdoc}
97+
*
98+
* @return MongodbMessage|null
99+
*/
100+
public function receiveNoWait()
101+
{
102+
return $this->receiveMessage();
103+
}
104+
105+
/**
106+
* {@inheritdoc}
107+
*
108+
* @param MongodbMessage $message
109+
*/
110+
public function acknowledge(PsrMessage $message)
111+
{
112+
// does nothing
113+
}
114+
115+
/**
116+
* {@inheritdoc}
117+
*
118+
* @param MongodbMessage $message
119+
*/
120+
public function reject(PsrMessage $message, $requeue = false)
121+
{
122+
InvalidMessageException::assertMessageInstanceOf($message, MongodbMessage::class);
123+
124+
if ($requeue) {
125+
$this->context->createProducer()->send($this->queue, $message);
126+
127+
return;
128+
}
129+
}
130+
131+
/**
132+
* @return MongodbMessage|null
133+
*/
134+
protected function receiveMessage()
135+
{
136+
$now = time();
137+
$collection = $this->context->getCollection();
138+
$message = $collection->findOneAndDelete(
139+
[
140+
'$or' => [
141+
['delayed_until' => ['$exists' => false]],
142+
['delayed_until' => ['$lte' => $now]],
143+
],
144+
],
145+
[
146+
'sort' => ['priority' => -1, 'published_at' => 1],
147+
'typeMap' => ['root' => 'array', 'document' => 'array'],
148+
]
149+
);
150+
151+
if (!$message) {
152+
return null;
153+
}
154+
if (empty($message['time_to_live']) || $message['time_to_live'] > time()) {
155+
return $this->convertMessage($message);
156+
}
157+
}
158+
159+
/**
160+
* @param array $dbalMessage
161+
*
162+
* @return MongodbMessage
163+
*/
164+
protected function convertMessage(array $mongodbMessage)
165+
{
166+
$message = $this->context->createMessage($mongodbMessage['body'], $mongodbMessage['properties'], $mongodbMessage['headers']);
167+
$message->setId((string) $mongodbMessage['_id']);
168+
$message->setPriority((int) $mongodbMessage['priority']);
169+
$message->setRedelivered((bool) $mongodbMessage['redelivered']);
170+
$message->setPublishedAt((int) $mongodbMessage['published_at']);
171+
172+
return $message;
173+
}
174+
}

0 commit comments

Comments
 (0)