salmanzafar/laravel-mqtt
A Laravel MQTT client for publishing and subscribing to MQTT brokers with configurable connections, QoS, retained messages, TLS, wildcard topics, and a simple facade with Laravel-friendly configuration.
Install and publish config
composer require salmanzafar/laravel-mqtt
php artisan vendor:publish --provider="SalmanZafar\MQTT\MQTTServiceProvider"
Package discovery registers the service provider and a MQTT facade. If the publish tag isn't documented in your version, use the --provider form above; if nothing lands, check vendor/salmanzafar/laravel-mqtt/src for the tag name — the package is small enough to read end to end in five minutes.
Configure .env
MQTT_HOST=broker.example.com
MQTT_PORT=1883
MQTT_AUTH_USERNAME=app
MQTT_AUTH_PASSWORD=secret
MQTT_CLIENT_ID=laravel-${APP_NAME}
The published config/mqtt.php supports multiple named connections under connections, with a default_connection. The keys are essentially a pass-through to php-mqtt/client (host, port, username, password, client_id, clean_session, keep_alive_interval, connect_timeout, socket_timeout, tls, lwt, protocol). Treat that config file as the source of truth and read it before reading the README.
As of v3.3.0 there is also a queue/connection config surface for the new async publishing path — check whether config/mqtt.php gained a queue block (connection, queue name, whether it's enabled) before you assume it uses your default queue connection.
Read these three things first, in order
config/mqtt.php — what knobs you actually have, including any new queue settings.src/MQTT.php — the whole public API is roughly connect(), publish(), subscribe(), disconnect(), and now queue().php-mqtt/client docs — QoS, retained messages, clean sessions, and reconnect behaviour all come from there; this package is a thin container/config wrapper, not a reimplementation.First use case: publish state on a model event
use SalmanZafar\MQTT\MQTT;
class DeviceStatusPublisher
{
public function __construct(private MQTT $mqtt) {}
public function publish(Device $device): void
{
$payload = json_encode(['online' => $device->is_online, 'ts' => now()->toIso8601String()]);
// v3.3.0+: hand it to the queue instead of blocking the request
$this->mqtt->queue("devices/{$device->uuid}/status", $payload, qos: 1, retain: true);
// Or synchronous, if you genuinely need the publish to have completed:
// $this->mqtt->connect();
// $this->mqtt->publish("devices/{$device->uuid}/status", $payload, qos: 1, retain: true);
// $this->mqtt->disconnect();
}
}
Hook it to Device::saved() or a domain event. If that works, the rest of the package is variations on publish(), queue() and subscribe(). Verify the exact queue() signature and argument order against src/MQTT.php — it's brand new and may not mirror publish() one-for-one.
Wrap it behind your own interface. Don't inject SalmanZafar\MQTT\MQTT into twenty classes. Define App\Mqtt\Publisher with publish(string $topic, string $payload, int $qos = 1, bool $retain = false): void and, now, enqueue(...) for the async path, implemented with the package and bound in a service provider. This is the single highest-value decision: it gives you a fake for tests, a place to add logging/metrics, and an escape hatch to swap in php-mqtt/laravel-client or raw php-mqtt/client later without touching call sites. It also means the arrival of queue() in v3.3.0 is a one-file change instead of a refactor.
Prefer queue() over hand-rolled publish jobs. Before v3.3.0 the standard advice was "write a PublishMqttMessage job yourself". That's now the package's job: Mqtt::queue() serialises the connect/publish/disconnect into a queued job, which means the broker round-trip leaves the request cycle and the connection is established inside the worker rather than reused across jobs. That last part is the real win — it structurally avoids the stale-socket class of bug described below.
Two caveats before you delete your own job class:
publish() that can't reach the broker throws at the call site; queue() returns successfully and the failure surfaces in failed_jobs. Wire up failed-job monitoring or you've traded a crash for a silent hole.$tries / $backoff / retryUntil. The package's job has whatever defaults its author chose. Read the job class in src/ and, if the retry policy doesn't suit you, call ->onConnection()/->onQueue() on the returned job — or keep your own wrapper job that calls publish() synchronously inside handle().Keep the payload self-describing either way: message IDs and timestamps in the JSON, dedupe on the consumer. QoS 1 is at-least-once on the wire, and queue retries will duplicate on top of that.
Long-running subscriber as an Artisan command. There is no mqtt:work command in this package; you write the loop. Pattern:
class MqttSubscribeCommand extends Command
{
protected $signature = 'mqtt:subscribe {topic} {--connection=default}';
public function handle(MQTT $mqtt): int
{
$mqtt->connect($this->option('connection'));
$mqtt->subscribe($this->argument('topic'), function ($topic, $message) {
// dispatch a job, don't do work inline
ProcessMqttMessage::dispatch($topic, $message);
}, qos: 1);
return self::SUCCESS;
}
}
Use php-mqtt's loop() / loopOnce() with a signal handler so SIGTERM from Supervisor exits cleanly, and run it under Supervisor or php artisan queue:work-style process management, never in a web route.
Bridge MQTT to Laravel events and back. The cleanest integration is: inbound subscriber → dispatch a typed Laravel event or job; outbound → listen to a Laravel event → queue(). Keep the MQTT topic string in one place (a Topics class or config map) so inbound and outbound can't drift.
Retained messages for last-known-state, not for commands. retain: true on devices/{uuid}/status means a new subscriber immediately gets current state without waiting for the next heartbeat — exactly right. Retained command topics are a bug factory: a device that reconnects will replay a stale "open valve" command. Publish commands with retain: false. This applies identically to the queued path — retention is a broker flag, not a transport detail, so queue(..., retain: true) will happily plant a retained payload too.
Multi-broker / multi-tenant. Use named connections and pass the connection name per call rather than mutating the default. If you're multi-tenant with per-tenant brokers, resolve the connection name from the tenant context and be explicit at every call site — and check that the queued path carries the connection name through to the worker, since the job runs outside the request where the tenant context was resolved. Serialise the connection name into the job payload if you have to.
Testing. Facade-level shouldReceive is brittle for chained calls (connect()->publish()->disconnect()). With your own Publisher interface, bind a fake that records messages and assert against it. For the queued path, Queue::fake() plus Queue::assertPushed(...) verifies a job was enqueued, and Bus::fake()/Queue::fake() around a job test verifies the publish payload — but nothing short of a real broker verifies the message actually left. If you must test the real thing, point it at a local Mosquitto/MQTT Explorer instance in CI; there is no in-memory transport.
Client ID collisions will silently kill your app. Brokers enforce unique client IDs and disconnect the older session. A static MQTT_CLIENT_ID is a problem for one process and a disaster with Horizon (N workers, N duplicate IDs). Suffix it: config('mqtt.client_id').'-'.gethostname().'-'.getmypid(). This matters more now: queue() multiplies the number of workers that connect, so duplicate client IDs go from "rare" to "guaranteed" as soon as you scale the queue.
Clean session + QoS 0 = messages that vanish. If clean_session is true, every reconnect drops subscriptions and the broker queues nothing. For durable subscribers set clean_session=false and pick a stable client ID, and use QoS 1 on both publish and subscribe or you lose the guarantee at one hop anyway. Note the tension with the previous tip: a durable subscriber needs a stable ID, a queue worker needs a unique one. Use stable IDs for your long-running subscriber command and process-suffixed IDs for workers.
A reused client goes stale. In queue workers and Octane, a client that sat idle past keep_alive_interval has a dead socket. php-mqtt's loop has auto-reconnect, but a bare publish() on a stale client throws ConnectionNotAvailableException. The queued path sidesteps this by connecting inside the job, but only if the package's job doesn't resolve a cached singleton — verify in src/. For your own synchronous publishes: try { … } catch (Throwable) { $this->mqtt->disconnect(); $this->mqtt->connect(); retry publish once; }, or better: connect → publish → disconnect inside one job so each job starts clean. Also reset the binding in Octane's RequestReceived/Tick listeners if the container holds a singleton.
Transaction ordering still bites the queued path. Publishing inside DB::transaction() sends the message before the commit; a rollback leaves reality and the broker disagreeing. queue() makes this worse, not better — dispatching to a queue inside a transaction is exactly the classic race, and with QUEUE_CONNECTION=sync or a fast worker the message goes out before the commit just like before. Use DB::afterCommit(fn () => Mqtt::queue(...)) or dispatch outside the transaction.
Queued publishing implies a worker. After switching to queue(), an app with no php artisan queue:work running publishes nothing at all, and nothing errors loudly — the jobs just pile up in jobs. Deploy the worker alongside the code change, and add an alert on queue depth for the MQTT queue specifically.
Timeouts are measured in "why is my web request hanging". connect_timeout defaults are long. For anything still synchronous, set connect_timeout to 2–3s and socket_timeout to similarly tight values, and fail fast. On the queued path, generous timeouts are fine and arguably preferable — the request is no longer waiting.
TLS. Port 8883 plus a tls block in config; self-signed brokers need verify_peer => false, which belongs only in local. Most "connection reset" reports on forum threads are TLS-vs-plaintext port mismatches. Test with mosquitto_sub -h host -p 1883 -v -t '#' before blaming Laravel.
Wildcards on subscribe only. devices/+/status and devices/# work in subscribe(); publish() to a wildcard topic is silently invalid.
Retained messages outlive your code. A stale retained payload will confuse you for an hour. Clear it with a zero-length retained publish to that topic, and add a mqtt:purge-topic command to your ops toolbox. Queued publishes make this slightly more likely, since the clear-out command you ran may itself be sitting in a queue.
No logging out of the box. php-mqtt/client accepts a PSR-3 logger; inject Log::getLogger() when constructing if you want to see the CONNECT/PUBLISH/SUBACK traffic. Without it, failures are silent until something times out — and on the queued path there's no timeout in front of you to notice.
Size and maintenance signal. Small star count, no dependents, one maintainer, and an odd future-dated release history. v3.3.0 adding queued publishing is a genuine usability improvement and a good sign, but it's also a brand-new code path with no battle testing — pin an exact version in composer.json (3.3.0 if you want queue(), 3.2.x if you'd rather write your own job and keep the surface you've already reviewed), and don't build anything load-bearing on undocumented behaviour. If you need actively maintained equivalents with the same feature set, php-mqtt/laravel-client is the direct comparison; since your own Publisher interface isolates the dependency, switching — or flipping between publish() and queue() — is a one-file change.
How can I help you explore Laravel packages today?