PHP code example of utopia-php / mqtt

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

utopia-php / mqtt example snippets

topia\Mqtt\Adapter;
use Utopia\Mqtt\Connection;
use Utopia\Mqtt\Handler;
use Utopia\Mqtt\Server;
use Utopia\Mqtt\Packet\Auth;
use Utopia\Mqtt\Packet\Connack;
use Utopia\Mqtt\Packet\Connect;
use Utopia\Mqtt\Packet\Disconnect;
use Utopia\Mqtt\Packet\Publish;
use Utopia\Mqtt\Packet\Puback;
use Utopia\Mqtt\Packet\Suback;
use Utopia\Mqtt\Packet\Subscribe;
use Utopia\Mqtt\Packet\Unsubscribe;
use Utopia\Mqtt\Packet\Unsuback;

final class Broker implements Handler
{
    public function onConnect(Connect $connect, Connection $connection): Connack|Auth
    {
        return Connack::accept();
    }

    public function onSubscribe(Subscribe $subscribe, Connection $connection): Suback
    {
        $suback = new Suback();
        foreach ($subscribe->filters() as $filter) {
            $suback->grant($filter->qos);
        }

        return $suback;
    }

    public function onPublish(Publish $publish, Connection $connection, iterable $subscribers): void
    {
        foreach ($subscribers as $subscriber) {
            $subscriber->publish($publish->topic, $publish->payload, qos: $publish->qos);
        }
    }

    public function onAuthenticate(Auth $auth, Connection $connection): Connack|Auth|Disconnect
    {
        return Auth::success($auth->method);
    }

    public function onUnsubscribe(Unsubscribe $unsubscribe, Connection $connection): Unsuback
    {
        return new Unsuback();
    }

    public function onPuback(Puback $puback, Connection $connection): void
    {
    }

    public function onDisconnect(?Disconnect $disconnect, Connection $connection): void
    {
    }
}

$adapter = new Adapter\Swoole([new Adapter\Swoole\Tcp('0.0.0.0', 1883)]);

$server = new Server($adapter, new Broker());
$server->start();

interface Handler
{
    public function onConnect(Connect $connect, Connection $connection): Connack|Auth;
    public function onAuthenticate(Auth $auth, Connection $connection): Connack|Auth|Disconnect;
    public function onSubscribe(Subscribe $subscribe, Connection $connection): Suback;
    public function onUnsubscribe(Unsubscribe $unsubscribe, Connection $connection): Unsuback;
    public function onPublish(Publish $publish, Connection $connection, iterable $subscribers): void;
    public function onPuback(Puback $puback, Connection $connection): void;
    public function onDisconnect(?Disconnect $disconnect, Connection $connection): void;
}

public function onConnect(Connect $connect, Connection $connection): Connack|Auth
{
    $identity = $this->authenticate($connect->username, $connect->password);
    if ($identity === null) {
        return Connack::refuse(Connack::NOT_AUTHORIZED); // encoded as the right CONNACK for v3 or v5
    }

    $connection->identity = $identity;      // opaque application state, stored and never read
    $connection->prefix   = $identity['tenant']; // the isolation key subscriptions are matched under

    return Connack::accept();
}

public function onSubscribe(Subscribe $subscribe, Connection $connection): Suback
{
    $suback = new Suback();
    foreach ($subscribe->filters() as $filter) {
        $suback->grant($this->acl->allows($connection->identity, $filter->topic) ? $filter->qos : Suback::DENIED);
    }

    return $suback;
}

public function onPublish(Publish $publish, Connection $connection, iterable $subscribers): void
{
    if (! $this->acl->canPublish($connection->identity, $publish->topic)) {
        return; // dropped
    }

    foreach ($subscribers as $subscriber) {
        $subscriber->publish($publish->topic, $publish->payload, qos: $publish->qos);
    }

    if ($publish->qos === 1) {
        $connection->puback($publish->packetId);
    }
}

final readonly class RateLimited implements Handler
{
    public function __construct(private Handler $inner) {}

    public function onConnect(Connect $connect, Connection $connection): Connack|Auth
    {
        return $this->isFlooding($connection) ? Connack::refuse(Connack::SERVER_BUSY) : $this->inner->onConnect($connect, $connection);
    }

    // delegate the rest to $this->inner …
}

$server = new Server($adapter, new RateLimited(new Broker()));

$connection->publish($topic, $payload, qos: 1, dup: false, sequence: 42);
$connection->puback($packetId);   // acknowledge an inbound QoS 1 PUBLISH
$connection->disconnect($reason);  // send a v5 DISCONNECT and close

use Utopia\Mqtt\Adapter;

$adapter = new Adapter\Swoole([
    new Adapter\Swoole\Tcp('0.0.0.0', 1883),
    new Adapter\Swoole\Tls('0.0.0.0', 8883, cert: '/etc/ssl/mqtt.crt', key: '/etc/ssl/mqtt.key'),
    new Adapter\Swoole\WebSocket('0.0.0.0', 8083),
], workers: 4);

$server->setTelemetry($telemetry);

bash
composer