File Coverage

lib/Bloomd/Client.pm

Criterion	Covered	Total	%
statement	69	103	66.9
branch	13	36	36.1
condition	2	5	40.0
subroutine	17	26	65.3
pod	12	12	100.0
total	113	182	62.0

#
# This file is part of Bloomd-Client
#
# This software is copyright (c) 2013 by Damien "dams" Krotkine.
#
# This is free software; you can redistribute it and/or modify it under
# the same terms as the Perl 5 programming language system itself.
#
package Bloomd::Client;
{
  $Bloomd::Client::VERSION = '0.27';
}

# ABSTRACT: Perl client to the bloomd server

use feature ':5.10';
use Moo;
use List::MoreUtils qw(any mesh);
use Carp;
use Socket qw(:crlf);
use IO::Socket::INET;
use Errno qw(:POSIX);
use POSIX qw(strerror);
use Config;
use Types::Standard -types;


has protocol => ( is => 'ro', default => sub {'tcp'} );


has host => ( is => 'ro', isa => StrMatch[qr/.+/], default => sub {'127.0.0.1'} );


has port => ( is => 'ro', isa => Int, default => sub {8673} );
has _socket => ( is => 'lazy', predicate => 1, clearer => 1 );


has timeout => ( is => 'ro', , default => sub { 10 } );

has '_pid' => (is => 'ro', lazy => 1, clearer => 1, default => sub { $$ });


sub _build__socket {
    my ($self) = @_;
    my $socket = IO::Socket::INET->new(
        Proto => $self->protocol,
        PeerHost => $self->host,
        PeerPort => $self->port,
        Timeout  => $self->timeout,
    ) or die "Can't connect to server: $!";

    $self->timeout
      or return $socket;

    $Config{osname} eq 'netbsd' || $Config{osname} eq 'solaris'
      and croak "the timeout option is not yet supported on NetBSD or Solaris";

    my $seconds  = int( $self->timeout );
    my $useconds = int( 1_000_000 * ( $self->timeout - $seconds ) );
    my $timeout  = pack( 'l!l!', $seconds, $useconds );

    $socket->setsockopt( SOL_SOCKET, SO_RCVTIMEO, $timeout )
      or croak "setsockopt(SO_RCVTIMEO): $!";

    $socket->setsockopt( SOL_SOCKET, SO_SNDTIMEO, $timeout )
      or croak "setsockopt(SO_SNDTIMEO): $!";

    return $socket;

}


sub disconnect {
    my ($self) = @_;
    $self->_has_socket
      and $self->_socket->close;
    $self->_clear_socket;
}


sub create {
    my ($self, $name, $capacity, $prob, $in_memory) = @_;
    my $args =
        ( $capacity ? "capacity=$capacity" : '' )
      . ( $prob ? " prob=$prob" : '' )
      . ( $in_memory ? " in_memory=$in_memory" : '' );
    $self->_execute("create $name $args" ) eq 'Done';
}


sub list {
    my ($self, $prefix) = @_;
    $prefix //= '';
    my @keys = qw(name prob size capacity items);
    [
     map {
         my @values = split / /;
         +{ mesh @keys, @values };
     }
     $self->_execute("list $prefix" )
    ];
}


sub drop {
    my ($self, $name) = @_;
    $self->_execute("drop $name") eq 'Done';
}



sub close {
    my ($self, $name) = @_;
    $self->_execute("close $name") eq 'Done';
}


sub clear {
    my ($self, $name) = @_;
    $self->_execute("clear $name") eq 'Done';
}


sub check {
    my ($self, $name, $key) = @_;
    $self->_execute("c $name $key") eq 'Yes';
}


sub multi {
    my ($self, $name, @keys) = @_;
    @keys
      or return {};
    my @values = map { $_ eq 'Yes' } split / /, $self->_execute("m $name @keys");
    +{mesh @keys, @values };
}


sub set {
    my ($self, $name, $key) = @_;
    $self->_execute("s $name $key") eq 'Yes';
}


sub bulk {
    my ($self, $name, @keys) = @_;
    @keys
      or return;
    $self->_execute("b $name @keys");
    return;
}


sub info {
    my ($self, $name) = @_;
    +{ map { split / / } $self->_execute("info $name") };
}


sub flush {
    my ($self, $name) = @_;
    $self->_execute("info $name") eq "Done";
}

sub _execute {
    my ($self, $command) = @_;
	if ($self->_pid ne $$) {
		$self->_clear_socket;
		$self->_clear_pid;
	}
    my $socket = $self->_socket;

    local $\;
    $socket->print($command . $CRLF)
      or croak "couldn't write to socket";

    my $line = $self->_check_line($socket->getline);
    $line =~ /^Client Error:/
      and croak "$line: $command";

    $line eq 'START'
      or return $line;

    my @lines;
    while (1) {
        $line = $self->_check_line($socket->getline);
        $line eq 'END'
          and last;
        push @lines, $line;
    }

    return @lines;
}

sub _check_line {
    my ($self, $line) = @_;
    if (!defined $line) {
        my $e = $!;
        if (any { $_ } ( $!{EWOULDBLOCK}, $!{EAGAIN}, $!{ETIMEDOUT} )) {
            $e = strerror(ETIMEDOUT);
            $self->disconnect;
        }
        undef $!;
        croak $e;
    }
    $line =~ s/$CR?$LF?$//;
    return $line;
}

1;

__END__