File Coverage

blib/lib/Memcached/Client.pm

Criterion	Covered	Total	%
statement	25	112	22.3
branch	0	32	0.0
condition	0	23	0.0
subroutine	9	20	45.0
pod	8	8	100.0
total	42	195	21.5

package Memcached::Client;
BEGIN {
  $Memcached::Client::VERSION = '2.01';
}
# ABSTRACT: All-singing, all-dancing Perl client for Memcached

use strict;
use warnings;
use AnyEvent qw{};
use AnyEvent::Handle qw{};
use Memcached::Client::Connection qw{};
use Memcached::Client::Log qw{DEBUG LOG};
use Memcached::Client::Request qw{};
use Module::Load qw{load};


sub new {
    my ($class, @args) = @_;
    my %args = 1 == scalar @args ? %{$args[0]} : @args;

    my $self = bless {}, $class;

    $self->log ("new: %s", \%args) if DEBUG;

    # Get all of our objects instantiated
    $self->{compressor} = $self->__class_loader (Compressor => $args{compressor} || 'Gzip')->new;
    $self->{selector} = $self->__class_loader (Selector => $args{selector} || 'Traditional')->new;
    $self->{serializer} = $self->__class_loader (Serializer => $args{serializer} || 'Storable')->new;
    $self->{protocol} = $self->__class_loader (Protocol => $args{protocol} || 'Text')->new (compressor => $self->{compressor}, serializer => $self->{serializer});

    $self->compress_threshold ($args{compress_threshold} || 10000);
    $self->hash_namespace ($args{hash_namespace} || 1);
    $self->namespace ($args{namespace} || "");
    $self->set_servers ($args{servers});
    $self->set_preprocessor ($args{preprocessor});

    $self->log ("new: done") if DEBUG;

    $self;
}


sub log {
    my ($self, $format, @args) = @_;
    LOG ("Client> " . $format, @args);
}

# This manages class loading for the sub-classes
sub __class_loader {
    my ($self, $prefix, $class) = @_;
    # Add our prefixes if the class name isn't called out as absolute
    $class = join ('::', 'Memcached::Client', $prefix, $class) if ($class !~ s/^\+//);
    # Sanitize our class name
    $class =~ s/[^\w:_]//g;
    $self->log ("loading %s", $class) if DEBUG;
    load $class;
    $class;
}


sub compress_threshold {
    my ($self, $new) = @_;
    $self->log ("compress threshold: %d", $new) if DEBUG;
    $self->{compressor}->compress_threshold ($new);
}


sub namespace {
    my ($self, $new) = @_;
    my $ret = $self->{namespace};
    $self->log ("namespace: %s", $new) if DEBUG;
    $self->{namespace} = $new if (defined $new);
    return $ret;
}


sub hash_namespace {
    my ($self, $new) = @_;
    my $ret = $self->{hash_namespace};
    $self->log ("hash namespace: %s", $new) if DEBUG;
    $self->{hash_namespace} = !!$new if (defined $new);
    return $ret;
}


sub set_preprocessor {
    my ($self, $new) = @_;
    $self->{preprocessor} = $new if (ref $new eq "CODE");
    return 1;
}


sub set_servers {
    my ($self, $servers) = @_;

    # Give the selector the list of servers first
    $self->{selector}->set_servers ($servers);

    # Shut down the servers that are no longer part of the list
    my $list = {map {(ref $_ ? $_->[0] : $_), {}} @{$servers}};
    for my $server (keys %{$self->{servers} || {}}) {
        next if (delete $list->{$server});
        $self->log ("disconnecting %s", $server) if DEBUG;
        my $connection = delete $self->{servers}->{$server};
        $connection->disconnect;
    }

    # Spawn connection handlers for all the others
    for my $server (sort keys %{$list}) {
        $self->log ("creating connection for %s", $server) if DEBUG;
        $self->{servers}->{$server} ||= Memcached::Client::Connection->new ($server, $self->{protocol});
    }

    return 1;
}


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

    $self->log ("disconnecting all") if DEBUG;
    for my $server (keys %{$self->{servers}}) {
        next unless defined $self->{servers}->{$server};
        $self->log ("disconnecting %s", $server) if DEBUG;
        $self->{servers}->{$server}->disconnect;
    }
}

# When the object leaves scope, be sure to run C to make
# certain that we shut everything down.
sub DESTROY {
    my $self = shift;
    $self->disconnect;
}


# This is really where all the action happens---where actual requests
# are submitted and handled.
#
# The routine iterates over the requests its given.  It serializes and
# compresses any data in the request as necessary.  If the request has
# a key, then it follows the keyed-submission process, preprocessing
# the key as necessary, checking its validity, mapping it to a server,
# adding a namespace, and finally submitting it.
#
# If the request has no key, it is assumed to be a broadcast request,
# so we call the ->server method on the request for each of our
# servers, to create the appropriate number of requests, and we queue
# each of them.

sub __submit {
    my ($self, @requests) = @_;
    $self->log ("Submitting request(s)") if DEBUG;
    for my $request (@requests) {
        $self->log ("Request is %s", $request) if DEBUG;
        if (defined $request->{key}) {
            if ($self->{preprocessor}) {
                if (ref $request->{key}) {
                    $request->{key}->[1] = $self->{preprocessor}->($request->{key}->[1]);
                } else {
                    $request->{key} = $self->{preprocessor}->($request->{key});
                }
            }
            if (ref $request->{key} # Pre-hashed
                ? ($request->{key}->[0] =~ m/^\d+$/ and # Hash is a decimal #
                   length $request->{key}->[1] > 0 and # Real key has a length
                   length $request->{key}->[1] <= 250 and # Real key is shorter than 250 chars
                   -1 == index $request->{key}->[1], " ") # Key contains no spaces
                : (length $request->{key} > 0 and # Real key has a length
                   length $request->{key} <= 250 and # Real key is shorter than 250 chars
                   -1 == index $request->{key}, " ") # Key contains no spaces
               ) {
                $self->log ("Finding server for key %s", $request->{key}) if DEBUG;
                my $server = $self->{selector}->get_server ($request->{key}, $self->{hash_namespace} ? $self->{namespace} : "");
                $request->{key} = ref $request->{key} ? $request->{key}->[1] : $request->{key};
                $request->{nskey} = $self->{namespace} . $request->{key};
                $self->{servers}->{$server}->enqueue ($request);
            } else {
                $self->log ("Key is invalid") if DEBUG;
                $request->result;
            }
        } else {
            $self->log ("Sending request to all servers") if DEBUG;
            for my $server (keys %{$self->{servers}} ) {
                $self->log ("Queueing for %s", $server) if DEBUG;
                $self->{servers}->{$server}->enqueue ($request->server ($server));
            }
        }
    }
}


1;

__END__