File Coverage

blib/lib/Memcached/Client/Connection.pm

Criterion	Covered	Total	%
statement	16	80	20.0
branch	0	24	0.0
condition	0	5	0.0
subroutine	6	17	35.2
pod	8	8	100.0
total	30	134	22.3

package Memcached::Client::Connection;
BEGIN {
  $Memcached::Client::Connection::VERSION = '2.01';
}
# ABSTRACT: Class to manage Memcached::Client server connections

use strict;
use warnings;
use AnyEvent qw{};
use AnyEvent::Handle qw{};
use Memcached::Client::Log qw{DEBUG LOG};


sub new {
    my ($class, $server, $protocol) = @_;
    die "You must give me a server to connect to.\n" unless ($server);
    die "You must give me a protocol to use.\n" unless ($protocol);
    $server .= ":11211" unless 0 < index $server, ':';
    my $self = {attempts => 0, protocol => $protocol, queue => [], server => $server};
    bless $self, $class;
}


sub log {
    my ($self, $format, @args) = @_;
    LOG ("Connection/%s> " . $format, $self->{server}, @args);
}


sub connect {
    my ($self) = @_;
    unless ($self->{handle}) {
        $self->log ("Initiating connection", $self->{server}) if DEBUG;
        $self->{handle} = AnyEvent::Handle->new (connect => [split /:/, $self->{server}],
                                                 keepalive => 1,
                                                 on_connect => sub {
                                                     $self->log ("Connected") if DEBUG;
                                                     $self->{attempts} = 0;
                                                     $self->dequeue;
                                                 },
                                                 on_error => sub {
                                                     my ($handle, $fatal, $message) = @_;
                                                     $self->log ("Removing handle") if DEBUG;
                                                     delete $self->{handle};
                                                     if ($message eq "Broken pipe") {
                                                         $self->log ("Broken pipe, reconnecting") if DEBUG;
                                                         $self->connect;
                                                     } elsif ($message eq "Connection timed out" and ++$self->{attempts} < 5) {
                                                         $self->log ("Connection timed out, retrying") if DEBUG;
                                                         $self->connect;
                                                     } else {
                                                         $self->log ("Error %s, failing", $message) if DEBUG;
                                                         $self->fail;
                                                     }
                                                 },
                                                 on_prepare => sub {
                                                     my ($handle) = @_;
                                                     $self->log ("Preparing handle") if DEBUG;
                                                     $self->{protocol}->prepare_handle ($handle) if ($self->{protocol}->can ("prepare_handle"));
                                                     return $self->{connect_timeout} || 0.5;
                                                 });
    }
}


sub disconnect {
    my ($self) = @_;

    $self->log ("Disconnecting") if DEBUG;
    if (my $handle = delete $self->{handle}) {
        $self->log ("Shutting down handle") if DEBUG;
        $handle->destroy();
    }

    $self->log ("Failing all requests") if DEBUG;
    $self->fail;
}


sub enqueue {
    my ($self, $request) = @_;
    $self->log ("Request is %s", $request) if DEBUG;
    push @{$self->{queue}}, $request;
    $self->dequeue;
}


sub dequeue {
    my ($self) = @_;
    if ($self->{handle}) {
        return if ($self->{executing});
        if ($self->{executing} = shift @{$self->{queue}}) {
            $self->log ("Initiating request") if DEBUG;
            my $command = $self->{executing}->{type};
            $self->{protocol}->$command ($self, $self->{executing});
        }
    } else {
        $self->connect;
    }
}


sub complete {
    my ($self) = @_;
    $self->log ("Done with request") if DEBUG;
    delete $self->{executing};
    goto &dequeue;
}


sub fail {
    my ($self) = @_;
    $self->log ("Checking for executing request") if DEBUG;
    if (my $executing = delete $self->{executing}) {
        $self->log ("Failing executing request %s", $executing) if DEBUG;
        $executing->result;
    }
    $self->log ("Failing requests in queue: %s", $self->{queue}) if DEBUG;
    while (my $request = shift @{$self->{queue}}) {
        $self->log ("Failing request %s", $request) if DEBUG;
        $request->result;
    }
}

1;

__END__