Skip to content

Commit 1c58aa8

Browse files
binsoulbinsoul
committed
Require at least PHP 7.2
1 parent b1bed6e commit 1c58aa8

File tree

4 files changed

+83
-81
lines changed

4 files changed

+83
-81
lines changed

‎composer.json

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,10 @@
1717
}
1818
],
1919
"require": {
20-
"php": "^7.0",
21-
"binsoul/net-mqtt": "^0.5",
20+
"php": "^7.2",
21+
"binsoul/net-mqtt": "^0.6",
22+
"evenement/evenement": "^3",
23+
"react/event-loop": "^1.1",
2224
"react/promise": "^2.7",
2325
"react/socket": "^1.3"
2426
},

‎src/ReactFlow.php

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ class ReactFlow implements Flow
3030
* @param Packet $packet
3131
* @param bool $isSilent
3232
*/
33-
public function __construct(Flow $decorated, Deferred $deferred, Packet $packet = null, bool $isSilent = false)
33+
public function __construct(Flow $decorated, Deferred $deferred, ?Packet $packet = null, bool $isSilent = false)
3434
{
3535
$this->decorated = $decorated;
3636
$this->deferred = $deferred;
@@ -43,7 +43,7 @@ public function getCode(): string
4343
return $this->decorated->getCode();
4444
}
4545

46-
public function start()
46+
public function start(): ?Packet
4747
{
4848
$this->packet = $this->decorated->start();
4949

@@ -55,7 +55,7 @@ public function accept(Packet $packet): bool
5555
return $this->decorated->accept($packet);
5656
}
5757

58-
public function next(Packet $packet)
58+
public function next(Packet $packet): ?Packet
5959
{
6060
$this->packet = $this->decorated->next($packet);
6161

@@ -97,7 +97,7 @@ public function getDeferred(): Deferred
9797
*
9898
* @return Packet|null
9999
*/
100-
public function getPacket()
100+
public function getPacket(): ?Packet
101101
{
102102
return $this->packet;
103103
}

‎src/ReactMqttClient.php

Lines changed: 40 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@
1919
use BinSoul\Net\Mqtt\StreamParser;
2020
use BinSoul\Net\Mqtt\Subscription;
2121
use Evenement\EventEmitter;
22+
use Exception;
23+
use LogicException;
2224
use React\EventLoop\LoopInterface;
2325
use React\EventLoop\TimerInterface;
2426
use React\Promise\CancellablePromiseInterface;
@@ -27,6 +29,7 @@
2729
use React\Promise\RejectedPromise;
2830
use React\Socket\ConnectorInterface;
2931
use React\Stream\DuplexStreamInterface;
32+
use RuntimeException;
3033

3134
/**
3235
* Connects to a MQTT broker and subscribes to topics or publishes messages.
@@ -99,9 +102,9 @@ class ReactMqttClient extends EventEmitter
99102
public function __construct(
100103
ConnectorInterface $connector,
101104
LoopInterface $loop,
102-
ClientIdentifierGenerator $identifierGenerator = null,
103-
FlowFactory $flowFactory = null,
104-
StreamParser $parser = null
105+
?ClientIdentifierGenerator $identifierGenerator = null,
106+
?FlowFactory $flowFactory = null,
107+
?StreamParser $parser = null
105108
) {
106109
$this->connector = $connector;
107110
$this->loop = $loop;
@@ -111,7 +114,7 @@ public function __construct(
111114
$this->parser = new StreamParser(new DefaultPacketFactory());
112115
}
113116

114-
$this->parser->onError(function (\Exception $e) {
117+
$this->parser->onError(function (Exception $e) {
115118
$this->emitWarning($e);
116119
});
117120

@@ -161,7 +164,7 @@ public function isConnected(): bool
161164
*
162165
* @return DuplexStreamInterface|null
163166
*/
164-
public function getStream()
167+
public function getStream(): ?DuplexStreamInterface
165168
{
166169
return $this->stream;
167170
}
@@ -179,7 +182,7 @@ public function getStream()
179182
public function connect(string $host, int $port = 1883, Connection $connection = null, int $timeout = 5): ExtendedPromiseInterface
180183
{
181184
if ($this->isConnected || $this->isConnecting) {
182-
return new RejectedPromise(new \LogicException('The client is already connected.'));
185+
return new RejectedPromise(new LogicException('The client is already connected.'));
183186
}
184187

185188
$this->isConnecting = true;
@@ -213,7 +216,7 @@ public function connect(string $host, int $port = 1883, Connection $connection =
213216
$this->emit('connect', [$connection, $this]);
214217
$deferred->resolve($result ?: $connection);
215218
})
216-
->otherwise(function (\Exception $e) use ($connection, $deferred) {
219+
->otherwise(function (Exception $e) use ($connection, $deferred) {
217220
$this->isConnecting = false;
218221

219222
$this->emitError($e);
@@ -226,7 +229,7 @@ public function connect(string $host, int $port = 1883, Connection $connection =
226229
$this->emit('close', [$connection, $this]);
227230
});
228231
})
229-
->otherwise(function (\Exception $e) use ($deferred) {
232+
->otherwise(function (Exception $e) use ($deferred) {
230233
$this->isConnecting = false;
231234

232235
$this->emitError($e);
@@ -246,7 +249,7 @@ public function connect(string $host, int $port = 1883, Connection $connection =
246249
public function disconnect(int $timeout = 5): ExtendedPromiseInterface
247250
{
248251
if (!$this->isConnected || $this->isDisconnecting) {
249-
return new RejectedPromise(new \LogicException('The client is not connected.'));
252+
return new RejectedPromise(new LogicException('The client is not connected.'));
250253
}
251254

252255
$this->isDisconnecting = true;
@@ -303,7 +306,7 @@ function () {
303306
public function subscribe(Subscription $subscription): ExtendedPromiseInterface
304307
{
305308
if (!$this->isConnected) {
306-
return new RejectedPromise(new \LogicException('The client is not connected.'));
309+
return new RejectedPromise(new LogicException('The client is not connected.'));
307310
}
308311

309312
return $this->startFlow($this->flowFactory->buildOutgoingSubscribeFlow([$subscription]));
@@ -319,7 +322,7 @@ public function subscribe(Subscription $subscription): ExtendedPromiseInterface
319322
public function unsubscribe(Subscription $subscription): ExtendedPromiseInterface
320323
{
321324
if (!$this->isConnected) {
322-
return new RejectedPromise(new \LogicException('The client is not connected.'));
325+
return new RejectedPromise(new LogicException('The client is not connected.'));
323326
}
324327

325328
return $this->startFlow($this->flowFactory->buildOutgoingUnsubscribeFlow([$subscription]));
@@ -335,7 +338,7 @@ public function unsubscribe(Subscription $subscription): ExtendedPromiseInterfac
335338
public function publish(Message $message): ExtendedPromiseInterface
336339
{
337340
if (!$this->isConnected) {
338-
return new RejectedPromise(new \LogicException('The client is not connected.'));
341+
return new RejectedPromise(new LogicException('The client is not connected.'));
339342
}
340343

341344
return $this->startFlow($this->flowFactory->buildOutgoingPublishFlow($message));
@@ -353,7 +356,7 @@ public function publish(Message $message): ExtendedPromiseInterface
353356
public function publishPeriodically(int $interval, Message $message, callable $generator): ExtendedPromiseInterface
354357
{
355358
if (!$this->isConnected) {
356-
return new RejectedPromise(new \LogicException('The client is not connected.'));
359+
return new RejectedPromise(new LogicException('The client is not connected.'));
357360
}
358361

359362
$deferred = new Deferred();
@@ -365,7 +368,7 @@ function () use ($message, $generator, $deferred) {
365368
static function ($value) use ($deferred) {
366369
$deferred->notify($value);
367370
},
368-
static function (\Exception $e) use ($deferred) {
371+
static function (Exception $e) use ($deferred) {
369372
$deferred->reject($e);
370373
}
371374
);
@@ -378,23 +381,23 @@ static function (\Exception $e) use ($deferred) {
378381
/**
379382
* Emits warnings.
380383
*
381-
* @param \Exception $e
384+
* @param Exception $e
382385
*
383386
* @return void
384387
*/
385-
private function emitWarning(\Exception $e)
388+
private function emitWarning(Exception $e): void
386389
{
387390
$this->emit('warning', [$e, $this]);
388391
}
389392

390393
/**
391394
* Emits errors.
392395
*
393-
* @param \Exception $e
396+
* @param Exception $e
394397
*
395398
* @return void
396399
*/
397-
private function emitError(\Exception $e)
400+
private function emitError(Exception $e): void
398401
{
399402
$this->emit('error', [$e, $this]);
400403
}
@@ -416,7 +419,7 @@ private function establishConnection(string $host, int $port, int $timeout): Ext
416419
$timer = $this->loop->addTimer(
417420
$timeout,
418421
static function () use ($deferred, $timeout, &$future) {
419-
$exception = new \RuntimeException(sprintf('Connection timed out after %d seconds.', $timeout));
422+
$exception = new RuntimeException(sprintf('Connection timed out after %d seconds.', $timeout));
420423
$deferred->reject($exception);
421424
if ($future instanceof CancellablePromiseInterface) {
422425
$future->cancel();
@@ -438,13 +441,13 @@ static function () use ($deferred, $timeout, &$future) {
438441
$this->handleClose();
439442
});
440443

441-
$stream->on('error', function (\Exception $e) {
444+
$stream->on('error', function (Exception $e) {
442445
$this->handleError($e);
443446
});
444447

445448
$deferred->resolve($stream);
446449
})
447-
->otherwise(static function (\Exception $e) use ($deferred) {
450+
->otherwise(static function (Exception $e) use ($deferred) {
448451
$deferred->reject($e);
449452
});
450453

@@ -466,7 +469,7 @@ private function registerClient(Connection $connection, int $timeout): ExtendedP
466469
$responseTimer = $this->loop->addTimer(
467470
$timeout,
468471
static function () use ($deferred, $timeout) {
469-
$exception = new \RuntimeException(sprintf('No response after %d seconds.', $timeout));
472+
$exception = new RuntimeException(sprintf('No response after %d seconds.', $timeout));
470473
$deferred->reject($exception);
471474
}
472475
);
@@ -483,7 +486,7 @@ function () {
483486
);
484487

485488
$deferred->resolve($result ?: $connection);
486-
})->otherwise(static function (\Exception $e) use ($deferred) {
489+
})->otherwise(static function (Exception $e) use ($deferred) {
487490
$deferred->reject($e);
488491
});
489492

@@ -497,7 +500,7 @@ function () {
497500
*
498501
* @return void
499502
*/
500-
private function handleReceive(string $data)
503+
private function handleReceive(string $data): void
501504
{
502505
if (!$this->isConnected && !$this->isConnecting) {
503506
return;
@@ -524,12 +527,12 @@ private function handleReceive(string $data)
524527
*
525528
* @return void
526529
*/
527-
private function handlePacket(Packet $packet)
530+
private function handlePacket(Packet $packet): void
528531
{
529532
switch ($packet->getPacketType()) {
530533
case Packet::TYPE_PUBLISH:
531534
if (!($packet instanceof PublishRequestPacket)) {
532-
throw new \RuntimeException(sprintf('Expected %s but got %s.', PublishRequestPacket::class, get_class($packet)));
535+
throw new RuntimeException(sprintf('Expected %s but got %s.', PublishRequestPacket::class, get_class($packet)));
533536
}
534537

535538
$message = new DefaultMessage(
@@ -564,13 +567,13 @@ private function handlePacket(Packet $packet)
564567

565568
if (!$flowFound) {
566569
$this->emitWarning(
567-
new \LogicException(sprintf('Received unexpected packet of type %d.', $packet->getPacketType()))
570+
new LogicException(sprintf('Received unexpected packet of type %d.', $packet->getPacketType()))
568571
);
569572
}
570573
break;
571574
default:
572575
$this->emitWarning(
573-
new \LogicException(sprintf('Cannot handle packet of type %d.', $packet->getPacketType()))
576+
new LogicException(sprintf('Cannot handle packet of type %d.', $packet->getPacketType()))
574577
);
575578
}
576579
}
@@ -580,7 +583,7 @@ private function handlePacket(Packet $packet)
580583
*
581584
* @return void
582585
*/
583-
private function handleSend()
586+
private function handleSend(): void
584587
{
585588
$flow = null;
586589
if ($this->writtenFlow !== null) {
@@ -609,7 +612,7 @@ private function handleSend()
609612
*
610613
* @return void
611614
*/
612-
private function handleClose()
615+
private function handleClose(): void
613616
{
614617
foreach ($this->timer as $timer) {
615618
$this->loop->cancelTimer($timer);
@@ -636,11 +639,11 @@ private function handleClose()
636639
/**
637640
* Handles errors of the stream.
638641
*
639-
* @param \Exception $e
642+
* @param Exception $e
640643
*
641644
* @return void
642645
*/
643-
private function handleError(\Exception $e)
646+
private function handleError(Exception $e): void
644647
{
645648
$this->emitError($e);
646649
}
@@ -657,7 +660,7 @@ private function startFlow(Flow $flow, bool $isSilent = false): ExtendedPromiseI
657660
{
658661
try {
659662
$packet = $flow->start();
660-
} catch (\Exception $e) {
663+
} catch (Exception $e) {
661664
$this->emitError($e);
662665

663666
return new RejectedPromise($e);
@@ -691,11 +694,11 @@ private function startFlow(Flow $flow, bool $isSilent = false): ExtendedPromiseI
691694
*
692695
* @return void
693696
*/
694-
private function continueFlow(ReactFlow $flow, Packet $packet)
697+
private function continueFlow(ReactFlow $flow, Packet $packet): void
695698
{
696699
try {
697700
$response = $flow->next($packet);
698-
} catch (\Exception $e) {
701+
} catch (Exception $e) {
699702
$this->emitError($e);
700703

701704
return;
@@ -723,7 +726,7 @@ private function continueFlow(ReactFlow $flow, Packet $packet)
723726
*
724727
* @return void
725728
*/
726-
private function finishFlow(ReactFlow $flow)
729+
private function finishFlow(ReactFlow $flow): void
727730
{
728731
if ($flow->isSuccess()) {
729732
if (!$flow->isSilent()) {
@@ -732,7 +735,7 @@ private function finishFlow(ReactFlow $flow)
732735

733736
$flow->getDeferred()->resolve($flow->getResult());
734737
} else {
735-
$result = new \RuntimeException($flow->getErrorMessage());
738+
$result = new RuntimeException($flow->getErrorMessage());
736739
$this->emitWarning($result);
737740

738741
$flow->getDeferred()->reject($result);

0 commit comments

Comments
 (0)