Skip to content

Commit f8be56a

Browse files
fibulafibula
committed
add spec tests
1 parent c88b011 commit f8be56a

17 files changed

+588
-17
lines changed

‎AmqpConnectionFactory.php

+161-10
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,9 @@
44

55
use Interop\Queue\PsrConnectionFactory;
66
use PhpAmqpLib\Connection\AbstractConnection;
7+
use PhpAmqpLib\Connection\AMQPLazyConnection;
8+
use PhpAmqpLib\Connection\AMQPLazySocketConnection;
9+
use PhpAmqpLib\Connection\AMQPSocketConnection;
710
use PhpAmqpLib\Connection\AMQPStreamConnection;
811

912
class AmqpConnectionFactory implements PsrConnectionFactory
@@ -19,10 +22,35 @@ class AmqpConnectionFactory implements PsrConnectionFactory
1922
private $connection;
2023

2124
/**
22-
* @param array $config
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+
* ]
36+
*
37+
* or
38+
*
39+
* amqp://user:pass@host:10000/vhost?lazy=true&socket=true
40+
*
41+
* @param array|string $config
2342
*/
24-
public function __construct(array $config = [])
43+
public function __construct($config = 'amqp://')
2544
{
45+
if (empty($config) || 'amqp://' === $config) {
46+
$config = [];
47+
} elseif (is_string($config)) {
48+
$config = $this->parseDsn($config);
49+
} elseif (is_array($config)) {
50+
} else {
51+
throw new \LogicException('The config must be either an array of options, a DSN string or null');
52+
}
53+
2654
$this->config = array_replace($this->defaultConfig(), $config);
2755
}
2856

@@ -40,29 +68,152 @@ public function createContext()
4068
private function establishConnection()
4169
{
4270
if (false == $this->connection) {
43-
$this->connection = new AMQPStreamConnection(
44-
$this->config['host'],
45-
$this->config['port'],
46-
$this->config['user'],
47-
$this->config['pass'],
48-
$this->config['vhost']
49-
);
71+
if ($this->config['stream']) {
72+
if ($this->config['lazy']) {
73+
$con = new AMQPLazyConnection(
74+
$this->config['host'],
75+
$this->config['port'],
76+
$this->config['user'],
77+
$this->config['pass'],
78+
$this->config['vhost'],
79+
$this->config['insist'],
80+
$this->config['login_method'],
81+
$this->config['login_response'],
82+
$this->config['locale'],
83+
$this->config['connection_timeout'],
84+
$this->config['read_write_timeout'],
85+
null,
86+
$this->config['keepalive'],
87+
$this->config['heartbeat']
88+
);
89+
} else {
90+
$con = new AMQPStreamConnection(
91+
$this->config['host'],
92+
$this->config['port'],
93+
$this->config['user'],
94+
$this->config['pass'],
95+
$this->config['vhost'],
96+
$this->config['insist'],
97+
$this->config['login_method'],
98+
$this->config['login_response'],
99+
$this->config['locale'],
100+
$this->config['connection_timeout'],
101+
$this->config['read_write_timeout'],
102+
null,
103+
$this->config['keepalive'],
104+
$this->config['heartbeat']
105+
);
106+
}
107+
} else {
108+
if ($this->config['lazy']) {
109+
$con = new AMQPLazySocketConnection(
110+
$this->config['host'],
111+
$this->config['port'],
112+
$this->config['user'],
113+
$this->config['pass'],
114+
$this->config['vhost'],
115+
$this->config['insist'],
116+
$this->config['login_method'],
117+
$this->config['login_response'],
118+
$this->config['locale'],
119+
$this->config['read_timeout'],
120+
$this->config['keepalive'],
121+
$this->config['write_timeout'],
122+
$this->config['heartbeat']
123+
);
124+
} else {
125+
$con = new AMQPSocketConnection(
126+
$this->config['host'],
127+
$this->config['port'],
128+
$this->config['user'],
129+
$this->config['pass'],
130+
$this->config['vhost'],
131+
$this->config['insist'],
132+
$this->config['login_method'],
133+
$this->config['login_response'],
134+
$this->config['locale'],
135+
$this->config['read_timeout'],
136+
$this->config['keepalive'],
137+
$this->config['write_timeout'],
138+
$this->config['heartbeat']
139+
);
140+
}
141+
}
142+
143+
$this->connection = $con;
50144
}
51145

52146
return $this->connection;
53147
}
54148

149+
/**
150+
* @param string $dsn
151+
*
152+
* @return array
153+
*/
154+
private function parseDsn($dsn)
155+
{
156+
$dsnConfig = parse_url($dsn);
157+
if (false === $dsnConfig) {
158+
throw new \LogicException(sprintf('Failed to parse DSN "%s"', $dsn));
159+
}
160+
161+
$dsnConfig = array_replace([
162+
'scheme' => null,
163+
'host' => null,
164+
'port' => null,
165+
'user' => null,
166+
'pass' => null,
167+
'path' => null,
168+
'query' => null,
169+
], $dsnConfig);
170+
171+
if ('amqp' !== $dsnConfig['scheme']) {
172+
throw new \LogicException(sprintf('The given DSN scheme "%s" is not supported. Could be "amqp" only.', $dsnConfig['scheme']));
173+
}
174+
175+
if ($dsnConfig['query']) {
176+
$query = [];
177+
parse_str($dsnConfig['query'], $query);
178+
179+
$dsnConfig = array_replace($query, $dsnConfig);
180+
}
181+
182+
$dsnConfig['vhost'] = ltrim($dsnConfig['path'], '/');
183+
184+
unset($dsnConfig['scheme'], $dsnConfig['query'], $dsnConfig['fragment'], $dsnConfig['path']);
185+
186+
$config = array_replace($this->defaultConfig(), $dsnConfig);
187+
$config = array_map(function ($value) {
188+
return urldecode($value);
189+
}, $config);
190+
191+
return $config;
192+
}
193+
55194
/**
56195
* @return array
57196
*/
58197
private function defaultConfig()
59198
{
60199
return [
200+
'stream' => true,
201+
'lazy' => true,
61202
'host' => 'localhost',
62203
'port' => 5672,
63-
'vhost' => '/',
64204
'user' => 'guest',
65205
'pass' => 'guest',
206+
'vhost' => '/',
207+
'insist' => false,
208+
'login_method' => 'AMQPLAIN',
209+
'login_response' => null,
210+
'locale' => 'en_US',
211+
'read_timeout' => 3,
212+
'keepalive' => false,
213+
'write_timeout' => 3,
214+
'heartbeat' => 0,
215+
'connection_timeout' => 3.0,
216+
'read_write_timeout' => 3.0,
66217
];
67218
}
68219
}

‎AmqpContext.php

+25-2
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
use Interop\Queue\InvalidDestinationException;
77
use Interop\Queue\PsrContext;
88
use Interop\Queue\PsrDestination;
9+
use Interop\Queue\PsrQueue;
910
use Interop\Queue\PsrTopic;
1011
use PhpAmqpLib\Channel\AMQPChannel;
1112
use PhpAmqpLib\Connection\AbstractConnection;
@@ -37,7 +38,7 @@ public function __construct(AbstractConnection $connection)
3738
*
3839
* @return AmqpMessage
3940
*/
40-
public function createMessage($body = null, array $properties = [], array $headers = [])
41+
public function createMessage($body = '', array $properties = [], array $headers = [])
4142
{
4243
return new AmqpMessage($body, $properties, $headers);
4344
}
@@ -69,7 +70,17 @@ public function createTopic($name)
6970
*/
7071
public function createConsumer(PsrDestination $destination)
7172
{
72-
InvalidDestinationException::assertDestinationInstanceOf($destination, AmqpQueue::class);
73+
$destination instanceof PsrTopic
74+
? InvalidDestinationException::assertDestinationInstanceOf($destination, AmqpTopic::class)
75+
: InvalidDestinationException::assertDestinationInstanceOf($destination, AmqpQueue::class)
76+
;
77+
78+
if ($destination instanceof AmqpTopic) {
79+
$queue = $this->createTemporaryQueue();
80+
$this->bind($destination, $queue);
81+
82+
return new AmqpConsumer($this->getChannel(), $queue);
83+
}
7384

7485
return new AmqpConsumer($this->getChannel(), $destination);
7586
}
@@ -189,6 +200,18 @@ public function bind(PsrDestination $source, PsrDestination $target)
189200
}
190201
}
191202

203+
/**
204+
* Purge all messages from the given queue.
205+
*
206+
* @param PsrQueue $queue
207+
*/
208+
public function purge(PsrQueue $queue)
209+
{
210+
InvalidDestinationException::assertDestinationInstanceOf($queue, AmqpQueue::class);
211+
212+
$this->getChannel()->queue_purge($queue->getQueueName());
213+
}
214+
192215
public function close()
193216
{
194217
if ($this->channel) {

‎AmqpMessage.php

+4-4
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
class AmqpMessage implements PsrMessage
88
{
99
/**
10-
* @var string|null
10+
* @var string
1111
*/
1212
private $body;
1313

@@ -51,7 +51,7 @@ class AmqpMessage implements PsrMessage
5151
* @param array $properties
5252
* @param array $headers
5353
*/
54-
public function __construct($body = null, array $properties = [], array $headers = [])
54+
public function __construct($body = '', array $properties = [], array $headers = [])
5555
{
5656
$this->body = $body;
5757
$this->properties = $properties;
@@ -60,15 +60,15 @@ public function __construct($body = null, array $properties = [], array $headers
6060
}
6161

6262
/**
63-
* @return null|string
63+
* @return string
6464
*/
6565
public function getBody()
6666
{
6767
return $this->body;
6868
}
6969

7070
/**
71-
* @param string|null $body
71+
* @param string $body
7272
*/
7373
public function setBody($body)
7474
{

‎LICENSE

+20
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
The MIT License (MIT)
2+
Copyright (c) 2017 Paul McLaren
3+
4+
Permission is hereby granted, free of charge, to any person obtaining a copy
5+
of this software and associated documentation files (the "Software"), to deal
6+
in the Software without restriction, including without limitation the rights
7+
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
8+
copies of the Software, and to permit persons to whom the Software is furnished
9+
to do so, subject to the following conditions:
10+
11+
The above copyright notice and this permission notice shall be included in all
12+
copies or substantial portions of the Software.
13+
14+
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
15+
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
16+
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
17+
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
18+
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
19+
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
20+
THE SOFTWARE.

‎Tests/Spec/AmqpConnectionFactoryTest.php

+14
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
<?php
2+
3+
namespace Enqueue\AmqpLib\Tests\Spec;
4+
5+
use Enqueue\AmqpLib\AmqpConnectionFactory;
6+
use Interop\Queue\Spec\PsrConnectionFactorySpec;
7+
8+
class AmqpConnectionFactoryTest extends PsrConnectionFactorySpec
9+
{
10+
protected function createConnectionFactory()
11+
{
12+
return new AmqpConnectionFactory();
13+
}
14+
}

‎Tests/Spec/AmqpContextTest.php

+24
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
<?php
2+
namespace Enqueue\AmqpLib\Tests\Spec;
3+
4+
use Enqueue\AmqpLib\AmqpContext;
5+
use Interop\Queue\Spec\PsrContextSpec;
6+
use PhpAmqpLib\Channel\AMQPChannel;
7+
use PhpAmqpLib\Connection\AbstractConnection;
8+
9+
class AmqpContextTest extends PsrContextSpec
10+
{
11+
protected function createContext()
12+
{
13+
$channel = $this->createMock(AMQPChannel::class);
14+
15+
$con = $this->createMock(AbstractConnection::class);
16+
$con
17+
->expects($this->any())
18+
->method('channel')
19+
->willReturn($channel)
20+
;
21+
22+
return new AmqpContext($con);
23+
}
24+
}

‎Tests/Spec/AmqpMessageTest.php

+17
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
<?php
2+
3+
namespace Enqueue\AmqpLib\Tests\Spec;
4+
5+
use Enqueue\AmqpLib\AmqpMessage;
6+
use Interop\Queue\Spec\PsrMessageSpec;
7+
8+
class AmqpMessageTest extends PsrMessageSpec
9+
{
10+
/**
11+
* {@inheritdoc}
12+
*/
13+
protected function createMessage()
14+
{
15+
return new AmqpMessage();
16+
}
17+
}

‎Tests/Spec/AmqpQueueTest.php

+14
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
<?php
2+
3+
namespace Enqueue\AmqpLib\Tests\Spec;
4+
5+
use Enqueue\AmqpLib\AmqpQueue;
6+
use Interop\Queue\Spec\PsrQueueSpec;
7+
8+
class AmqpQueueTest extends PsrQueueSpec
9+
{
10+
protected function createQueue()
11+
{
12+
return new AmqpQueue(self::EXPECTED_QUEUE_NAME);
13+
}
14+
}

0 commit comments

Comments
 (0)