File Coverage

blib/lib/IPC/Message/Minivan.pm

Criterion	Covered	Total	%
statement	10	12	83.3
branch			n/a
condition			n/a
subroutine	4	4	100.0
pod			n/a
total	14	16	87.5

package IPC::Message::Minivan;
use warnings;
use strict;
use 5.008;
use IPC::Messaging;
use JSON::XS;
use Time::HiRes;
use Regexp::Common;

use vars '$VERSION';
$VERSION = '0.01_08';

my $DEF_PORT = 6826;

sub new
{
	my ($class, %p) = @_;
	die "want host\n" unless $p{host};
	$p{port} ||= $DEF_PORT;
	eval { $p{sock} = IPC::Messaging->tcp_client($p{host}, $p{port}, by_line => 1); };
	$p{connected} = 0;
	$p{queue} = [];
	$p{chan} = {};
	$p{json} = JSON::XS->new->ascii->allow_nonref;
	my $me = bless \%p, $class;
	$me->_poll;
	$me;
}

sub subscribe
{
	my ($me, @chan) = @_;
	for my $chan (@chan) {
		$me->{chan}{$chan} = 1;
	}
	$me->_need_connect;
	return 0 unless $me->{connected};
	for my $chan (@chan) {
		syswrite $me->{sock}, "subscribe $chan\n";
	}
	return 1;
}

sub msg
{
	my ($me, $chan, $msg) = @_;
	$me->_need_connect;
	return 0 unless $me->{connected};
	my $json = $me->{json}->encode($msg);
	syswrite $me->{sock}, "put $chan $json\n";
	return 1;
}

sub get
{
	my ($me, @chan) = @_;

	my $classify = 0;
	my $want_one = !wantarray;
	my $timeout = 0;
	if (@chan && $chan[0] =~ /^$RE{num}{real}$/ && $chan[0] >= 0) {
		$timeout = shift @chan;
	}
	if (@chan == 1 && ref $chan[0]) {
		@chan = @{$chan[0]};
		$classify = 1;
	}

	my @r = $me->_scan(want_one => $want_one, classify => $classify, chan => \@chan);
	if (@r) {
		return @r unless $want_one;
		return $r[0] if defined $r[0];
	}

	$me->_poll($timeout);
	return $me->_scan(want_one => $want_one, classify => $classify, chan => \@chan);
}

sub _scan
{
	my ($me, %p) = @_;

	my (@r, $r, $v, $found);
	unless (@{$p{chan}}) {
		my @m = @{$me->{queue}};
		$me->{queue} = [];
		for my $m (@m) {
			if (!$found) {
				$v = $me->{json}->decode($m->[1]);
				$v = [$m->[0], $v] if $p{classify};
				if ($p{want_one}) {
					$r = $v;
					$found = 1;
				} else {
					push @r, $v;
				}
			} else {
				push @{$me->{queue}}, $m;
			}
		}
	} else {
		my %c = map { $_ => 1 } @{$p{chan}};
		my @m = @{$me->{queue}};
		$me->{queue} = [];
		for my $m (@m) {
			if ($c{$m->[0]} && !$found) {
				$v = $me->{json}->decode($m->[1]);
				$v = [$m->[0], $v] if $p{classify};
				if ($p{want_one}) {
					$r = $v;
					$found = 1;
				} else {
					push @r, $v;
				}
			} else {
				push @{$me->{queue}}, $m;
			}
		}
	}
	return $p{want_one} ? ($r) : @r;
}

sub _poll
{
	my ($me, $to) = @_;
	$to ||= 0;
	my $timeout = 0;
	if (!$me->{sock}) {
		eval { $me->{sock} = IPC::Messaging->tcp_client($me->{host}, $me->{port}, by_line => 1); };
		return unless $me->{sock};
	}
	while (1) {
		receive {
			got tcp_connected => $me->{sock} => then {
				$me->{connected} = 1;
				for my $chan (keys %{$me->{chan}}) {
					syswrite $me->{sock}, "subscribe $chan\n";
				}
			};
			got tcp_line => $me->{sock} => then {
				my $s = $_[1]->{line};
				$s =~ s/\r?\n?$//;
				if ($s =~ /^msg\s+(\S+)\s+(.*)$/) {
					push @{$me->{queue}}, [$1, $2];
				}
			};
			got tcp_error => $me->{sock} => then {
				$me->{connected} = 0;
				$me->{sock} = undef;
			};
			got tcp_disconnect => $me->{sock} => then {
				$me->{connected} = 0;
				$me->{sock} = undef;
			};
			after $to => then { $timeout = 1; };
		};
		$to = 0;  # XXX tricky, this
		return if $timeout;
		return unless $me->{sock};
	}
}

sub _need_connect
{
	my $me = shift;
	if ($me->{connected}) {
		$me->_poll;
	} else {
		$me->_poll(5);
	}
}

1;
__END__