view release on metacpan or  search on metacpan

lib/Fugu/MQTT.pm  view on Meta::CPAN

}

sub new ( $class, %args )
{
	my $self = bless {
		host     => $args{host} // '127.0.0.1',
		port     => $args{port} // 1883,
		username => $args{username},
		password => $args{password},

		subscriptions    => {},
		pending_messages => [],
		connected        => 0,
		last_tick        => 0,
	}, $class;

	return $self;
}

sub mqtt_connect ( $self, $timeout = 10 )
{
	Fugu::Log->default->debug(
		'Connecting to MQTT broker at %s:%d (timeout: %ds)',
		$self->{host}, $self->{port}, $timeout );

	# Try to load Net::MQTT::Simple if it is available
	local $@;

	# Capture the warnings from Net::MQTT::Simple
	my @warnings;
	local $SIG{__WARN__} = sub {
		push @warnings, shift;
	};

	# The connect can block in the resolver or the handshake, and a
	# poll loop cannot interrupt either. Thus the guard is an alarm.
	my $success = eval {
		Fugu::Timeout::bounded(
			$timeout,
			sub {

				# On OpenBSD the packages install under
				# site_perl, which a perl started with a
				# pruned @INC does not hold
				unshift @INC,
				    '/usr/local/libdata/perl5/site_perl'
				    unless grep {
					$_ eq
					    '/usr/local/libdata/perl5/site_perl'
				    } @INC;

				require Net::MQTT::Simple;

				my $server = $self->{host};
				if ( $self->{port} != 1883 ) {
					$server .= ':' . $self->{port};
				}

				my $mqtt = Net::MQTT::Simple->new($server);

				# Set the login credentials if the
				# configuration has a username
				if ( defined $self->{username} ) {
					$mqtt->login(
						$self->{username},
						$self->{password} // ''
					);
				}

				$self->{client}    = $mqtt;
				$self->{connected} = 1;
				Fugu::Log->default->debug(
					'Successfully connected to MQTT broker'
				);
				return 1;
			} );
	};

	_log_warnings( 'MQTT connection warning: %s', \@warnings );

	if ( $@ || !$success ) {
		my $err = $@ || "no answer within ${timeout}s";
		Fugu::Log->default->error( 'MQTT connection failed: %s', $err );
		$self->{connected} = 0;
	}

	return $self->{connected};
}

# $self->subscribe($topic, $callback):
#	Subscribe to an MQTT topic with a callback for messages.
#	$callback receives ($topic, $payload).
sub subscribe ( $self, $topic, $callback )
{
	Fugu::Log->default->debug( 'Subscribing to MQTT topic: %s', $topic );
	$self->{subscriptions}{$topic} = $callback;

	return unless $self->{connected} && $self->{client};

	$self->_register($topic);
}

# $self->_register($topic):
#	Register one topic with the client. Net::MQTT::Simple uses a
#	different subscription model: the module registers the topics
#	and polls for messages in tick(). Since 1.33,
#	Net::MQTT::Simple passes a retain flag as a third argument.
#	Accept and ignore all extra arguments.
sub _register ( $self, $topic )
{
	eval {
		$self->{client}->subscribe(
			$topic,
			sub ( $topic_received, $payload, @ ) {
				push @{ $self->{pending_messages} },
				    [ $topic_received, $payload ];
			} );
	};

	if ($@) {
		Fugu::Log->default->error( 'MQTT subscribe error for %s: %s',