/ 
Beekeeper-0.08
River stage zero No dependents
/ Beekeeper::MQTT

NAME

Beekeeper::MQTT - Asynchronous MQTT 5.0 client

VERSION

Version 0.08

SYNOPSIS

my $mqtt = Beekeeper::MQTT->new(
    host     => 'localhost',
    username => 'guest',
    password => 'guest',
);

$mqtt->connect( 
    blocking => 1,
    on_connack => sub {
        my ($success, $properties) = @_;
        die $properties->{reason_string} unless $success;
    },
);

$mqtt->subscribe(
    topic => 'foo/bar',
    on_publish => sub {
        my ($payload, $properties) = @_;
        print "Got a message: $$payload";
    },
);

$mqtt->publish(
    topic   => 'foo/bar',
    payload => 'Hello',
);

$mqtt->unsubscribe(
    topic => 'foo/bar',
);

$mqtt->disconnect;

Most methods allow to send arbitrary properties as key-value pairs of utf8 strings.

Except for trivial cases, error checking is delegated to the server.

This client can send and receive around 20000 messages per second.

TODO

- Keep Alive

CONSTRUCTOR

new ( %options )

host

Hostname or IP address of the MQTT server. It also accepts an array of adresses which conforms a cluster, in which case the connection will be stablished against a randomly choosen node of the cluster.

port

Port of the MQTT server. If not specified use the MQTT default of 1883.

tls

Enable the use of TLS for MQTT connections.

username

Username used to authenticate against the server.

password

Password used to authenticate against the server.

timeout

Connection timeout in fractional seconds before giving up. Default is 30 seconds. If set to zero the connection to server it retried forever.

on_error => $cb->( $errmsg )

Optional callback which is executed when an error condition occurs. If not specified, the default is to die with $errmsg. Usually the server has already closed the connection when this is called.

METHODS

connect ( %options )

Connects to the MQTT server. On failure retries until timeout.

blocking => $bool

When set to true this method acts as a blocking call: it does not return until a connection has been established and handshake has been completed.

on_connack => $cb->( $success, \%properties )

Callback which is executed after the server accepted the connection.

disconnect ( %args )

Does a graceful disconnection from the server.

$reason_code

Disconnect Reason Code as stated in the chapter 3.14.2.1 of the specification. Default is zero, meaning normal disconnection.

subscribe ( %args )

Creates a subscription to a topic (or a list of topics). When a message is received, it will be passed to given on_publish callback:

$mqtt->subscribe(
    topic       => 'topic/foo',
    maximum_qos => 1,
    on_publish  => sub {
        my ($payload, \%properties) = @_;
        print "Got message from topic/foo : $$payload";
    },
);
topics => \@topics

List of topics that the client wants to subscribe to.

on_publish => $cb->( \$payload, \%properties )

Required callback which is executed when a message matching any of the subscription topics is received.

on_suback => $cb->( $success, \%properties, ... )

Optional callback which is executed after subscription is acknowledged by the server with a SUBACK packet.

User properties

Any other argument other than maximum_qos, no_local, retain_as_published or retain_handling is sent as an "User Property" (a key-value pair of utf8 strings).

unsubscribe ( %params )

Cancels existing subscriptions, the client will no longer receive messages from the specified topics. Example:

$mqtt->unsubscribe( 
    topic => 'topic/foo',
);

$mqtt->unsubscribe( 
    topics => ['topic/bar','topic/baz']
);
topic => $topic

The destination of an existing subscription.

topics => \@topics

Reference to an array of destinations of existing subscriptions.

on_unsuback => $cb->()

Optional user defined callback which is called after unsubscription is completed.

publish ( %args )

Sends a message to a topic. Example:

$mqtt->publish(
    topic     => 'foo/bar',
    payload   => 'Hello!',
    qos       => 1,
    on_puback => sub {
       my ($reason_code) = @_;
       print "Message was sent";
    }
);
topic => $str

Utf8 string containing the topic name.

payload => \$data

Scalar or scalar reference containing either an utf8 string or a binary blob which conforms the payload of the message. It is allowed to publish messages without payload.

qos => $bool

Quality of service level (QoS). Only levels 0 and 1 are supported. Default is 0.

duplicate => $bool

Must be set to a true value to indicate a message retransmission.

retain => $bool

When true sets the message retain flag.

message_expiry_interval => $int

Expiration period in seconds. The server will discard retained messages after this period has ellapsed.

response_topic => $str

Utf8 string containing the response topic name.

on_puback => $cb->($reason_code)

Optional callback which is executed after the server answers with a PUBACK packet, acknowledging that it has received it. Allowed only for messages published with QoS 1.

User properties

Any aditional argument will be sent as an "User Property", this is as a key-value pair of utf8 strings.

puback ( %args )

Acknowledges the receipt of a message received from a subscription with QoS 1. The server should resend the message until it is acknowledged.

packet_id => $int

Packet ID of the message being acknowledged.

reason_code => $int

If not specified it will default to zero, signaling success.

flush_buffer

Send several packets into a single socket write. This is more efficient than individual send() calls because Nagle's algorithm is disabled.

discard_buffer

Discard buffered packets.

SEE ALSO

The MQTT specification can be found at https://mqtt.org/mqtt-specification

AUTHOR

José Micó, jose.mico@gmail.com

COPYRIGHT AND LICENSE

Copyright 2021 José Micó.

This is free software; you can redistribute it and/or modify it under the same terms as the Perl 5 programming language itself.

This software is distributed in the hope that it will be useful, but it is provided “as is” and without any express or implied warranties. For details, see the full text of the license in the file LICENSE.