### ----------------------------------------------------
### -- MQTT::Client
### -- Licenses: Artistic-2.0
### -- Authors: 
### -- File: lib/MQTT/Client.rakumod
### ----------------------------------------------------
use MQTT::Client::MyPack;

unit class MQTT::Client;

has Int      $.keepalive-interval is rw = 60;
has Int      $.maximum-length     is rw = 2097152;  # 2 MB
has Str      $.client-identifier  is rw = "perl6";
has Str      $.server             is rw;
has Int      $.port               is rw = 1883;
has Supply   $!messages;
has IO::Socket::Async $!connection;

our sub filter-as-regex($filter) {
    my sub quotemeta ($str is copy) {
        $str ~~ s:g[\W+] = "'$/'";
        return $str;
    }

    return /^ <!before '$'>/ if $filter eq '#';
    return /^ '/'          / if $filter eq '/#';

    my $regex = '^';
    my $anchor = True;
    $regex ~= "<!before '\$'>" if $filter ~~ /^\+/;

    my @parts = $filter.comb(/ '/#' | '/' | <-[/]>+ /);
    for @parts {
        when '/#' { $anchor = False; last }
        when '/'  { $regex ~= '\\/' }
        when '+'  { $regex ~= '<-[/]>*' }
        default   { $regex ~= quotemeta $_ }
    }
    $regex ~= '$' if $anchor;

    /<$regex>/;
}

method connect() {
    start {
        use experimental :pack;

        $!connection = await IO::Socket::Async.connect: $!server, $!port;
        my $packets = self!incoming-packets($!connection.Supply(:bin)).share;

        $!messages = supply {
            whenever $packets.grep(*.<type> == 3) {
                my $topic-length = .<data>.unpack("n");
                my $topic   = .<data>.subbuf(2, $topic-length)
                                .decode("utf8-c8");
                my $message = .<data>.subbuf(2 + $topic-length);
                my $retain  = .<retain>;
                emit { :$topic, :$message, :$retain };
            }
        }

        my $initialized = $packets.grep(*.<type> == 2).head(1).Promise;

        $!connection.write: mypack "C m/(n/a* C C n n/a*)", 0x10,
            "MQIsdp", 3, 2, $!keepalive-interval, $!client-identifier;

        await $initialized;

        Supply.interval( $!keepalive-interval ).tap: {
            $!connection.write: pack "C x", 0xc0;
        }

        True
    }
}

method !parse (Buf $buf is rw) {
    my $offset = 1;

    my $multiplier = 1;
    my $length = 0;
    my $d = 0;

    repeat {
        return False if $offset >= $buf.elems;
        $d = $buf[$offset++];
        $length += ($d +& 0x7f) * $multiplier;
        $multiplier *= 128;
    } while $d +& 0x80;

    return False if $length + $offset > $buf.elems;

    my $first_byte = $buf[0];
    my %packet := {
        type   => ($first_byte +& 0xf0) +> 4,
        dup    => ($first_byte +& 0x08) +> 3,
        qos    => ($first_byte +& 0x06) +> 1,
        retain => ($first_byte +& 0x01),
        data   => $buf.subbuf($offset, $length);
    };

    $buf .= subbuf($offset + $length);

    %packet
}

method !incoming-packets (Supply $binary) {
    my Buf $buf .= new;
    return supply {
        whenever $binary -> $received {
            $buf.push: $received;
            while self!parse($buf) -> %hash {
                emit %hash;
            }
        }
    }
}

multi method publish(Str $topic, Blob $message) {
    $!connection.write: mypack "C m/(n/a* a*)", 0x30, $topic, $message
}

multi method publish(Str $topic, Str $message) {
    self.publish: $topic, $message.encode
}

multi method retain(Str $topic, Blob $message) {
    $!connection.write: mypack "C m/(n/a* a*)", 0x31, $topic, $message
}

multi method retain(Str $topic, Str $message) {
    self.publish: $topic, $message.encode
}

method subscribe(Str $topic --> Supply:D) {
    $!connection.write: mypack "C m/(C C n/a* C)", 0x82,
        0, 0, $topic, 0;

    my $regex = filter-as-regex($topic);

    $!messages.grep: { .<topic> ~~ $regex }
}

method messages(--> Supply:D) {
    $!messages.Supply
}

method connection (--> Promise:D) {
    ...
}

=begin pod

=head1 NAME

MQTT::Client - Minimal MQTT v3 client interface for Perl 6

=head1 SYNOPSIS

=begin code :lang<raku>

use MQTT::Client;

my $m = MQTT::Client.new('test.mosquitto.org');
await $m.connect;

$m.publish("hello-world", "$*PID is still here!");

react {
    whenever $m.subscribe("typing-speed-test.aoeu.eu") {
        say "Typing test completed at { .<message>.decode("utf8-c8") }";
    }

    whenever Supply.interval(10) {
        $m.publish("hello-world", "$*PID is still here!");
    }
}

=end code

=head1 METHODS

=head2 new($server, $port)

Returns a new C<MQTT::Client> object. Note: it does not automatically
connect. Before doing anything, C<.connect> should be called!

=head2 connect

Attempts to connect to the MQTT broker, and returns a C<Promise> that
will be kept after connection confirmation from the broker.

=head2 connection

Returns a C<Promise> that isn't kept until the connection ends.

=head2 publish(Str $topic, Buf|Str $message)

=head2 retain(Str $topic, Buf|Str $message)

Publish a message to the MQTT broker. If the $message is a Str, it will be
UTF-8 encoded. Topics are always UTF-8 encoded in MQTT.

=head2 subscribe($topic)

Subscribes to the topic, returning a C<Supply> of messages that
match the topic.  Note that messages that are matched by multiple
subscriptions will be passed to all of the supplies that match.

Tap the supply to receive, for each message, a hash with the keys
C<topic>, C<message>, and a boolean C<retain>. Note that C<message>
is a C<Buf> (binary buffer), which in most cases you will need to
C<.decode> before you can use it.

=head1 FUNCTIONS

=head2 MQTT::Client::filter_as_regex(topic_filter)

Given a valid MQTT topic filter, returns the corresponding regular
expression.

=head1 NOT YET IMPLEMENTED

The following features that are present in Net::MQTT::Simple for
Perl, have not yet been implemented in C<MQTT::Client> for Raku:

=item dropping the connection when a large message is received

=item unsubscribe

=item automatic reconnecting

=item SSL

=item connection timeouts

=item simple functional interface

=head1 NOT SUPPORTED

=item QoS (Quality of Service)

Every message is published at QoS level 0, that is, "at most once",
also known as "fire and forget".

=item DUP (Duplicate message)

Since QoS is not supported, no retransmissions are done, and no
message will indicate that it has already been sent before.

=item Authentication

No username and password are sent to the server.

=item Last will

The server won't publish a "last will" message on behalf of us when
our connection's gone.

=item Large data

Because everything is handled in memory and there's no way to indicate
to the server that large messages are not desired, the connection is
dropped as soon as the server announces a packet larger than 2 megabytes.

=item Validation of server-to-client communication

The MQTT spec prescribes mandatory validation of all incoming data,
and disconnecting if anything (really, anything) is wrong with it.
However, this minimal implementation silently ignores anything it
doesn't specifically handle, which may result in weird behaviour if
the server sends out bad data.

Most clients do not adhere to this part of the specifications.

=head1 AUTHOR

Juerd Waalboer

=head1 COPYRIGHT AND LICENSE

Copyright 2015 - 2019 Juerd Waalboer

Copyright 2024 Raku Community

This library is free software; you can redistribute it and/or modify it under the Artistic License 2.0.

=head1 SEE ALSO

L<Net::MQTT::Simple|https://metacpan.org/pod/Net::MQTT::Simple> inr Perl

=end pod

# vim: expandtab shiftwidth=4


