File Coverage

blib/lib/PJob/Client.pm

Criterion	Covered	Total	%
statement	27	92	29.3
branch	3	26	11.5
condition			n/a
subroutine	8	23	34.7
pod	4	4	100.0
total	42	145	28.9

package PJob::Client;
our $VERSION = '0.41';

use Any::Moose;
use Term::ANSIColor qw/:constants/;
use Carp qw/carp croak/;
use POE qw/Component::Client::TCP/;

$| = 1;
my $yellow = YELLOW;
my $bold   = BOLD;
my $red    = RED;
my $blue   = BLUE;
my $reset  = RESET;
my $green  = GREEN;

has 'server' => (
    is       => 'rw',
    isa      => 'Str',
    required => 1,
);

has 'port' => (
    is  => 'rw',
    isa => 'Int',
);

has 'job' => (
    is      => 'rw',
    isa     => 'ArrayRef',
    default => sub { [] },
);

has '_cqueue' => (
    is      => 'rw',
    isa     => 'ArrayRef',
    default => sub { [] },
);

sub BUILD {
    my $self = shift;
    $self->{_queued} = 0;
}

sub run {
    shift->_run;
}

sub run_queue {
    my $self = shift;
    $self->{_queued} = 1;
    $self->_run;
}

sub queue_command {
    my $self = shift;
    push @{$self->_cqueue}, @_;
    return $self;
}

sub _run {
    my $self = shift;
    my $sub_for_input = sub { $self->_get_input_interactive(@_) };
    $sub_for_input = sub { $self->_get_input_queue(@_) }
      if $self->{_queued};
    my ($server, $port) = $self->_get_remote;
    $self->{_session} = POE::Component::Client::TCP->new(
        RemoteAddress => $server,
        RemotePort    => $port,
        Connected     => sub { $self->_connected(@_) },
        Disconnected  => sub { $self->_disconnected(@_) },
        ServerError   => sub { $self->_server_error(@_) },
        ServerInput   => $sub_for_input,
    );
    POE::Kernel->run();
    return $self;
}

sub _get_remote {
    my $self = shift;

    my ($server, $port) = split ':', $self->server, 2;

    if (!$self->port) {
        carp "no port specified\n" if !$port;
        $self->port($port);
    }
    return ($server, $self->port);
}

sub _connected {
    my $self = shift;
    my ($peer_addr, $peer_port) = @_[ARG1, ARG2];
    print "Connected at ${peer_addr}:${peer_port}\n";
}

sub _get_input_interactive {
    my $self  = shift;
    my $input = $_[ARG0];
    if ($input eq '.') {
        print _($bold, $yellow, ">");
        my $k = <>;
        chomp $k;
        $_[HEAP]{server}->put($k);
    }
    else {
        $self->_format_and_output($input);
    }
}

sub _get_input_queue {
    my $self = shift;
    my ($input, $heap) = @_[ARG0, HEAP];
    if ($input eq '.') {
        if (!scalar @{$self->_cqueue}) {
            $heap->{server}->put('quit');
            return;
        }
        my $command = shift @{$self->_cqueue};
        chomp $command;
        return unless $command;
        $heap->{server}->put($command);
    }
    else {
        print $input, "\n";
    }
}

sub _disconnected {
    my $self = shift;

    print "Server Disconnected, shutting down.....\n";
    $_[KERNEL]->yield('shutdown');
}

sub _server_error {
    my $self = shift;
    my ($oper, $nexit, $sexit) = @_[ARG0, ARG1, ARG2];
    return if $nexit == 0;
    print "Server Error:\n";
    print "\toperation\t$oper\n";
    print "\t   Reason\t$sexit\n";
    $_[KERNEL]->yield('shutdown');
}

sub _ {
    push @_, $reset;
    return @_;
}

sub _format_and_output {
    my ($self, $input) = @_;
    if ($input =~ /^Usage:(.*)$/) {
        print _("Usage:", $bold, $blue, $1, "\n");
        return;
    }
    elsif ($input =~ /^Job\s+(.*?)\s+:::(\d+)\sstarted\.$/) {
        print _("Job ", $bold, $green, $1, ' :::', $2, $reset, " started\n");
        return;
    }
    elsif ($input =~ /Out\t(.*)$/) {
        print _($1, "\n");
        return;
    }
    elsif ($input =~ /Err\t(.*)$/) {
        print _($yellow, '[stderr] ', $reset, $1, "\n");
        return;
    }
    elsif ($input =~ /^Job\s+(.*?)\s+:::(\d+)(.*?)(\d+)$/) {
        print _(
            "Job ", $bold, $green, $1,    ' :::', $2,
            $reset, $3,    $red,   $bold, $4,     "\n"
        );
        return;
    }
    else {
        print $input, "\n";
    }
}

no Any::Moose;
__PACKAGE__->meta->make_immutable;
1;

__END__