NAME
Fugu::MQTT - a subscribing MQTT client for a single-threaded daemon
SYNOPSIS
use Fugu::MQTT;
my $mqtt = Fugu::MQTT->new(
host => '127.0.0.1',
port => 1883,
);
$mqtt->mqtt_connect(10) or die "no broker\n";
$mqtt->subscribe('stat/+/RESULT', sub ($topic, $payload) {
...
});
$mqtt->publish('cmnd/lamp/POWER', 'ON');
# From the event loop, once each pass
$mqtt->tick;DESCRIPTION
Fugu::MQTT is a wrapper around Net::MQTT::Simple for a daemon that has one process and one event loop. It never blocks the caller. tick drains what arrived and returns, thus the loop keeps the MQTT connection beside its own descriptors.
Net::MQTT::Simple loads at connect time, not at compile time. Thus the module keeps the core-Perl load contract of Fugu, and a daemon whose broker is optional still starts without the library.
Every message goes through a callback. The module holds the incoming messages in a queue and dispatches them in tick, in the process that called it. A callback thus runs where the caller can reason about it, and never inside a signal handler.
new
new(%args) creates a client. The method opens no socket. Call mqtt_connect.
These are the arguments:
host-
The broker. The default is 127.0.0.1.
port-
The broker port. The default is 1883.
username-
The login name. The client sends no credentials when this argument is absent.
password-
The login password.
mqtt_connect
mqtt_connect($timeout) connects to the broker. The default timeout is 10 seconds.
A connect can block in the resolver or in the handshake, and no poll loop can interrupt either. Thus the guard is the alarm of Fugu::Timeout. A broker that does not answer costs the caller the timeout, not the run.
The method returns true when the client is connected.
subscribe
subscribe($topic, $callback) registers a topic and its callback. The callback gets ($topic, $payload), where the topic is the one that arrived, not the pattern.
The pattern accepts the MQTT wildcards: + matches one level and # matches the remaining levels. The module registers the subscription even when the client is not connected, so a caller can subscribe before the broker is up. resubscribe sends the set again after a reconnection.
publish
publish($topic, $payload, $retain) sends one message. A true $retain makes the broker keep the message for the next subscriber. The method does nothing when the client is not connected.
tick
tick($timeout) reads what arrived and runs the callbacks. The default timeout is 0, which does not block. The method returns the number of dispatched messages.
The method reports a lost connection: it clears the connected state and returns 0. The caller then calls reconnect on a schedule of its own.
reconnect
reconnect disconnects, connects again, and subscribes again. The method returns 1 on success and 0 on failure.
resubscribe
resubscribe sends every registered subscription to the broker again. reconnect calls it. A caller needs it only when it connects by itself.
disconnect
disconnect closes the connection and discards the queued messages. The registered subscriptions stay, thus a later reconnect restores them.
is_connected
is_connected reports if the client holds a connection.
subscriptions
subscriptions returns the registered topic patterns.
RETURN VALUES
mqtt_connect and is_connected return true when the client is connected. reconnect returns 1 or 0. tick returns a count. subscriptions returns a list.
No method dies. A broker error, a bad callback, and a lost connection all go to the log of Fugu::Log. A daemon must not exit because a broker went away.
EXAMPLES
This example reconnects when the broker goes away:
$loop->every(1, sub {
$mqtt->tick;
$mqtt->reconnect unless $mqtt->is_connected;
});SEE ALSO
Fugu::Log, Fugu::Timeout, Net::MQTT::Simple
AUTHORS
Dick Olsson <hi@senzilla.io>
CAVEATS
A callback runs inside tick. A callback that blocks blocks the loop. Keep the work in a callback short, or queue it.
The module holds no quality of service beyond what Net::MQTT::Simple gives, which is QoS 0. A message can be lost. A caller that must know a device answered reads the state topic of that device.