File Coverage

blib/lib/Beanstalk/Client.pm

Criterion	Covered	Total	%
statement	47	257	18.2
branch	6	160	3.7
condition	3	30	10.0
subroutine	15	42	35.7
pod	27	27	100.0
total	98	516	18.9

package Beanstalk::Client;

use strict;
use warnings;

use base qw(Class::Accessor::Fast);

use YAML::Syck;
use Socket;
use IO::Socket::INET;

use Beanstalk::Job;
use Beanstalk::Stats;

our $VERSION = "1.07";

# use namespace::clean;

our $CRLF = "\015\012";
our $MSG_NOSIGNAL = eval { Socket::MSG_NOSIGNAL() } || 0;

BEGIN {
  __PACKAGE__->mk_accessors(
    qw(
      connect_timeout
      debug
      decoder
      default_tube
      delay
      encoder
      error
      priority
      server
      socket
      ttr
      _watching
      _using
      )
  );
}

# no namespace::clean;

sub _interact {
  my ($self, $cmd, $data) = @_;
  my $sock = $self->socket || $self->connect
    or return;

  local $SIG{PIPE} = "IGNORE" unless $MSG_NOSIGNAL;

  my $debug = $self->debug;
  warn $cmd ."\n" if $debug;

  $cmd .= $CRLF;
  $cmd .= $data . $CRLF if defined $data;

  my $offset = 0;
WRITE: {
    my $sent = send($sock, substr($cmd, $offset), $MSG_NOSIGNAL);
    if ($sent) {
      $offset += $sent;
      redo WRITE if $offset < length($cmd);
    }
    else {
      $self->error("$!");
      redo WRITE if $!{EINTR} and fileno($sock);
      return $self->disconnect;
    }
  }

  my $buffer;
  $offset = 0;
READ: {
    my $read = sysread($sock, $buffer, 1024, $offset);
    if ($read) {
      if ($buffer =~ /^([^\015\012]+)\015\012/) {
        $self->{_recv_buffer} = substr($buffer, 2 + length($1));
        warn $1,"\n" if $debug;
        return split(' ', $1);
      }
      $offset += length $buffer;
      redo READ;
    }
    else {
      $self->error("$!");
      redo READ if $!{EINTR} and fileno($sock);
    }
  }
  $self->disconnect;
  return;
}


sub _recv_data {
  my ($self, $bytes) = @_;
  my $sock = $self->socket;

  my $need   = $bytes + 2;                      # count CRLF
  my $offset = length($self->{_recv_buffer});
  my $more   = $need - $offset;

READ: while ($more > 0) {
    my $read = sysread($sock, $self->{_recv_buffer}, $more, $offset);
    if ($read) {
      $offset += $read;
      $more -= $read;
      last if $more == 0;
      redo READ;
    }
    else {
      redo READ if $!{EINTR};
      $self->error("$!");
      return $self->disconnect;
    }
  }
  warn substr($self->{_recv_buffer}, 0, $bytes),"\n" if $self->debug;
  return substr($self->{_recv_buffer}, 0, $bytes);
}


sub _interact_yaml_resp {
  my ($self, $cmd) = @_;

  my @resp = _interact($self, $cmd)
    or return;

  if ($resp[0] eq 'OK') {
    my $data = _recv_data($self, $resp[1])
      or return undef;
    return YAML::Syck::Load($data);
  }

  $self->error(join ' ', @resp);
  return undef;
}


sub _interact_stats {
  my $ret = _interact_yaml_resp(@_)
    or return undef;
  return Beanstalk::Stats->new($ret);
}


sub _peek {
  my $self = shift;
  my $cmd  = shift;

  my @resp = _interact($self, $cmd)
    or return undef;

  if ($resp[0] eq 'FOUND') {
    my $data = _recv_data($self, $resp[2]);
    return undef unless defined $data;
    return Beanstalk::Job->new(
      { id     => $resp[1],
        client => $self,
        data   => $data,
      }
    );
  }

  $self->error(join ' ', @resp);
  return undef;
}

sub __watching {
  my $self = shift;
  my $watching = $self->_watching;
  return $watching if $watching;
  $self->list_tubes_watched;
  $self->_watching;
}

# use namespace::clean;

sub new {
  my $proto  = shift;
  my $fields = shift || {};
  my $self   = $proto->SUPER::new(
    { delay    => 0,
      ttr      => 120,
      priority => 10_000,
      encoder  => \&YAML::Syck::Dump,
      decoder  => \&YAML::Syck::Load,
      %$fields,
    }
  );
  $self->{_recv_buffer} = '';
  $self;
}


sub connect {
  my $self   = shift;
  my $server = $self->server || "127.0.0.1";

  $server .= ":11300" unless $server =~ /:/;

  my $timeout = $self->connect_timeout;

  my $sock = IO::Socket::INET->new(
    PeerAddr => $server,
    Timeout  => $timeout,
  );

  unless ($sock) {
    $self->error("connect: $@");
    return $self->disconnect;
  }

  $self->socket($sock);

  my $was_watching = $self->_watching;
  my $was_using = $self->_using;

  $self->list_tubes_watched;
  if ($was_watching) {
    $self->watch_only(keys %$was_watching)
      or return $self->disconnect;
  }
  elsif (my $default_tube = $self->default_tube) {
    $self->use($default_tube) && $self->watch_only($default_tube)
      or return $self->disconnect;
  }

  if (defined $was_using) {
    $self->use($was_using)
      or return $self->disconnect;
  }

  $sock;
}


sub disconnect {
  my $self = shift;
  if (my $sock = $self->socket) {
    close($sock);
  }
  $self->socket(undef);
}

sub quit {
  shift->disconnect;
  return 1;
}


sub put {
  my $self = shift;
  my $opt  = shift || {};

  my $pri   = exists $opt->{priority} ? $opt->{priority} : $self->priority;
  my $ttr   = exists $opt->{ttr}      ? $opt->{ttr}      : $self->ttr;
  my $delay = exists $opt->{delay}    ? $opt->{delay}    : $self->delay;
  my $data  = exists $opt->{data}     ? $opt->{data}     : $self->encoder->(@_);

  utf8::encode($data) if utf8::is_utf8($data);    # need bytes

  my $bytes = length($data);

  my @resp = _interact($self, "put $pri $delay $ttr $bytes", $data)
    or return undef;

  if ($resp[0] =~ /( INSERTED | BURIED )/x) {
    return Beanstalk::Job->new(
      { id     => $resp[1],
        client => $self,
        buried => $1 eq 'BURIED' ? 1 : 0,
        data   => $data,
      }
    );
  }

  $self->error(join ' ', @resp);

  return undef;
}


sub stats {
  my $self = shift;
  _interact_stats($self, "stats");
}


sub stats_tube {
  my $self = shift;
  my $tube = @_ ? shift: 'default';
  _interact_stats($self, "stats-tube $tube");
}


sub stats_job {
  my $self = shift;
  my $id = shift || 0;
  _interact_stats($self, "stats-job $id");
}


sub kick {
  my $self  = shift;
  my $bound = shift || 1;

  my @resp = _interact($self, "kick $bound")
    or return undef;

  return $resp[1] if $resp[0] eq 'KICKED';

  $self->error(join ' ', @resp);
  return undef;
}


sub kick_job {
  my $self  = shift;
  my $job   = shift;

  my @resp = _interact($self, "kick-job $job")
    or return undef;

  return 1 if $resp[0] eq 'KICKED';

  $self->error(join ' ', @resp);
  return undef;
}


sub use {
  my $self = shift;
  my $tube = shift;

  my @resp = _interact($self, "use $tube")
    or return undef;

  return $self->_using($resp[1]) if $resp[0] eq 'USING';

  $self->error(join ' ', @resp);
  return undef;
}


sub reserve {
  my $self    = shift;
  my $timeout = shift;

  my $cmd     = defined($timeout) ? "reserve-with-timeout $timeout" : "reserve";
  my @resp    = _interact($self, $cmd)
    or return undef;

  if ($resp[0] eq 'RESERVED') {
    my $data = _recv_data($self, $resp[2]);
    return undef unless defined $data;

    return Beanstalk::Job->new(
      { id       => $resp[1],
        client   => $self,
        reserved => 1,
        data     => $data,
      }
    );
  }

  $self->error(join ' ', @resp);
  return undef;
}


sub delete {
  my $self = shift;
  my $id   = shift;
  my @resp = _interact($self, "delete $id")
    or return undef;
  return 1 if $resp[0] eq 'DELETED';

  $self->error(join ' ', @resp);
  return undef;
}

sub touch {
  my $self = shift;
  my $id   = shift;
  my @resp = _interact($self, "touch $id")
    or return undef;
  return 1 if $resp[0] eq 'TOUCHED';

  $self->error(join ' ', @resp);
  return undef;
}


sub release {
  my $self = shift;
  my $id   = shift;
  my $opt  = shift || {};

  my $pri   = exists $opt->{priority} ? $opt->{priority} : $self->priority;
  my $delay = exists $opt->{delay}    ? $opt->{delay}    : $self->delay;

  my @resp = _interact($self, "release $id $pri $delay")
    or return undef;
  return 1 if $resp[0] eq 'RELEASED';

  $self->error(join ' ', @resp);
  return undef;
}


sub bury {
  my $self = shift;
  my $id   = shift;
  my $opt  = shift || {};

  my $pri = exists $opt->{priority} ? $opt->{priority} : $self->priority;

  my @resp = _interact($self, "bury $id $pri")
    or return undef;
  return 1 if $resp[0] eq 'BURIED';

  $self->error(join ' ', @resp);
  return undef;
}


sub watch {
  my $self = shift;
  my $tube = shift;

  my $watching = $self->__watching or return undef;
  return scalar keys %$watching if $watching->{$tube};

  my @resp = _interact($self, "watch $tube")
    or return undef;

  if ($resp[0] eq 'WATCHING') {
    $watching->{$tube}++;
    return $resp[1];
  }

  $self->error(join ' ', @resp);
  return undef;
}


sub ignore {
  my $self = shift;
  my $tube = shift;

  my $watching = $self->__watching or return undef;
  return scalar keys %$watching unless $watching->{$tube};

  my @resp = _interact($self, "ignore $tube")
    or return undef;

  if ($resp[0] eq 'WATCHING') {
    delete $watching->{$tube};
    return $resp[1];
  }

  $self->error(join ' ', @resp);
  return undef;
}


sub watch_only {
  my $self = shift;
  my $watching = $self->__watching or return undef;
  my %watched = %$watching;
  my $ret;
  foreach my $watch (@_) {
    next if delete $watched{$watch};
    $ret = $self->watch($watch) or return undef;
  }
  foreach my $ignore (keys %watched) {
    $ret = $self->ignore($ignore) or return undef;
  }
  return $ret || scalar keys %$watching;
}


sub peek         { _peek($_[0], "peek $_[1]") }
sub peek_ready   { _peek(shift, "peek-ready") }
sub peek_delayed { _peek(shift, "peek-delayed") }
sub peek_buried  { _peek(shift, "peek-buried") }


sub list_tubes {
  my $self = shift;
  my $ret = _interact_yaml_resp($self, "list-tubes")
    or return undef;
  return @$ret;
}


sub list_tube_used {
  my $self = shift;
  my @resp = _interact($self, "list-tube-used")
    or return undef;
  return $resp[1] if $resp[0] eq 'USING';

  $self->error(join ' ', @resp);
  return undef;
}


sub list_tubes_watched {
  my $self = shift;
  my $ret = _interact_yaml_resp($self, "list-tubes-watched")
    or return;
  $self->_watching( { map { ($_,1) } @$ret });
  @$ret;
}


sub pause_tube {
  my $self = shift;
  my $tube = shift;
  my $delay= shift || 0;
  my @resp = _interact($self, "pause-tube $tube $delay")
    or return undef;
  return 1 if $resp[0] eq 'PAUSED';

  $self->error(join ' ', @resp);
  return undef;
}

1;

__END__