Skip to content

Commit 8c31eb1

Browse files
oliverlorenzoliverlorenz
authored
Merge pull request #29 from jjok/update-dependencies
Update React dependencies
2 parents 44ff9ba + 0f586c7 commit 8c31eb1

File tree

9 files changed

+98
-107
lines changed

9 files changed

+98
-107
lines changed

‎.travis.yml‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,23 @@
1+
env:
2+
global:
3+
- CC_TEST_REPORTER_ID=731b1224e90b7ad7304e86768a8ed7b98b63d9e30bdc37b6295097c7bb5bedfd
4+
15
language: php
26

37
php:
4-
- 5.5
58
- 5.6
69
- 7.0
710
- 7.1
11+
- 7.2
12+
13+
before_script:
14+
- curl -L https://codeclimate.com/downloads/test-reporter/test-reporter-latest-linux-amd64 > ./cc-test-reporter
15+
- chmod +x ./cc-test-reporter
16+
- if [ $(phpenv version-name) = "7.2" ]; then ./cc-test-reporter before-build; fi
817

918
script:
1019
- composer install
11-
- php vendor/bin/phpunit --coverage-text --coverage-clover build/logs/clover.xml --bootstrap tests/bootstrap.php tests/
12-
- CODECLIMATE_REPO_TOKEN=731b1224e90b7ad7304e86768a8ed7b98b63d9e30bdc37b6295097c7bb5bedfd ./vendor/bin/test-reporter
20+
- php vendor/bin/phpunit -c ./tests/phpunit.xml --coverage-text --coverage-clover build/logs/clover.xml tests/
21+
22+
after_script:
23+
- if [ $(phpenv version-name) = "7.2" ]; then ./cc-test-reporter after-build --exit-code $TRAVIS_TEST_RESULT; fi

‎composer.json‎

Lines changed: 2 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@
33
"description": "a mqtt client library for php",
44
"keywords": ["mqtt", "IoT"],
55
"license": "LGPL",
6-
"minimum-stability": "dev",
76
"authors": [
87
{
98
"name": "Oliver Lorenz",
@@ -16,14 +15,9 @@
1615
}
1716
},
1817
"require": {
19-
"react/socket-client": "0.5.*",
20-
"react/stream": "0.4.*",
21-
"react/event-loop": "0.4.*",
22-
"react/dns": "0.4.*",
23-
"react/promise": "2.4.*"
18+
"react/socket": "^1.1"
2419
},
2520
"require-dev": {
26-
"phpunit/phpunit": "4.6.*",
27-
"codeclimate/php-test-reporter": "dev-master"
21+
"phpunit/phpunit": "^5.0.0"
2822
}
2923
}

‎examples/config.php.example‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,7 @@
11
<?php
22

33
$config = array(
4-
'server' => 'yourMqttBroker.tld',
5-
'port' => 1883,
4+
'broker' => 'yourMqttBroker.tld:1883',
65
'options' => new \oliverlorenz\reactphpmqtt\packet\ConnectionOptions(array(
76
'keepAlive' => 120,
87
)),

‎examples/connectPublish.php‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,17 +2,19 @@
22

33
use oliverlorenz\reactphpmqtt\ClientFactory;
44
use oliverlorenz\reactphpmqtt\protocol\Version4;
5-
use React\Stream\Stream;
5+
use React\Socket\ConnectionInterface as Connection;
66

77
require __DIR__ . '/../vendor/autoload.php';
88

99
$config = require 'config.php';
1010

11-
$connector = ClientFactory::createClient(new Version4(), '8.8.8.8');
11+
$client = ClientFactory::createClient(new Version4(), '8.8.8.8');
1212

13-
$p = $connector->create($config['server'], $config['port'], $config['options']);
14-
$p->then(function(Stream $stream) use ($connector) {
15-
return $connector->publish($stream, 'hello/world', 'example message');
13+
$p = $client->connect($config['broker'], $config['options']);
14+
$client->getLoop()->addPeriodicTimer(10, function () use ($p, $client) {
15+
$p->then(function(Connection $stream) use ($client) {
16+
return $client->publish($stream, 'hello/world', 'example message');
17+
});
1618
});
1719

18-
$connector->getLoop()->run();
20+
$client->getLoop()->run();

‎examples/connectSubscribe.php‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -3,16 +3,16 @@
33
use oliverlorenz\reactphpmqtt\ClientFactory;
44
use oliverlorenz\reactphpmqtt\packet\Publish;
55
use oliverlorenz\reactphpmqtt\protocol\Version4;
6-
use React\Stream\Stream;
6+
use React\Socket\ConnectionInterface as Stream;
77

88
require __DIR__ . '/../vendor/autoload.php';
99

1010
$config = require 'config.php';
1111

12-
$connector = ClientFactory::createClient(new Version4(), '8.8.8.8');
12+
$client = ClientFactory::createClient(new Version4(), '8.8.8.8');
1313

14-
$p = $connector->create($config['server'], $config['port'], $config['options']);
15-
$p->then(function(Stream $stream) use ($connector) {
14+
$p = $client->connect($config['broker'], $config['options']);
15+
$p->then(function(Stream $stream) use ($client) {
1616
$stream->on(Publish::EVENT, function(Publish $message) {
1717
printf(
1818
'Received payload "%s" for topic "%s"%s',
@@ -22,7 +22,7 @@
2222
);
2323
});
2424

25-
return $connector->subscribe($stream, 'hello/world', 0);
25+
$client->subscribe($stream, 'hello/world', 0);
2626
});
2727

28-
$connector->getLoop()->run();
28+
$client->getLoop()->run();

‎src/ClientFactory.php‎

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5,29 +5,34 @@
55
use oliverlorenz\reactphpmqtt\protocol\Version;
66
use React\Dns\Resolver\Factory as DnsResolverFactory;
77
use React\EventLoop\Factory as EventLoopFactory;
8+
use React\Socket\DnsConnector;
9+
use React\Socket\SecureConnector;
10+
use React\Socket\TcpConnector;
811

912
class ClientFactory
1013
{
1114
public static function createClient(Version $version, $resolverIp = '8.8.8.8')
1215
{
1316
$loop = EventLoopFactory::create();
14-
$resolver = self::createDnsResolver($resolverIp, $loop);
17+
$connector = self::createDnsConnector($resolverIp, $loop);
1518

16-
return new Connector($loop, $resolver, $version);
19+
return new MqttClient($loop, $connector, $version);
1720
}
1821

1922
public static function createSecureClient(Version $version, $resolverIp = '8.8.8.8')
2023
{
2124
$loop = EventLoopFactory::create();
22-
$resolver = self::createDnsResolver($resolverIp, $loop);
25+
$connector = self::createDnsConnector($resolverIp, $loop);
26+
$secureConnector = new SecureConnector($connector, $loop);
2327

24-
return new SecureConnector($loop, $resolver, $version);
28+
return new MqttClient($loop, $secureConnector, $version);
2529
}
2630

27-
private static function createDnsResolver($resolverIp, $loop)
31+
private static function createDnsConnector($resolverIp, $loop)
2832
{
2933
$dnsResolverFactory = new DnsResolverFactory();
34+
$resolver = $dnsResolverFactory->createCached($resolverIp, $loop);
3035

31-
return $dnsResolverFactory->createCached($resolverIp, $loop);
36+
return new DnsConnector(new TcpConnector($loop), $resolver);
3237
}
3338
}

‎src/Connector.php‎ renamed to ‎src/MqttClient.php‎

Lines changed: 54 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -22,66 +22,66 @@
2222
use oliverlorenz\reactphpmqtt\packet\UnsubscribeAck;
2323
use oliverlorenz\reactphpmqtt\protocol\Version;
2424
use oliverlorenz\reactphpmqtt\protocol\Violation as ProtocolViolation;
25-
use React\Dns\Resolver\Resolver;
26-
use React\EventLoop\LoopInterface;
25+
use React\EventLoop\LoopInterface as Loop;
2726
use React\EventLoop\Timer\Timer;
2827
use React\Promise\Deferred;
2928
use React\Promise\FulfilledPromise;
3029
use React\Promise\PromiseInterface;
31-
use React\SocketClient\ConnectorInterface;
32-
use React\Stream\Stream;
33-
34-
class Connector implements ConnectorInterface {
30+
use React\Socket\ConnectionInterface as Connection;
31+
use React\Socket\ConnectorInterface as ReactConnector;
3532

33+
class MqttClient
34+
{
3635
/**
37-
* @var $loop LoopInterface
36+
* @var $loop Loop
3837
*/
3938
private $loop;
40-
protected $socketConnector;
39+
private $socketConnector;
4140
private $version;
41+
4242
private $messageCounter = 1;
4343

44-
public function __construct(LoopInterface $loop, Resolver $resolver, Version $version)
44+
public function __construct(Loop $loop, ReactConnector $connector, Version $version)
4545
{
4646
$this->version = $version;
47-
$this->socketConnector = new \React\SocketClient\Connector($loop, $resolver);
47+
$this->socketConnector = $connector;
4848
$this->loop = $loop;
4949
}
5050

5151
/**
5252
* Creates a new connection
5353
*
54-
* @param string $host
55-
* @param int $port [optional]
54+
* @param string $uri
5655
* @param ConnectionOptions|null $options [optional]
5756
*
5857
* @return PromiseInterface Resolves to a \React\Stream\Stream once a connection has been established
5958
*/
60-
public function create(
61-
$host,
62-
$port = 1883,
59+
public function connect(
60+
$uri,
6361
ConnectionOptions $options = null
6462
) {
6563
// Set default connection options, if none provided
6664
if($options == null) {
6765
$options = $this->getDefaultConnectionOptions();
6866
}
6967

70-
return $this->socketConnector->create($host, $port)
71-
->then(function (Stream $stream) use ($options) {
72-
return $this->connect($stream, $options);
73-
})
74-
->then(function (Stream $stream) {
75-
return $this->listenForPackets($stream);
76-
})
77-
->then(function(Stream $stream) use ($options) {
78-
return $this->keepAlive($stream, $options->keepAlive);
79-
});
68+
$promise = $this->socketConnector->connect($uri);
69+
$promise->then(function(Connection $stream) {
70+
$this->listenForPackets($stream);
71+
});
72+
$connection = $promise->then(function(Connection $stream) use ($options) {
73+
return $this->sendConnectPacket($stream, $options);
74+
});
75+
$connection->then(function(Connection $stream) use ($options) {
76+
return $this->keepAlive($stream, $options->keepAlive);
77+
});
78+
79+
return $connection;
8080
}
8181

82-
private function listenForPackets(Stream $stream)
82+
private function listenForPackets(Connection $stream)
8383
{
84-
$stream->on('data', function ($rawData) use ($stream) {
84+
$stream->on('data', function($rawData) use ($stream) {
8585
try {
8686
foreach (Factory::getNextPacket($this->version, $rawData) as $packet) {
8787
$stream->emit($packet::EVENT, [$packet]);
@@ -93,16 +93,16 @@ private function listenForPackets(Stream $stream)
9393
$stream->emit('INVALID', [$e]);
9494
}
9595
});
96-
97-
$deferred = new Deferred();
98-
$stream->on(ConnectionAck::EVENT, function($message) use ($stream, $deferred) {
99-
$deferred->resolve($stream);
100-
});
101-
102-
return $deferred->promise();
96+
//
97+
// $deferred = new Deferred();
98+
// $stream->on(ConnectionAck::EVENT, function($message) use ($stream, $deferred) {
99+
// $deferred->resolve($stream);
100+
// });
101+
//
102+
// return $deferred->promise();
103103
}
104104

105-
private function keepAlive(Stream $stream, $keepAlive)
105+
private function keepAlive(Connection $stream, $keepAlive)
106106
{
107107
if($keepAlive > 0) {
108108
$interval = $keepAlive / 2;
@@ -119,7 +119,7 @@ private function keepAlive(Stream $stream, $keepAlive)
119119
/**
120120
* @return \React\Promise\Promise
121121
*/
122-
public function connect(Stream $stream, ConnectionOptions $options) {
122+
public function sendConnectPacket(Connection $stream, ConnectionOptions $options) {
123123
$packet = new Connect(
124124
$this->version,
125125
$options->username,
@@ -136,16 +136,22 @@ public function connect(Stream $stream, ConnectionOptions $options) {
136136
echo MessageHelper::getReadableByRawString($message);
137137

138138
$deferred = new Deferred();
139-
if ($stream->write($message)) {
139+
$stream->on(ConnectionAck::EVENT, function($message) use ($stream, $deferred) {
140140
$deferred->resolve($stream);
141-
} else {
142-
$deferred->reject();
143-
}
141+
});
142+
143+
$stream->write($message);
144+
// $deferred = new Deferred();
145+
// if ($stream->write($message)) {
146+
// $deferred->resolve($stream);
147+
// } else {
148+
// $deferred->reject();
149+
// }
144150

145151
return $deferred->promise();
146152
}
147153

148-
private function sendPacketToStream(Stream $stream, ControlPacket $controlPacket)
154+
private function sendPacketToStream(Connection $stream, ControlPacket $controlPacket)
149155
{
150156
echo "send:\t\t" . get_class($controlPacket) . "\n";
151157
$message = $controlPacket->get();
@@ -154,12 +160,12 @@ private function sendPacketToStream(Stream $stream, ControlPacket $controlPacket
154160
}
155161

156162
/**
157-
* @param Stream $stream
163+
* @param Connection $stream
158164
* @param string $topic
159165
* @param int $qos
160166
* @return \React\Promise\Promise
161167
*/
162-
public function subscribe(Stream $stream, $topic, $qos = 0)
168+
public function subscribe(Connection $stream, $topic, $qos = 0)
163169
{
164170
$packet = new Subscribe($this->version);
165171
$packet->addSubscription($topic, $qos);
@@ -174,11 +180,11 @@ public function subscribe(Stream $stream, $topic, $qos = 0)
174180
}
175181

176182
/**
177-
* @param Stream $stream
183+
* @param Connection $stream
178184
* @param string $topic
179185
* @return \React\Promise\Promise
180186
*/
181-
public function unsubscribe(Stream $stream, $topic)
187+
public function unsubscribe(Connection $stream, $topic)
182188
{
183189
$packet = new Unsubscribe($this->version);
184190
$packet->removeSubscription($topic);
@@ -192,7 +198,7 @@ public function unsubscribe(Stream $stream, $topic)
192198
return $deferred->promise();
193199
}
194200

195-
public function disconnect(Stream $stream)
201+
public function disconnect(Connection $stream)
196202
{
197203
$packet = new Disconnect($this->version);
198204
$this->sendPacketToStream($stream, $packet);
@@ -204,7 +210,7 @@ public function disconnect(Stream $stream)
204210
/**
205211
* @return \React\Promise\Promise
206212
*/
207-
public function publish(Stream $stream, $topic, $message, $qos = 0, $dup = false, $retain = false)
213+
public function publish(Connection $stream, $topic, $message, $qos = 0, $dup = false, $retain = false)
208214
{
209215
$packet = new Publish($this->version);
210216
$packet->setTopic($topic);
@@ -227,7 +233,7 @@ public function publish(Stream $stream, $topic, $message, $qos = 0, $dup = false
227233
}
228234

229235
/**
230-
* @return LoopInterface
236+
* @return Loop
231237
*/
232238
public function getLoop()
233239
{

‎src/SecureConnector.php‎

Lines changed: 0 additions & 26 deletions
This file was deleted.

0 commit comments

Comments
 (0)