428 lines
19 KiB
PHP
428 lines
19 KiB
PHP
|
<?php
|
||
|
|
|
||
|
|
/** @noinspection PhpDocSignatureInspection */
|
||
|
|
/** @noinspection PhpUnhandledExceptionInspection */
|
||
|
|
|
||
|
|
declare(strict_types=1);
|
||
|
|
|
||
|
|
namespace Tests\Feature;
|
||
|
|
|
||
|
|
use PhpMqtt\Client\ConnectionSettings;
|
||
|
|
use PhpMqtt\Client\MqttClient;
|
||
|
|
use Tests\TestCase;
|
||
|
|
|
||
|
|
/**
|
||
|
|
* Tests that publishing messages and subscribing to topics using an MQTT broker works.
|
||
|
|
*
|
||
|
|
* @package Tests\Feature
|
||
|
|
*/
|
||
|
|
class PublishSubscribeTest extends TestCase
|
||
|
|
{
|
||
|
|
public function publishSubscribeData(): array
|
||
|
|
{
|
||
|
|
$data = [
|
||
|
|
[false, 'foo/bar/baz', 'foo/bar/baz', 'hello world', []],
|
||
|
|
[false, 'foo/bar/+', 'foo/bar/baz', 'hello world', ['baz']],
|
||
|
|
[false, 'foo/+/baz', 'foo/bar/baz', 'hello world', ['bar']],
|
||
|
|
[false, 'foo/#', 'foo/bar/baz', 'hello world', ['bar/baz']],
|
||
|
|
[false, 'foo/+/bar/#', 'foo/my/bar/baz', 'hello world', ['my', 'baz']],
|
||
|
|
[false, 'foo/+/bar/#', 'foo/my/bar/baz/blub', 'hello world', ['my', 'baz/blub']],
|
||
|
|
[false, 'foo/bar/baz', 'foo/bar/baz', random_bytes(2 * 1024 * 1024), []], // 2MB message
|
||
|
|
[true, 'foo/bar/baz', 'foo/bar/baz', 'hello world', []],
|
||
|
|
[true, 'foo/bar/+', 'foo/bar/baz', 'hello world', ['baz']],
|
||
|
|
[true, 'foo/+/baz', 'foo/bar/baz', 'hello world', ['bar']],
|
||
|
|
[true, 'foo/#', 'foo/bar/baz', 'hello world', ['bar/baz']],
|
||
|
|
[true, 'foo/+/bar/#', 'foo/my/bar/baz', 'hello world', ['my', 'baz']],
|
||
|
|
[true, 'foo/+/bar/#', 'foo/my/bar/baz/blub', 'hello world', ['my', 'baz/blub']],
|
||
|
|
[true, 'foo/bar/baz', 'foo/bar/baz', random_bytes(2 * 1024 * 1024), []], // 2MB message
|
||
|
|
];
|
||
|
|
|
||
|
|
// Because our tests are run against a real MQTT broker and some messages are retained,
|
||
|
|
// we need to prevent false-positives by giving each test case its own 'test space' using a random prefix.
|
||
|
|
for ($i = 0; $i < count($data); $i++) {
|
||
|
|
$prefix = 'test/' . uniqid('', true) . '/';
|
||
|
|
$data[$i][1] = $prefix . $data[$i][1];
|
||
|
|
$data[$i][2] = $prefix . $data[$i][2];
|
||
|
|
}
|
||
|
|
|
||
|
|
return $data;
|
||
|
|
}
|
||
|
|
|
||
|
|
/**
|
||
|
|
* @dataProvider publishSubscribeData
|
||
|
|
*/
|
||
|
|
public function test_publishing_and_subscribing_using_quality_of_service_0_works_as_intended(
|
||
|
|
bool $useBlockingSocket,
|
||
|
|
string $subscriptionTopicFilter,
|
||
|
|
string $publishTopic,
|
||
|
|
string $publishMessage,
|
||
|
|
array $matchedTopicWildcards
|
||
|
|
): void
|
||
|
|
{
|
||
|
|
// We connect and subscribe to a topic using the first client.
|
||
|
|
$connectionSettings = (new ConnectionSettings())
|
||
|
|
->useBlockingSocket($useBlockingSocket);
|
||
|
|
|
||
|
|
$subscriber = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'subscriber');
|
||
|
|
$subscriber->connect($connectionSettings, true);
|
||
|
|
|
||
|
|
$subscriber->subscribe(
|
||
|
|
$subscriptionTopicFilter,
|
||
|
|
function (string $topic, string $message, bool $retained, array $wildcards) use ($subscriber, $publishTopic, $publishMessage, $matchedTopicWildcards) {
|
||
|
|
// By asserting something here, we will avoid a no-assertions-in-test warning, making the test pass.
|
||
|
|
$this->assertEquals($publishTopic, $topic);
|
||
|
|
$this->assertEquals($publishMessage, $message);
|
||
|
|
$this->assertFalse($retained);
|
||
|
|
$this->assertEquals($matchedTopicWildcards, $wildcards);
|
||
|
|
|
||
|
|
$subscriber->interrupt(); // This allows us to exit the test as soon as possible.
|
||
|
|
},
|
||
|
|
0
|
||
|
|
);
|
||
|
|
|
||
|
|
// We publish a message from a second client on the same topic.
|
||
|
|
$publisher = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'publisher');
|
||
|
|
$publisher->connect(null, true);
|
||
|
|
|
||
|
|
$publisher->publish($publishTopic, $publishMessage, 0, false);
|
||
|
|
|
||
|
|
// Then we loop on the subscriber to (hopefully) receive the published message.
|
||
|
|
$subscriber->loop(true);
|
||
|
|
|
||
|
|
// Finally, we disconnect for a graceful shutdown on the broker side.
|
||
|
|
$publisher->disconnect();
|
||
|
|
$subscriber->disconnect();
|
||
|
|
}
|
||
|
|
|
||
|
|
/**
|
||
|
|
* @dataProvider publishSubscribeData
|
||
|
|
*/
|
||
|
|
public function test_publishing_and_subscribing_using_quality_of_service_0_with_message_retention_works_as_intended(
|
||
|
|
bool $useBlockingSocket,
|
||
|
|
string $subscriptionTopicFilter,
|
||
|
|
string $publishTopic,
|
||
|
|
string $publishMessage,
|
||
|
|
array $matchedTopicWildcards
|
||
|
|
): void
|
||
|
|
{
|
||
|
|
// We publish a message from the first client, which disconnects before the other client even subscribes.
|
||
|
|
$publisher = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'publisher');
|
||
|
|
$publisher->connect(null, true);
|
||
|
|
|
||
|
|
$publisher->publish($publishTopic, $publishMessage, 0, true);
|
||
|
|
|
||
|
|
$publisher->disconnect();
|
||
|
|
|
||
|
|
// Because we need to make sure the message reached the broker, we delay the execution for a short period (100ms) intentionally.
|
||
|
|
// With higher QoS, this is replaced by awaiting delivery of the message.
|
||
|
|
usleep(100_000);
|
||
|
|
|
||
|
|
// We connect and subscribe to a topic using the second client.
|
||
|
|
$connectionSettings = (new ConnectionSettings())
|
||
|
|
->useBlockingSocket($useBlockingSocket);
|
||
|
|
|
||
|
|
$subscriber = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'subscriber');
|
||
|
|
$subscriber->connect($connectionSettings, true);
|
||
|
|
|
||
|
|
$subscriber->subscribe(
|
||
|
|
$subscriptionTopicFilter,
|
||
|
|
function (string $topic, string $message, bool $retained, array $wildcards) use ($subscriber, $publishTopic, $publishMessage, $matchedTopicWildcards) {
|
||
|
|
// By asserting something here, we will avoid a no-assertions-in-test warning, making the test pass.
|
||
|
|
$this->assertEquals($publishTopic, $topic);
|
||
|
|
$this->assertEquals($publishMessage, $message);
|
||
|
|
$this->assertTrue($retained);
|
||
|
|
$this->assertEquals($matchedTopicWildcards, $wildcards);
|
||
|
|
|
||
|
|
$subscriber->interrupt(); // This allows us to exit the test as soon as possible.
|
||
|
|
},
|
||
|
|
0
|
||
|
|
);
|
||
|
|
|
||
|
|
// Then we loop on the subscriber to (hopefully) receive the published message.
|
||
|
|
$subscriber->loop(true);
|
||
|
|
|
||
|
|
// Finally, we disconnect for a graceful shutdown on the broker side.
|
||
|
|
$subscriber->disconnect();
|
||
|
|
}
|
||
|
|
|
||
|
|
/**
|
||
|
|
* @dataProvider publishSubscribeData
|
||
|
|
*/
|
||
|
|
public function test_publishing_and_subscribing_using_quality_of_service_1_works_as_intended(
|
||
|
|
bool $useBlockingSocket,
|
||
|
|
string $subscriptionTopicFilter,
|
||
|
|
string $publishTopic,
|
||
|
|
string $publishMessage,
|
||
|
|
array $matchedTopicWildcards
|
||
|
|
): void
|
||
|
|
{
|
||
|
|
// We connect and subscribe to a topic using the first client.
|
||
|
|
$connectionSettings = (new ConnectionSettings())
|
||
|
|
->useBlockingSocket($useBlockingSocket);
|
||
|
|
|
||
|
|
$subscriber = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'subscriber');
|
||
|
|
$subscriber->connect($connectionSettings, true);
|
||
|
|
|
||
|
|
$subscriber->subscribe(
|
||
|
|
$subscriptionTopicFilter,
|
||
|
|
function (string $topic, string $message, bool $retained, array $wildcards) use ($subscriber, $publishTopic, $publishMessage, $matchedTopicWildcards) {
|
||
|
|
// By asserting something here, we will avoid a no-assertions-in-test warning, making the test pass.
|
||
|
|
$this->assertEquals($publishTopic, $topic);
|
||
|
|
$this->assertEquals($publishMessage, $message);
|
||
|
|
$this->assertFalse($retained);
|
||
|
|
$this->assertEquals($matchedTopicWildcards, $wildcards);
|
||
|
|
|
||
|
|
$subscriber->interrupt(); // This allows us to exit the test as soon as possible.
|
||
|
|
},
|
||
|
|
1
|
||
|
|
);
|
||
|
|
|
||
|
|
// We publish a message from a second client on the same topic.
|
||
|
|
$publisher = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'publisher');
|
||
|
|
$publisher->connect(null, true);
|
||
|
|
|
||
|
|
$publisher->publish($publishTopic, $publishMessage, 1, false);
|
||
|
|
|
||
|
|
// Then we loop on the subscriber to (hopefully) receive the published message.
|
||
|
|
$subscriber->loop(true);
|
||
|
|
|
||
|
|
// Finally, we disconnect for a graceful shutdown on the broker side.
|
||
|
|
$publisher->disconnect();
|
||
|
|
$subscriber->disconnect();
|
||
|
|
}
|
||
|
|
|
||
|
|
/**
|
||
|
|
* @dataProvider publishSubscribeData
|
||
|
|
*/
|
||
|
|
public function test_publishing_and_subscribing_using_quality_of_service_1_with_message_retention_works_as_intended(
|
||
|
|
bool $useBlockingSocket,
|
||
|
|
string $subscriptionTopicFilter,
|
||
|
|
string $publishTopic,
|
||
|
|
string $publishMessage,
|
||
|
|
array $matchedTopicWildcards
|
||
|
|
): void
|
||
|
|
{
|
||
|
|
// We publish a message from the first client, which disconnects before the other client even subscribes.
|
||
|
|
$publisher = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'publisher');
|
||
|
|
$publisher->connect(null, true);
|
||
|
|
|
||
|
|
$publisher->publish($publishTopic, $publishMessage, 1, true);
|
||
|
|
$publisher->loop(true, true);
|
||
|
|
|
||
|
|
$publisher->disconnect();
|
||
|
|
|
||
|
|
// We connect and subscribe to a topic using the second client.
|
||
|
|
$connectionSettings = (new ConnectionSettings())
|
||
|
|
->useBlockingSocket($useBlockingSocket);
|
||
|
|
|
||
|
|
$subscriber = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'subscriber');
|
||
|
|
$subscriber->connect($connectionSettings, true);
|
||
|
|
|
||
|
|
$subscriber->subscribe(
|
||
|
|
$subscriptionTopicFilter,
|
||
|
|
function (string $topic, string $message, bool $retained, array $wildcards) use ($subscriber, $publishTopic, $publishMessage, $matchedTopicWildcards) {
|
||
|
|
// By asserting something here, we will avoid a no-assertions-in-test warning, making the test pass.
|
||
|
|
$this->assertEquals($publishTopic, $topic);
|
||
|
|
$this->assertEquals($publishMessage, $message);
|
||
|
|
$this->assertTrue($retained);
|
||
|
|
$this->assertEquals($matchedTopicWildcards, $wildcards);
|
||
|
|
|
||
|
|
$subscriber->interrupt(); // This allows us to exit the test as soon as possible.
|
||
|
|
},
|
||
|
|
1
|
||
|
|
);
|
||
|
|
|
||
|
|
// Then we loop on the subscriber to (hopefully) receive the published message.
|
||
|
|
$subscriber->loop(true);
|
||
|
|
|
||
|
|
// Finally, we disconnect for a graceful shutdown on the broker side.
|
||
|
|
$subscriber->disconnect();
|
||
|
|
}
|
||
|
|
|
||
|
|
/**
|
||
|
|
* @dataProvider publishSubscribeData
|
||
|
|
*/
|
||
|
|
public function test_publishing_and_subscribing_using_quality_of_service_2_works_as_intended(
|
||
|
|
bool $useBlockingSocket,
|
||
|
|
string $subscriptionTopicFilter,
|
||
|
|
string $publishTopic,
|
||
|
|
string $publishMessage,
|
||
|
|
array $matchedTopicWildcards
|
||
|
|
): void
|
||
|
|
{
|
||
|
|
// We connect and subscribe to a topic using the first client.
|
||
|
|
$connectionSettings = (new ConnectionSettings())
|
||
|
|
->useBlockingSocket($useBlockingSocket);
|
||
|
|
|
||
|
|
$subscriber = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'subscriber');
|
||
|
|
$subscriber->connect($connectionSettings, true);
|
||
|
|
|
||
|
|
$subscription = function (string $topic, string $message, bool $retained, array $wildcards) use ($subscriber, $subscriptionTopicFilter, $publishTopic, $publishMessage, $matchedTopicWildcards) {
|
||
|
|
// By asserting something here, we will avoid a no-assertions-in-test warning, making the test pass.
|
||
|
|
$this->assertEquals($publishTopic, $topic);
|
||
|
|
$this->assertEquals($publishMessage, $message);
|
||
|
|
$this->assertFalse($retained);
|
||
|
|
$this->assertEquals($matchedTopicWildcards, $wildcards);
|
||
|
|
|
||
|
|
$subscriber->unsubscribe($subscriptionTopicFilter);
|
||
|
|
$subscriber->interrupt(); // This allows us to exit the test as soon as possible.
|
||
|
|
};
|
||
|
|
|
||
|
|
$subscriber->subscribe($subscriptionTopicFilter, $subscription, 2);
|
||
|
|
|
||
|
|
// We publish a message from a second client on the same topic. The loop is called until all QoS 2 handshakes are done.
|
||
|
|
$publisher = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'publisher');
|
||
|
|
$publisher->connect(null, true);
|
||
|
|
|
||
|
|
$publisher->publish($publishTopic, $publishMessage, 2, false);
|
||
|
|
$publisher->loop(true, true);
|
||
|
|
|
||
|
|
// Then we loop on the subscriber to (hopefully) receive the published message until the receive handshake is done.
|
||
|
|
$subscriber->loop(true, true);
|
||
|
|
|
||
|
|
// Finally, we disconnect for a graceful shutdown on the broker side.
|
||
|
|
$publisher->disconnect();
|
||
|
|
$subscriber->disconnect();
|
||
|
|
}
|
||
|
|
|
||
|
|
/**
|
||
|
|
* @dataProvider publishSubscribeData
|
||
|
|
*/
|
||
|
|
public function test_publishing_and_subscribing_using_quality_of_service_2_with_message_retention_works_as_intended(
|
||
|
|
bool $useBlockingSocket,
|
||
|
|
string $subscriptionTopicFilter,
|
||
|
|
string $publishTopic,
|
||
|
|
string $publishMessage,
|
||
|
|
array $matchedTopicWildcards
|
||
|
|
): void
|
||
|
|
{
|
||
|
|
// We publish a message from the first client. The loop is called until all QoS 2 handshakes are done.
|
||
|
|
$publisher = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'publisher');
|
||
|
|
$publisher->connect(null, true);
|
||
|
|
|
||
|
|
$publisher->publish($publishTopic, $publishMessage, 2, true);
|
||
|
|
$publisher->loop(true, true);
|
||
|
|
|
||
|
|
$publisher->disconnect();
|
||
|
|
|
||
|
|
// We connect and subscribe to a topic using the second client.
|
||
|
|
$connectionSettings = (new ConnectionSettings())
|
||
|
|
->useBlockingSocket($useBlockingSocket);
|
||
|
|
|
||
|
|
$subscriber = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'subscriber');
|
||
|
|
$subscriber->connect($connectionSettings, true);
|
||
|
|
|
||
|
|
$subscription = function (string $topic, string $message, bool $retained, array $wildcards) use ($subscriber, $subscriptionTopicFilter, $publishTopic, $publishMessage, $matchedTopicWildcards) {
|
||
|
|
// By asserting something here, we will avoid a no-assertions-in-test warning, making the test pass.
|
||
|
|
$this->assertEquals($publishTopic, $topic);
|
||
|
|
$this->assertEquals($publishMessage, $message);
|
||
|
|
$this->assertTrue($retained);
|
||
|
|
$this->assertEquals($matchedTopicWildcards, $wildcards);
|
||
|
|
|
||
|
|
$subscriber->unsubscribe($subscriptionTopicFilter);
|
||
|
|
$subscriber->interrupt(); // This allows us to exit the test as soon as possible.
|
||
|
|
};
|
||
|
|
|
||
|
|
$subscriber->subscribe($subscriptionTopicFilter, $subscription, 2);
|
||
|
|
|
||
|
|
// Then we loop on the subscriber to (hopefully) receive the published message until the receive handshake is done.
|
||
|
|
$subscriber->loop(true, true);
|
||
|
|
|
||
|
|
// Finally, we disconnect for a graceful shutdown on the broker side.
|
||
|
|
$subscriber->disconnect();
|
||
|
|
}
|
||
|
|
|
||
|
|
public function test_unsubscribe_stops_receiving_messages_on_topic(): void
|
||
|
|
{
|
||
|
|
// We connect and subscribe to a topic using the first client.
|
||
|
|
$subscriber = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'subscriber');
|
||
|
|
$subscriber->connect(null, true);
|
||
|
|
|
||
|
|
$subscribedTopic = 'test/foo/bar/baz';
|
||
|
|
$receivedMessageCount = 0;
|
||
|
|
$subscriber->subscribe(
|
||
|
|
$subscribedTopic,
|
||
|
|
function (string $topic, string $message, bool $retained) use ($subscriber, $subscribedTopic, &$receivedMessageCount) {
|
||
|
|
$receivedMessageCount++;
|
||
|
|
|
||
|
|
// By asserting something here, we will avoid a no-assertions-in-test warning, making the test pass.
|
||
|
|
$this->assertEquals('test/foo/bar/baz', $topic);
|
||
|
|
$this->assertEquals('hello world', $message);
|
||
|
|
$this->assertFalse($retained);
|
||
|
|
|
||
|
|
$subscriber->unsubscribe($subscribedTopic);
|
||
|
|
},
|
||
|
|
0
|
||
|
|
);
|
||
|
|
|
||
|
|
// We publish a message from a second client on the same topic.
|
||
|
|
$publisher = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'publisher');
|
||
|
|
$publisher->connect(null, true);
|
||
|
|
|
||
|
|
$publisher->publish($subscribedTopic, 'hello world', 0, false);
|
||
|
|
|
||
|
|
// Then we loop on the subscriber to (hopefully) receive the published message.
|
||
|
|
$subscriber->loop(true, true);
|
||
|
|
|
||
|
|
$this->assertSame(1, $receivedMessageCount);
|
||
|
|
|
||
|
|
$publisher->publish($subscribedTopic, 'hello world #2', 0, false);
|
||
|
|
$subscriber->loop(true, true);
|
||
|
|
|
||
|
|
// Ensure no second message has been received since we are not subscribed anymore.
|
||
|
|
$this->assertSame(1, $receivedMessageCount);
|
||
|
|
|
||
|
|
// Finally, we disconnect for a graceful shutdown on the broker side.
|
||
|
|
$publisher->disconnect();
|
||
|
|
$subscriber->disconnect();
|
||
|
|
}
|
||
|
|
|
||
|
|
public function test_shared_subscriptions_using_quality_of_service_0_work_as_intended(): void
|
||
|
|
{
|
||
|
|
$subscriptionTopicFilter = '$share/test-shared-subscriptions/foo/+';
|
||
|
|
$publishTopic = 'foo/bar';
|
||
|
|
|
||
|
|
// We connect and subscribe to a topic using the first client with a shared subscription.
|
||
|
|
$subscriber1 = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'subscriber1');
|
||
|
|
$subscriber1->connect(null, true);
|
||
|
|
|
||
|
|
$subscriber1->subscribe($subscriptionTopicFilter, function (string $topic, string $message, bool $retained) use ($subscriber1, $publishTopic) {
|
||
|
|
// By asserting something here, we will avoid a no-assertions-in-test warning, making the test pass.
|
||
|
|
$this->assertEquals($publishTopic, $topic);
|
||
|
|
$this->assertEquals('hello world #1', $message);
|
||
|
|
$this->assertFalse($retained);
|
||
|
|
|
||
|
|
$subscriber1->interrupt(); // This allows us to exit the test as soon as possible.
|
||
|
|
}, 0);
|
||
|
|
|
||
|
|
// We connect and subscribe to a topic using the second client with a shared subscription.
|
||
|
|
$subscriber2 = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'subscriber2');
|
||
|
|
$subscriber2->connect(null, true);
|
||
|
|
|
||
|
|
$subscriber2->subscribe($subscriptionTopicFilter, function (string $topic, string $message, bool $retained) use ($subscriber2, $publishTopic) {
|
||
|
|
// By asserting something here, we will avoid a no-assertions-in-test warning, making the test pass.
|
||
|
|
$this->assertEquals($publishTopic, $topic);
|
||
|
|
$this->assertEquals('hello world #2', $message);
|
||
|
|
$this->assertFalse($retained);
|
||
|
|
|
||
|
|
$subscriber2->interrupt(); // This allows us to exit the test as soon as possible.
|
||
|
|
}, 0);
|
||
|
|
|
||
|
|
// We publish a message from a second client on the same topic.
|
||
|
|
$publisher = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'publisher');
|
||
|
|
$publisher->connect(null, true);
|
||
|
|
|
||
|
|
$publisher->publish($publishTopic, 'hello world #1', 0, false);
|
||
|
|
$publisher->publish($publishTopic, 'hello world #2', 0, false);
|
||
|
|
|
||
|
|
// Then we loop on the subscribers to (hopefully) receive the published messages.
|
||
|
|
$subscriber1->loop(true);
|
||
|
|
$subscriber2->loop(true);
|
||
|
|
|
||
|
|
// Finally, we disconnect for a graceful shutdown on the broker side.
|
||
|
|
$publisher->disconnect();
|
||
|
|
$subscriber1->disconnect();
|
||
|
|
$subscriber2->disconnect();
|
||
|
|
}
|
||
|
|
}
|