File Coverage

blib/lib/Net/MQTT/Simple.pm

Criterion	Covered	Total	%
statement	24	217	11.0
branch	7	132	5.3
condition	0	54	0.0
subroutine	6	32	18.7
pod	11	11	100.0
total	48	446	10.7

package Net::MQTT::Simple;

use strict;
use warnings;

our $VERSION = '1.27';

# Please note that these are not documented and are subject to change:
our $KEEPALIVE_INTERVAL = 60;
our $PING_TIMEOUT = 10;
our $RECONNECT_INTERVAL = 5;
our $MAX_LENGTH = 2097152;    # 2 MB
our $READ_BYTES = 16 * 1024;  # 16 kB per IO::Socket::SSL recommendation
our $WRITE_BYTES = 16 * 1024; # 16 kB per IO::Socket::SSL maximum
our $PROTOCOL_LEVEL = 0x04;   # 0x03 in v3.1, 0x04 in v3.1.1
our $PROTOCOL_NAME = "MQTT";  # MQIsdp in v3.1, MQTT in v3.1.1

my $global;

BEGIN {
    *_socket_class =
      eval { require IO::Socket::IP; 1 }   ? sub { "IO::Socket::IP" }
    : eval { require IO::Socket::INET; 1 } ? sub { "IO::Socket::INET" }
    : die "Neither IO::Socket::IP nor IO::Socket::INET found";
}

sub _default_port { 1883 }
sub _socket_error { "$@" }
sub _secure { 0 }


sub _client_identifier { my ($class) = @_; return "Net::MQTT::Simple[" . $class->{random_id} . "]"; }

# Carp might not be available either.
sub _croak {
    die sprintf "%s at %s line %d.\n", "@_", (caller 1)[1, 2];
}

sub filter_as_regex {
    my ($filter) = @_;

    return "^(?!\\\$)" if $filter eq '#';   # Match everything except /^\$/
    return "^/"        if $filter eq '/#';  # Parent (empty topic) is invalid

    $filter = quotemeta $filter;

    $filter =~ s{ \z (?
    $filter =~ s{ \\ \/ \\ \#           } {}x;
    $filter =~ s{ \\ \+                 } {[^/]*+}xg;
    $filter =~ s{ ^ (?= \[ \^ / \] \* ) } {(?!\\\$)}x;  # No /^\$/ if /^\+/

    return "^$filter";
}

sub import {
    my ($class, $server) = @_;
    @_ <= 2 or _croak "Too many arguments for use " . __PACKAGE__;

    $server or return;

    $global = $class->new($server);

    no strict 'refs';
    *{ (caller)[0] . "::publish" } = \&publish;
    *{ (caller)[0] . "::retain"  } = \&retain;
}

sub new {
    my ($class, $server, $sockopts) = @_;
    @_ == 2 or @_ == 3 or _croak "Wrong number of arguments for $class->new";

    my $port = $class->_default_port;

    # Add port for bare IPv6 address
    $server = "[$server]:$port" if $server =~ /:.*:/ and not $server =~ /\[/;

    # Add port for bare IPv4 address or bracketed IPv6 address
    $server .= ":$port" if $server !~ /:/ or $server =~ /^\[.*\]$/;

	# Create a random ID for the instance of the object
	my $random_id = join "", map chr 65 + int rand 26, 1 .. 10;

    return bless {
        server       => $server,
        last_connect => 0,
        sockopts     => $sockopts // {},
        random_id    => $random_id
    }, $class;
}

sub last_will {
    my ($self, $topic, $message, $retain) = @_;

    my %old;
    %old = %{ $self->{will} } if $self->{will};

    _croak "Wrong number of arguments for last_will" if @_ > 4;

    if (@_ >= 2) {
        if (not defined $topic and not defined $message) {
            delete $self->{will};
            delete $self->{encoded_will};

            return;
        } else {
            $self->{will} = {
                topic   => $topic    // $old{topic}   // '',
                message => $message  // $old{message} // '',
                retain  => !!$retain // $old{retain}  // 0,
            };
            _croak("Topic is empty") if not length $self->{will}->{topic};

            my $e = $self->{encoded_will} = { %{ $self->{will} } };
            utf8::encode($e->{topic});
            utf8::downgrade($e->{message}, 1) or do {
                my ($file, $line, $method) = (caller 1)[1, 2, 3];
                warn "Wide character in $method at $file line $line.\n";
                utf8::encode($e->{message});
            };
        }
    }

    return @{ $self->{will} }{qw/topic message retain/};
}

sub login {
    my ($self, $username, $password) = @_;


    if (@_ > 1) {
        _croak "Password login is disabled for insecure connections"
            if defined $password
            and not $self->_secure || $ENV{MQTT_SIMPLE_ALLOW_INSECURE_LOGIN};

        utf8::encode($username);
        $self->{username} = $username;
        $self->{password} = $password;
    }

    return $username;
}

sub _connect {
    my ($self) = @_;

    return if $self->{socket} and $self->{socket}->connected;

    if ($self->{last_connect} > time() - $RECONNECT_INTERVAL) {
        select undef, undef, undef, .01;
        return;
    }

    # Reset state
    $self->{last_connect} = time;
    $self->{buffer} = "";
    delete $self->{ping};

    # Connect
    my $socket_class = $self->_socket_class;
    my %socket_options = (
        PeerAddr => $self->{server},
        %{ $self->{sockopts} }
    );
    $self->{socket} = $socket_class->new( %socket_options )
        or warn "$0: connect: " . $self->_socket_error . "\n";

    # Say hello
    local $self->{skip_connect} = 1;  # avoid infinite recursion :-)
    $self->_send_connect;
    $self->_send_subscribe;
}

sub _prepend_variable_length {
    # Copied from Net::MQTT::Constants
    my ($data) = @_;
    my $v = length $data;
    my $o = "";
    my $d;
    do {
        $d = $v % 128;
        $v = int($v/128);
        $d |= 0x80 if $v;
        $o .= pack "C", $d;
    } while $d & 0x80;
    return "$o$data";
}

sub _send {
    my ($self, $data) = @_;

    $self->_connect unless exists $self->{skip_connect};
    delete $self->{skip_connect};

    my $socket = $self->{socket} or return;

    while (my $chunk = substr $data, 0, $WRITE_BYTES, "") {
        syswrite $socket, $chunk
            or $self->_drop_connection;  # reconnect on next message
    }

    $self->{last_send} = time;
}

sub _send_connect {
    my ($self) = @_;

    my $will = $self->{encoded_will};
    my $flags = 0x02;
    $flags |= 0x04 if $will;
    $flags |= 0x20 if $will and $will->{retain};

    $flags |= 0x80 if defined $self->{username};
    $flags |= 0x40 if defined $self->{username} and defined $self->{password};

    $self->_send("\x10" . _prepend_variable_length(pack(
        "x C/a* C C n n/a*"
            . ($flags & 0x04 ? "n/a* n/a*" : "")
            . ($flags & 0x80 ? "n/a*" : "")
            . ($flags & 0x40 ? "n/a*" : ""),
        $PROTOCOL_NAME,
        $PROTOCOL_LEVEL,
        $flags,
        $KEEPALIVE_INTERVAL,
        $self->_client_identifier,
        ($flags & 0x04 ? ($will->{topic}, $will->{message}) : ()),
        ($flags & 0x80 ? $self->{username} : ()),
        ($flags & 0x40 ? $self->{password} : ()),
    )));
}

sub _send_subscribe {
    my ($self, @topics) = @_;

    if (not @topics) {
        @topics = keys %{ $self->{sub} } or return;
    }
    return if not @topics;

    utf8::encode($_) for @topics;

    # Hardcoded "packet identifier" \0\x01 for now (was \0\0 but the spec
    # disallows it for subscribe packets and mosquitto started enforcing that.)
    $self->_send("\x82" . _prepend_variable_length("\0\x01" .
        pack("(n/a* x)*", @topics)  # x = QoS 0
    ));
}

sub _send_unsubscribe {
    my ($self, @topics) = @_;

    return if not @topics;

    utf8::encode($_) for @topics;

    # Hardcoded "packet identifier" \0\0x01 for now; see above.
    $self->_send("\xa2" . _prepend_variable_length("\0\x01" .
        pack("(n/a*)*", @topics)
    ));
}

sub _parse {
    my ($self) = @_;

    my $bufref = \$self->{buffer};

    return if length $$bufref < 2;

    my $offset = 1;

    my $length = do {
        my $multiplier = 1;
        my $v = 0;
        my $d;
        do {
            return if $offset >= length $$bufref;  # not enough data yet
            $d = unpack "C", substr $$bufref, $offset++, 1;
            $v += ($d & 0x7f) * $multiplier;
            $multiplier *= 128;
        } while ($d & 0x80);
        $v;
    };

    if ($length > $MAX_LENGTH) {
        # On receiving an enormous packet, just disconnect to avoid exhausting
        # RAM on tiny systems.
        # TODO: just slurp and drop the data
        $self->_drop_connection;
        return;
    }

    return if length($$bufref) < $offset + $length;  # not enough data yet

    my $first_byte = unpack "C", substr $$bufref, 0, 1;

    my $packet = {
        type   => ($first_byte & 0xF0) >> 4,
        dup    => ($first_byte & 0x08) >> 3,
        qos    => ($first_byte & 0x06) >> 1,
        retain => ($first_byte & 0x01),
        data   => substr($$bufref, $offset, $length),
    };

    substr $$bufref, 0, $offset + $length, "";  # remove the parsed bits.

    return $packet;
}

sub _incoming_publish {
    my ($self, $packet) = @_;

    # Because QoS is not supported, no packed ID in the data. It would
    # have been 16 bits between $topic and $message.
    my ($topic, $message) = unpack "n/a a*", $packet->{data};

    utf8::decode($topic);

    for my $cb (@{ $self->{callbacks} }) {
        if ($topic =~ /$cb->{regex}/) {
            $cb->{callback}->($topic, $message, $packet->{retain});
            return;
        }
    }
}

sub _publish {
    my ($self, $retain, $topic, $message) = @_;

    $message //= "" if $retain;

    utf8::encode($topic);
    utf8::downgrade($message, 1) or do {
        my ($file, $line, $method) = (caller 1)[1, 2, 3];
        warn "Wide character in $method at $file line $line.\n";
        utf8::encode($message);
    };

    $self->_send(
        ($retain ? "\x31" : "\x30")
        . _prepend_variable_length(
            pack("n/a*", $topic) . $message
        )
    );
}

sub publish {
    my $method = UNIVERSAL::isa($_[0], __PACKAGE__);
    @_ == ($method ? 3 : 2) or _croak "Wrong number of arguments for publish";

    ($method ? shift : $global)->_publish(0, @_);
}

sub retain {
    my $method = UNIVERSAL::isa($_[0], __PACKAGE__);
    @_ == ($method ? 3 : 2) or _croak "Wrong number of arguments for retain";

    ($method ? shift : $global)->_publish(1, @_);
}

sub run {
    my ($self, @subscribe_args) = @_;

    $self->subscribe(@subscribe_args) if @subscribe_args;

    until ($self->{stop_loop}) {
        my @timeouts;
        push @timeouts, $KEEPALIVE_INTERVAL - (time() - $self->{last_send})
            if exists $self->{last_send};
        push @timeouts, $PING_TIMEOUT - (time() - $self->{ping})
            if exists $self->{ping};

        my $timeout = @timeouts
            ? (sort { $a <=> $b } @timeouts)[0]
            : 1;  # default to 1

        $self->tick($timeout);
    }

    delete $self->{stop_loop};
}

sub subscribe {
    my ($self, @kv) = @_;

    while (my ($topic, $callback) = splice @kv, 0, 2) {
        $self->{sub}->{ $topic } = 1;
        push @{ $self->{callbacks} }, {
            topic => $topic,
            regex => filter_as_regex($topic),
            callback => $callback,
        };
    }

    $self->_send_subscribe() if $self->{socket};
}

sub unsubscribe {
    my ($self, @topics) = @_;

    $self->_send_unsubscribe(@topics);

    my $cb = $self->{callbacks};
    for my $topic ( @topics ) {
      @$cb = grep {$_->{topic} ne $topic} @$cb;
    }

    delete @{ $self->{sub} }{ @topics };
}

sub tick {
    my ($self, $timeout) = @_;

    $self->_connect;

    my $socket = $self->{socket} or return;
    my $bufref = \$self->{buffer};

    my $r = '';
    vec($r, fileno($socket), 1) = 1;

    if (select $r, undef, undef, $timeout // 0) {
        sysread $socket, $$bufref, $READ_BYTES, length $$bufref
            or delete $self->{socket};

        while (length $$bufref) {
            my $packet = $self->_parse() or last;
            $self->_incoming_publish($packet) if $packet->{type} == 3;
            delete $self->{ping}              if $packet->{type} == 13;
        }
    }

    if (time() >= $self->{last_send} + $KEEPALIVE_INTERVAL) {
        $self->_send("\xc0\0");  # PINGREQ
        $self->{ping} = time;
    }
    if ($self->{ping} and time() >= $self->{ping} + $PING_TIMEOUT) {
        $self->_drop_connection;
    }

    return !! $self->{socket};
}

sub disconnect {
    my ($self) = @_;

    $self->_send(pack "C x", 0xe0)
        if $self->{socket} and $self->{socket}->connected;

    $self->_drop_connection;
}

sub _drop_connection {
    my ($self) = @_;

    delete $self->{socket};
    $self->{last_connect} = 0;
}

1;

__END__