Skip to content

Commit 8c82c7c

Browse files
makasimmakasim
authored
Merge pull request #136 from fibula/amqplib
Add AmqpLib support
2 parents 89d8d95 + 8a212e6 commit 8c82c7c

34 files changed

+3331
-0
lines changed

Diff for: ‎composer.json

+6
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,8 @@
77
"enqueue/enqueue": "*@dev",
88
"enqueue/stomp": "*@dev",
99
"enqueue/amqp-ext": "*@dev",
10+
"enqueue/amqp-lib": "*@dev",
11+
"php-amqplib/php-amqplib": "^2.7@dev",
1012
"enqueue/redis": "*@dev",
1113
"enqueue/fs": "*@dev",
1214
"enqueue/null": "*@dev",
@@ -68,6 +70,10 @@
6870
"type": "path",
6971
"url": "pkg/amqp-ext"
7072
},
73+
{
74+
"type": "path",
75+
"url": "pkg/amqp-lib"
76+
},
7177
{
7278
"type": "path",
7379
"url": "pkg/redis"

Diff for: ‎pkg/amqp-lib/.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/

Diff for: ‎pkg/amqp-lib/.travis.yml

+21
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
sudo: false
2+
3+
git:
4+
depth: 1
5+
6+
language: php
7+
8+
php:
9+
- '5.6'
10+
- '7.0'
11+
12+
cache:
13+
directories:
14+
- $HOME/.composer/cache
15+
16+
install:
17+
- composer self-update
18+
- composer install --prefer-source --ignore-platform-reqs
19+
20+
script:
21+
- vendor/bin/phpunit --exclude-group=functional

Diff for: ‎pkg/amqp-lib/AmqpConnectionFactory.php

+229
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,229 @@
1+
<?php
2+
3+
namespace Enqueue\AmqpLib;
4+
5+
use Interop\Queue\PsrConnectionFactory;
6+
use PhpAmqpLib\Connection\AbstractConnection;
7+
use PhpAmqpLib\Connection\AMQPLazyConnection;
8+
use PhpAmqpLib\Connection\AMQPLazySocketConnection;
9+
use PhpAmqpLib\Connection\AMQPSocketConnection;
10+
use PhpAmqpLib\Connection\AMQPStreamConnection;
11+
12+
class AmqpConnectionFactory implements PsrConnectionFactory
13+
{
14+
/**
15+
* @var array
16+
*/
17+
private $config;
18+
19+
/**
20+
* @var AbstractConnection
21+
*/
22+
private $connection;
23+
24+
/**
25+
* The config could be an array, string DSN or null. In case of null it will attempt to connect to localhost with default credentials.
26+
*
27+
* [
28+
* 'host' => 'amqp.host The host to connect too. Note: Max 1024 characters.',
29+
* 'port' => 'amqp.port Port on the host.',
30+
* 'vhost' => 'amqp.vhost The virtual host on the host. Note: Max 128 characters.',
31+
* 'user' => 'amqp.user The user name to use. Note: Max 128 characters.',
32+
* 'pass' => 'amqp.password Password. Note: Max 128 characters.',
33+
* 'lazy' => 'the connection will be performed as later as possible, if the option set to true',
34+
* 'stream' => 'stream or socket connection',
35+
* 'receive_method' => 'Could be either basic_get or basic_consume',
36+
* ]
37+
*
38+
* or
39+
*
40+
* amqp://user:pass@host:10000/vhost?lazy=true&socket=true
41+
*
42+
* @param array|string $config
43+
*/
44+
public function __construct($config = 'amqp://')
45+
{
46+
if (empty($config) || 'amqp://' === $config) {
47+
$config = [];
48+
} elseif (is_string($config)) {
49+
$config = $this->parseDsn($config);
50+
} elseif (is_array($config)) {
51+
} else {
52+
throw new \LogicException('The config must be either an array of options, a DSN string or null');
53+
}
54+
55+
$this->config = array_replace($this->defaultConfig(), $config);
56+
57+
$supportedMethods = ['basic_get', 'basic_consume'];
58+
if (false == in_array($this->config['receive_method'], $supportedMethods, true)) {
59+
throw new \LogicException(sprintf(
60+
'Invalid "receive_method" option value "%s". It could be only "%s"',
61+
$this->config['receive_method'],
62+
implode('", "', $supportedMethods)
63+
));
64+
}
65+
}
66+
67+
/**
68+
* @return AmqpContext
69+
*/
70+
public function createContext()
71+
{
72+
return new AmqpContext($this->establishConnection(), $this->config['receive_method']);
73+
}
74+
75+
/**
76+
* @return AbstractConnection
77+
*/
78+
private function establishConnection()
79+
{
80+
if (false == $this->connection) {
81+
if ($this->config['stream']) {
82+
if ($this->config['lazy']) {
83+
$con = new AMQPLazyConnection(
84+
$this->config['host'],
85+
$this->config['port'],
86+
$this->config['user'],
87+
$this->config['pass'],
88+
$this->config['vhost'],
89+
$this->config['insist'],
90+
$this->config['login_method'],
91+
$this->config['login_response'],
92+
$this->config['locale'],
93+
$this->config['connection_timeout'],
94+
$this->config['read_write_timeout'],
95+
null,
96+
$this->config['keepalive'],
97+
$this->config['heartbeat']
98+
);
99+
} else {
100+
$con = new AMQPStreamConnection(
101+
$this->config['host'],
102+
$this->config['port'],
103+
$this->config['user'],
104+
$this->config['pass'],
105+
$this->config['vhost'],
106+
$this->config['insist'],
107+
$this->config['login_method'],
108+
$this->config['login_response'],
109+
$this->config['locale'],
110+
$this->config['connection_timeout'],
111+
$this->config['read_write_timeout'],
112+
null,
113+
$this->config['keepalive'],
114+
$this->config['heartbeat']
115+
);
116+
}
117+
} else {
118+
if ($this->config['lazy']) {
119+
$con = new AMQPLazySocketConnection(
120+
$this->config['host'],
121+
$this->config['port'],
122+
$this->config['user'],
123+
$this->config['pass'],
124+
$this->config['vhost'],
125+
$this->config['insist'],
126+
$this->config['login_method'],
127+
$this->config['login_response'],
128+
$this->config['locale'],
129+
$this->config['read_timeout'],
130+
$this->config['keepalive'],
131+
$this->config['write_timeout'],
132+
$this->config['heartbeat']
133+
);
134+
} else {
135+
$con = new AMQPSocketConnection(
136+
$this->config['host'],
137+
$this->config['port'],
138+
$this->config['user'],
139+
$this->config['pass'],
140+
$this->config['vhost'],
141+
$this->config['insist'],
142+
$this->config['login_method'],
143+
$this->config['login_response'],
144+
$this->config['locale'],
145+
$this->config['read_timeout'],
146+
$this->config['keepalive'],
147+
$this->config['write_timeout'],
148+
$this->config['heartbeat']
149+
);
150+
}
151+
}
152+
153+
$this->connection = $con;
154+
}
155+
156+
return $this->connection;
157+
}
158+
159+
/**
160+
* @param string $dsn
161+
*
162+
* @return array
163+
*/
164+
private function parseDsn($dsn)
165+
{
166+
$dsnConfig = parse_url($dsn);
167+
if (false === $dsnConfig) {
168+
throw new \LogicException(sprintf('Failed to parse DSN "%s"', $dsn));
169+
}
170+
171+
$dsnConfig = array_replace([
172+
'scheme' => null,
173+
'host' => null,
174+
'port' => null,
175+
'user' => null,
176+
'pass' => null,
177+
'path' => null,
178+
'query' => null,
179+
], $dsnConfig);
180+
181+
if ('amqp' !== $dsnConfig['scheme']) {
182+
throw new \LogicException(sprintf('The given DSN scheme "%s" is not supported. Could be "amqp" only.', $dsnConfig['scheme']));
183+
}
184+
185+
if ($dsnConfig['query']) {
186+
$query = [];
187+
parse_str($dsnConfig['query'], $query);
188+
189+
$dsnConfig = array_replace($query, $dsnConfig);
190+
}
191+
192+
$dsnConfig['vhost'] = ltrim($dsnConfig['path'], '/');
193+
194+
unset($dsnConfig['scheme'], $dsnConfig['query'], $dsnConfig['fragment'], $dsnConfig['path']);
195+
196+
$dsnConfig = array_map(function ($value) {
197+
return urldecode($value);
198+
}, $dsnConfig);
199+
200+
return $dsnConfig;
201+
}
202+
203+
/**
204+
* @return array
205+
*/
206+
private function defaultConfig()
207+
{
208+
return [
209+
'stream' => true,
210+
'lazy' => true,
211+
'host' => 'localhost',
212+
'port' => 5672,
213+
'user' => 'guest',
214+
'pass' => 'guest',
215+
'vhost' => '/',
216+
'insist' => false,
217+
'login_method' => 'AMQPLAIN',
218+
'login_response' => null,
219+
'locale' => 'en_US',
220+
'read_timeout' => 3,
221+
'keepalive' => false,
222+
'write_timeout' => 3,
223+
'heartbeat' => 0,
224+
'connection_timeout' => 3.0,
225+
'read_write_timeout' => 3.0,
226+
'receive_method' => 'basic_get',
227+
];
228+
}
229+
}

0 commit comments

Comments
 (0)