Skip to content
This repository was archived by the owner on Oct 30, 2024. It is now read-only.

Commit 0bff9c3

Browse files
repejotarepejota
committed
Encoded connections & Minimal PHP
* Encoded connection for publish and subscribe. * Endoded connection for request. * Testing encoded connection for strings and arrays. * Minimal version PHP 5.6+
1 parent 654d03a commit 0bff9c3

9 files changed

Lines changed: 164 additions & 46 deletions

File tree

‎.travis.yml‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@ services:
33
- docker
44
language: php
55
php:
6-
- 5.5
76
- 5.6
87
- 7
98
- 7.1

‎README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ A PHP client for the [NATS messaging system](https://nats.io).
2121
Requirements
2222
------------
2323

24-
* php 5.5+
24+
* php 5.6+
2525
* [nats](https://github.com/derekcollison/nats) or [gnatsd](https://github.com/apcera/gnatsd)
2626

2727

‎composer.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
"ircmaxell/random-lib": "^1.1"
99
},
1010
"require-dev": {
11-
"phpunit/phpunit": "4.7.*",
11+
"phpunit/phpunit": "5.*",
1212
"satooshi/php-coveralls": "dev-master",
1313
"squizlabs/php_codesniffer": "~2.0",
1414
"phpspec/phpspec": "~2.0"

‎foo.php‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
<?php
2+
require_once __DIR__.'/vendor/autoload.php';
3+
4+
use Nats\ConnectionOptions;
5+
use Nats\EncodedConnection;
6+
use Nats\Encoders\JSONEncoder;
7+
8+
$options = new ConnectionOptions();
9+
$encoder = new JSONEncoder();
10+
$c = new EncodedConnection($options, $encoder);
11+
$c->connect();
12+
13+
$a = array('foo', 'bar', 1, 2, 3);
14+
15+
$callback = function($message) {
16+
print($message->getBody()[1]);
17+
};
18+
$sid = $c->subscribe('foo', $callback);
19+
20+
$c->request('foo', $a);

‎src/Nats/Connection.php‎

Lines changed: 11 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -315,23 +315,24 @@ public function ping()
315315
/**
316316
* Request does a request and executes a callback with the response.
317317
*
318-
* @param string $subject Message topic.
319-
* @param string $payload Message data.
320-
* @param mixed $callback Closure to be executed as callback.
321-
* @param integer $wait Number of messages to wait for.
318+
* @param string $subject Message topic.
319+
* @param string $payload Message data.
322320
*
323321
* @return void
324322
*/
325-
public function request($subject, $payload, $callback, $wait = 1)
323+
public function request($subject, $payload)
326324
{
327325
$inbox = uniqid('_INBOX.');
328-
$this->subscribe($inbox, $callback);
329-
326+
$this->subscribe(
327+
$inbox,
328+
function ($message) {
329+
}
330+
);
330331
$msg = 'PUB '.$subject.' '.$inbox.' '.strlen($payload);
331332
$this->send($msg."\r\n".$payload);
332333
$this->pubs += 1;
333334

334-
$this->wait($wait);
335+
$this->wait(1);
335336
}
336337

337338

@@ -365,7 +366,6 @@ public function subscribe($subject, \Closure $callback)
365366
$msg = 'SUB '.$subject.' '.$sid;
366367
$this->send($msg);
367368
$this->subscriptions[$sid] = $callback;
368-
369369
return $sid;
370370
}
371371

@@ -385,7 +385,6 @@ public function queueSubscribe($subject, $queue, \Closure $callback)
385385
$msg = 'SUB '.$subject.' '.$queue.' '.$sid;
386386
$this->send($msg);
387387
$this->subscriptions[$sid] = $callback;
388-
389388
return $sid;
390389
}
391390

@@ -401,7 +400,6 @@ public function unsubscribe($sid)
401400
{
402401
$msg = 'UNSUB '.$sid;
403402
$this->send($msg);
404-
405403
unset($this->subscriptions[$sid]);
406404
}
407405

@@ -422,9 +420,8 @@ private function handlePING()
422420
*
423421
* @param string $line Message command from Nats.
424422
*
425-
* @return void
426-
* @throws Exception If subscription not found.
427-
* @codeCoverageIgnore
423+
* @return void
424+
* @throws Exception If subscription not found.
428425
*/
429426
private function handleMSG($line)
430427
{
@@ -454,8 +451,6 @@ private function handleMSG($line)
454451
} else {
455452
throw Exception::forSubscriptionCallbackInvalid($sid);
456453
}
457-
458-
return;
459454
}
460455

461456

‎src/Nats/EncodedConnection.php‎

Lines changed: 15 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -32,20 +32,15 @@ public function __construct(ConnectionOptions $options = null, \Nats\Encoders\En
3232
/**
3333
* Request does a request and executes a callback with the response.
3434
*
35-
* @param string $subject Message topic.
36-
* @param string $payload Message data.
37-
* @param mixed $callback Closure to be executed as callback.
38-
* @param integer $wait Number of messages to wait for.
35+
* @param string $subject Message topic.
36+
* @param string $payload Message data.
3937
*
4038
* @return void
4139
*/
42-
public function request($subject, $payload, $callback, $wait = 1)
40+
public function request($subject, $payload)
4341
{
44-
$payload = $this->encoder->encode($payload);
45-
$decodeCallback = function ($payload) use ($callback) {
46-
$callback($this->encoder->decode($payload));
47-
};
48-
parent::request($subject, $payload, $decodeCallback, $wait);
42+
$payload = $this->encoder->encode($payload);
43+
parent::request($subject, $payload);
4944
}
5045

5146
/**
@@ -68,14 +63,15 @@ public function publish($subject, $payload = null)
6863
* @param string $subject Message topic.
6964
* @param \Closure $callback Closure to be executed as callback.
7065
*
71-
* @return string
66+
* @return void
7267
*/
7368
public function subscribe($subject, \Closure $callback)
7469
{
75-
$decodeCallback = function ($payload) use ($callback) {
76-
$callback($this->encoder->decode($payload));
70+
$c = function ($message) use ($callback) {
71+
$message->setBody($this->encoder->decode($message->getBody()));
72+
$callback($message);
7773
};
78-
return parent::subscribe($subject, $decodeCallback);
74+
parent::subscribe($subject, $c);
7975
}
8076

8177
/**
@@ -85,13 +81,14 @@ public function subscribe($subject, \Closure $callback)
8581
* @param string $queue Queue name.
8682
* @param \Closure $callback Closure to be executed as callback.
8783
*
88-
* @return string
84+
* @return void
8985
*/
9086
public function queueSubscribe($subject, $queue, \Closure $callback)
9187
{
92-
$decodeCallback = function ($payload) use ($callback) {
93-
$callback($this->encoder->decode($payload));
88+
$c = function ($message) use ($callback) {
89+
$message->setBody($this->encoder->decode($message->getBody()));
90+
$callback($message);
9491
};
95-
return parent::queueSubscribe($subject, $queue, $decodeCallback);
92+
parent::queueSubscribe($subject, $queue, $c);
9693
}
9794
}

‎src/Nats/Encoders/JSONEncoder.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
*
99
* @package Nats
1010
*/
11-
class JSONEncoder implements \Nats\Encoders\Encoder
11+
class JSONEncoder implements Encoder
1212
{
1313

1414

‎test/ConnectionTest.php‎

Lines changed: 4 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,8 @@
22
namespace Nats\tests\Unit;
33

44
use Nats;
5+
use Nats\Connection;
56
use Nats\ConnectionOptions;
6-
use Prophecy\Argument;
77

88
/**
99
* Class ConnectionTest.
@@ -27,7 +27,7 @@ class ConnectionTest extends \PHPUnit_Framework_TestCase
2727
public function setUp()
2828
{
2929
$options = new ConnectionOptions();
30-
$this->c = new Nats\Connection($options);
30+
$this->c = new Connection($options);
3131
$this->c->connect();
3232
}
3333

@@ -113,11 +113,7 @@ function ($res) {
113113

114114
$this->c->request(
115115
'sayhello'.$i,
116-
'McFly',
117-
function ($message) {
118-
$this->assertNotNull($message);
119-
$this->assertEquals($message, 'Hello, McFly !!!');
120-
}
116+
'McFly'
121117
);
122118

123119
$i++;
@@ -225,7 +221,7 @@ public function testRefusedConnection()
225221
$options->setHost('localhost');
226222
$options->setPort(4223);
227223

228-
$c = new Nats\Connection($options);
224+
$c = new Connection($options);
229225
$c->connect();
230226
$this->assertFalse($this->c->isConnected());
231227
}

‎test/EncodedConnectionTest.php‎

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,111 @@
1+
<?php
2+
namespace Nats\tests\Unit;
3+
4+
use Nats;
5+
use Nats\ConnectionOptions;
6+
use Nats\EncodedConnection;
7+
use Nats\Encoders\JSONEncoder;
8+
9+
/**
10+
* Class EncodedConnectionTest.
11+
*/
12+
class EncodedConnectionTest extends \PHPUnit_Framework_TestCase
13+
{
14+
15+
/**
16+
* Client.
17+
*
18+
* @var Nats\Connection Client
19+
*/
20+
private $c;
21+
22+
23+
/**
24+
* SetUp test suite.
25+
*
26+
* @return void
27+
*/
28+
public function setUp()
29+
{
30+
$encoder = new JSONEncoder();
31+
$options = new ConnectionOptions();
32+
$this->c = new EncodedConnection($options, $encoder);
33+
$this->c->connect();
34+
}
35+
36+
37+
/**
38+
* Test Connection.
39+
*
40+
* @return void
41+
*/
42+
public function testConnection()
43+
{
44+
// Connect.
45+
$this->c->connect();
46+
$this->assertTrue($this->c->isConnected());
47+
48+
// Disconnect.
49+
$this->c->close();
50+
$this->assertFalse($this->c->isConnected());
51+
}
52+
53+
/**
54+
* Test Publish command.
55+
*
56+
* @return void
57+
*/
58+
public function testPublish()
59+
{
60+
$this->c->ping();
61+
$this->c->publish('foo', 'bar');
62+
$count = $this->c->pubsCount();
63+
$this->assertInternalType('int', $count);
64+
$this->assertGreaterThan(0, $count);
65+
$this->c->close();
66+
}
67+
68+
69+
/**
70+
* Test Request command.
71+
*
72+
* @return void
73+
*/
74+
public function testRequest()
75+
{
76+
$this->c->subscribe(
77+
'sayhello',
78+
function ($res) {
79+
$res->reply('Hello, '.$res->getBody().' !!!');
80+
}
81+
);
82+
83+
$this->c->request(
84+
'sayhello',
85+
'McFly'
86+
);
87+
}
88+
89+
/**
90+
* Test Request command.
91+
*
92+
* @return void
93+
*/
94+
public function testRequestArray()
95+
{
96+
$this->c->subscribe(
97+
'sayhello',
98+
function ($res) {
99+
$res->reply('Hello, '.$res->getBody()[1].' !!!');
100+
}
101+
);
102+
103+
$this->c->request(
104+
'sayhello',
105+
[
106+
'foo',
107+
'McFly',
108+
]
109+
);
110+
}
111+
}

0 commit comments

Comments
 (0)