AnyEvent-MQTT
view release on metacpan or search on metacpan
lib/AnyEvent/MQTT.pm view on Meta::CPAN
use strict;
use warnings;
package AnyEvent::MQTT;
$AnyEvent::MQTT::VERSION = '1.212810';
# ABSTRACT: AnyEvent module for an MQTT client
use constant DEBUG => $ENV{ANYEVENT_MQTT_DEBUG};
use AnyEvent;
use AnyEvent::Handle;
use Net::MQTT::Constants;
use Net::MQTT::Message;
use Net::MQTT::TopicStore;
use Carp qw/croak carp/;
use Sub::Name;
use Scalar::Util qw/weaken/;
sub new {
my ($pkg, %p) = @_;
my $self =
bless {
socket => undef,
host => '127.0.0.1',
port => '1883',
timeout => 30,
wait => 'nothing',
keep_alive_timer => 120,
qos => MQTT_QOS_AT_MOST_ONCE,
message_id => 1,
user_name => undef,
password => undef,
tls => undef,
will_topic => undef,
will_qos => MQTT_QOS_AT_MOST_ONCE,
will_retain => 0,
will_message => '',
client_id => undef,
clean_session => 1,
handle_args => [],
write_queue => [],
inflight => {},
_sub_topics => Net::MQTT::TopicStore->new(),
%p,
}, $pkg;
}
sub DESTROY {
$_[0]->cleanup;
}
sub cleanup {
my $self = shift;
print STDERR "cleanup\n" if DEBUG;
if ($self->{handle}) {
my $cv = AnyEvent->condvar;
my $handle = $self->{handle};
weaken $handle;
$cv->cb(sub { $handle->destroy });
$self->_send(message_type => MQTT_DISCONNECT, cv => $cv);
}
delete $self->{handle};
delete $self->{connected};
delete $self->{wait};
delete $self->{_keep_alive_handle};
delete $self->{_keep_alive_waiting};
$self->{write_queue} = [];
}
sub _error {
my ($self, $fatal, $message, $reconnect) = @_;
$self->cleanup($message);
$self->{on_error}->($fatal, $message) if ($self->{on_error});
$self->_reconnect() if ($reconnect);
}
sub publish {
my ($self, %p) = @_;
my $topic = exists $p{topic} ? $p{topic} :
croak ref $self, '->publish requires "topic" parameter';
my $qos = exists $p{qos} ? $p{qos} : MQTT_QOS_AT_MOST_ONCE;
my $cv = exists $p{cv} ? delete $p{cv} : AnyEvent->condvar;
my $expect;
if ($qos) {
$expect = ($qos == MQTT_QOS_AT_LEAST_ONCE ? MQTT_PUBACK : MQTT_PUBREC);
}
my $message = $p{message};
if (defined $message) {
print STDERR "publish: message[$message] => $topic\n" if DEBUG;
$self->_send_with_ack({
lib/AnyEvent/MQTT.pm view on Meta::CPAN
sub _keep_alive_received {
my $self = shift;
print STDERR "keep alive received\n" if DEBUG;
return unless (defined $self->{_keep_alive_waiting});
$self->_reset_keep_alive_timer();
}
sub connect {
my ($self, $msg, $cv) = @_;
print STDERR "connect\n" if DEBUG;
$self->{_waiting} = 'connect';
if ($msg) {
$cv = AnyEvent->condvar unless ($cv);
$self->_queue_write($msg, $cv);
} else {
$self->{connect_cv} = AnyEvent->condvar unless (exists $self->{connect_cv});
$cv = $self->{connect_cv};
}
return $cv if ($self->{handle});
my $weak_self = $self;
weaken $weak_self;
my $hd;
$hd = $self->{handle} =
AnyEvent::Handle->new(connect => [$self->{host}, $self->{port}],
($self->{tls} ? (tls => "connect") : ()),
on_error => subname('on_error_cb' => sub {
my ($handle, $fatal, $message) = @_;
print STDERR "handle error $_[1]\n" if DEBUG;
$handle->destroy;
$weak_self->_error($fatal, 'Error: '.$message, 0);
}),
on_eof => subname('on_eof_cb' => sub {
my ($handle) = @_;
print STDERR "handle eof\n" if DEBUG;
$handle->destroy;
$weak_self->_error(1, 'EOF', 1);
}),
on_timeout => subname('on_timeout_cb' => sub {
$weak_self->_error(0, $weak_self->{wait}.' timeout', 1);
$weak_self->{wait} = 'nothing';
}),
on_connect => subname('on_connect_cb' => sub {
my ($handle, $host, $port, $retry) = @_;
print STDERR "TCP handshake complete\n" if DEBUG;
# call user-defined on_connect function.
$weak_self->{on_connect}->($handle, $retry) if $weak_self->{on_connect};
my $msg =
Net::MQTT::Message->new(
message_type => MQTT_CONNECT,
keep_alive_timer => $weak_self->{keep_alive_timer},
client_id => $weak_self->{client_id},
clean_session => $weak_self->{clean_session},
will_topic => $weak_self->{will_topic},
will_qos => $weak_self->{will_qos},
will_retain => $weak_self->{will_retain},
will_message => $weak_self->{will_message},
user_name => $weak_self->{user_name},
password => $weak_self->{password},
);
$weak_self->_write_now($msg);
$handle->timeout($weak_self->{timeout});
$weak_self->{wait} = 'connack';
$handle->on_read(subname 'on_read_cb' => sub {
my ($hdl) = @_;
$hdl->push_read(ref $weak_self =>
subname 'reader_cb' => sub {
$weak_self->_handle_message(@_);
1;
});
});
}),
@{$self->{handle_args}},
);
return $cv
}
sub _reconnect {
my $self = shift;
print STDERR "reconnecting:\n" if DEBUG;
# must resubscribe everything
if ($self->{clean_session}) {
$self->{_sub_topics} = Net::MQTT::TopicStore->new();
$self->{_sub_reconnect} = delete $self->{_sub} || {};
}
$self->connect(@_);
}
sub _handle_message {
my $self = shift;
my ($handle, $msg, $error) = @_;
return $self->_error(0, $error, 1) if ($error);
$self->{message_log_callback}->('<', $msg) if ($self->{message_log_callback});
$self->_call_callback('before_msg_callback' => $msg) or return;
my $msg_type = lc ref $msg;
$msg_type =~ s/^.*:://;
$self->_call_callback('before_'.$msg_type.'_callback' => $msg) or return;
my $method = '_process_'.$msg_type;
unless ($self->can($method)) {
carp 'Unsupported message ', $msg->string(), "\n";
return;
}
my $res = $self->$method(@_);
$self->_call_callback('after_'.$msg_type.'_callback' => $msg, $res);
$res;
}
sub _call_callback {
my $self = shift;
my $cb_name = shift;
return 1 unless (exists $self->{$cb_name});
$self->{$cb_name}->(@_);
}
sub _process_connack {
my ($self, $handle, $msg, $error) = @_;
$handle->timeout(undef);
lib/AnyEvent/MQTT.pm view on Meta::CPAN
use AnyEvent::MQTT;
my $mqtt = AnyEvent::MQTT->new;
my $cv = $mqtt->subscribe(topic => '/topic',
callback => sub {
my ($topic, $message) = @_;
print $topic, ' ', $message, "\n"
});
my $qos = $cv->recv; # subscribed, negotiated QoS == $qos
# publish a simple message
$cv = $mqtt->publish(message => 'simple message',
topic => '/topic');
$cv->recv; # sent
# publish line-by-line from file handle
$cv = $mqtt->publish(handle => \*STDIN,
topic => '/topic');
$cv->recv; # sent
# publish from AnyEvent::Handle
$cv = $mqtt->publish(handle => AnyEvent::Handle->new(my %handle_args),
topic => '/topic');
$cv->recv; # sent
=head1 DESCRIPTION
AnyEvent module for MQTT client.
B<IMPORTANT:> This is an early release and the API is still subject to
change.
=head1 METHODS
=head2 C<new(%params)>
Constructs a new C<AnyEvent::MQTT> object. The supported parameters
are:
=over
=item C<host>
The server host. Defaults to C<127.0.0.1>.
=item C<port>
The server port. Defaults to C<1883>.
=item C<timeout>
The timeout for responses from the server.
=item C<keep_alive_timer>
The keep alive timer.
=item C<user_name>
The user name for the MQTT broker.
=item C<password>
The password for the MQTT broker.
=item C<tls>
Set flag to enable TLS encryption, Default is no encryption.
=item C<will_topic>
Set topic for will message. Default is undef which means no will
message will be configured.
=item C<will_qos>
Set QoS for will message. Default is 'at-most-once'.
=item C<will_retain>
Set retain flag for will message. Default is 0.
=item C<will_message>
Set message for will message. Default is the empty message.
=item C<clean_session>
Set clean session flag for connect message. Default is 1.
=item C<client_id>
Sets the client id for the client overriding the default which
is C<NetMQTTpmNNNNN> where NNNNN is the current process id.
=item C<message_log_callback>
Defines a callback to call on every message.
=item C<on_error>
Defines a callback to call when some error occurs.
Two parameters are passed to the callback.
$on_error->($fatal, $message)
where C<$fatal> is a boolean flag and C<$message> is the error message.
If the error is fatal, C<$fatal> is true.
=item C<handle_args>
a reference to a list to pass as arguments to the
L<AnyEvent::Handle> constructor (defaults to
an empty list reference).
=back
=head2 C<cleanup()>
This method attempts to destroy any resources in the event of a
disconnection or fatal error.
=head2 C<publish( %parameters )>
( run in 5.283 seconds using v1.01-cache-2.11-cpan-a5162978ef8 )