MQTT

Apps can exchange messages with an MQTT server, for example to read Zigbee2MQTT devices directly or to talk to devices Home Assistant does not know. Stewart keeps one connection to the server, and apps ask for Mqtt in their constructor. This app copies temperatures from MQTT into Home Assistant, and publishes the living room thermostat’s state back:

apps/ThermostatBridge.php

<?php

declare(strict_types=1);

namespace App;

use Stewart\Contracts\App;
use Stewart\Contracts\Automation;
use Stewart\Contracts\HaContext;
use Stewart\Contracts\Mqtt\Mqtt;
use Stewart\Contracts\Mqtt\MqttMessage;
use Stewart\Contracts\Mqtt\MqttQos;
use Stewart\Contracts\State\StateChange;

#[Automation(id: 'thermostat-bridge')]
final readonly class ThermostatBridge implements App
{
    public function __construct(
        private HaContext $ha,
        private Mqtt $mqtt,
    ) {}

    public function initialize(): void
    {
        $this->mqtt->watchMessages('zigbee2mqtt/+/temperature')->subscribe(function (MqttMessage $message): void {
            $this->ha->callAction('input_number.set_value', ['value' => (float) $message->payload]);
        });

        $this->ha->watchStateChanges('climate.living_room')->subscribe(function (StateChange $change): void {
            $this->mqtt->publish('home/living_room/climate', ['state' => $change->to?->state], MqttQos::AtLeastOnce, retain: true);
        });
    }
}

Setting it up

Set the server in .env:

STEWART_MQTT__URL=mqtt://user:password@mosquitto:1883

Use mqtts://…:8883 for TLS. Without a URL, an app that asks for Mqtt does not start. Projects from the skeleton already have stewart-php/mqtt installed; other projects run composer require stewart-php/mqtt.

Optional settings in stewart.yaml:

mqtt:
  client_id: stewart-home        # default: stewart-<hostname>
  keepalive: 30s
  will: { topic: stewart/status, payload: offline, qos: 1, retain: true }

The will is a message the server publishes for Stewart when its connection dies.

Receiving

watchMessages('zigbee2mqtt/+/temperature') returns a stream of MqttMessages with topic, payload (text), qos and retain. decodeJsonPayload() reads a JSON payload.

Topic filters use MQTT’s wildcards: + matches one level, # the rest. A filter that starts with a wildcard does not match $SYS/… topics.

Publishing

publish($topic, $payload, $qos, retain: …) sends a message. A string payload is sent as it is, and an array as JSON.

QoS 0 and 1 are supported. A QoS 1 message published while the server is unreachable waits, up to mqtt.outbound_buffer messages, and is sent after the reconnect.

Watching the connection

make status shows the connection on its mqtt row: connected since 5m ago, 1 reconnect, last outage 3s, or reconnecting since 20s ago: Connection refused, followed by the QoS 1 messages still waiting and how many messages were dropped. A message is dropped when it is QoS 0 and the server is unreachable, when sending it fails, when a full mqtt.outbound_buffer pushes out the oldest, or when the server keeps dropping the connection over it. The log warns about the first drop of each kind and then every 100th, with the count so far.

The same figures are metrics: stewart_mqtt_connection_phase{phase}, stewart_mqtt_reconnects_total, stewart_mqtt_messages_dropped_total{reason} and stewart_mqtt_outbound_queued; the metrics reference lists them.

Testing

$harness->mqtt stands in for the server. pushMessage() sends a message to the app, and published lists what the app sent:

tests/ThermostatBridgeTest.php

<?php

declare(strict_types=1);

namespace App\Tests;

use App\ThermostatBridge;
use PHPUnit\Framework\TestCase;
use Stewart\Contracts\Mqtt\MqttMessage;
use Stewart\Contracts\Mqtt\MqttQos;
use Stewart\Testing\App\AppHarness;

final class ThermostatBridgeTest extends TestCase
{
    private AppHarness $harness;

    protected function setUp(): void
    {
        $this->harness = new AppHarness();
        $this->harness->ha->seedState('climate.living_room', 'off');
        $this->harness->startApp(ThermostatBridge::class);
    }

    public function testTemperatureFromMqttSetsTheInputNumber(): void
    {
        $this->harness->mqtt->pushMessage(new MqttMessage('zigbee2mqtt/kitchen/temperature', '21.5'));

        $call = $this->harness->ha->getLastCall();
        self::assertSame('input_number.set_value', $call->getActionName());
        self::assertSame(['value' => 21.5], $call->data);
    }

    public function testClimateStateIsPublishedRetained(): void
    {
        $this->harness->ha->pushState('climate.living_room', 'heat');

        $message = $this->harness->mqtt->published->getLast();
        self::assertNotNull($message);
        self::assertSame('home/living_room/climate', $message->topic);
        self::assertSame(['state' => 'heat'], $message->decodeJsonPayload());
        self::assertSame(MqttQos::AtLeastOnce, $message->qos);
        self::assertTrue($message->retain);
    }
}

Common problems

MQTT is not configured. Set STEWART_MQTT__URL in .env.

Nothing arrives. Check the topic with an MQTT client, such as mosquitto_sub -h host -t '#' -v. Topics are case-sensitive.