File Coverage

blib/lib/Coro/Amazon/SimpleDB.pm

Criterion	Covered	Total	%
statement	1	3	33.3
branch			n/a
condition			n/a
subroutine	1	1	100.0
pod			n/a
total	2	4	50.0

package Coro::Amazon::SimpleDB;
use common::sense;

$Coro::Amazon::SimpleDB::VERSION = 0.04;

use EV;
use AnyEvent;
use Coro;
use Coro::AnyEvent;

use Carp qw(croak carp);
use Scalar::Util qw(blessed);
use List::Util qw(first);

use Amazon::SimpleDB::Client;
use Amazon::SimpleDB::Model::BatchPutAttributesRequest;
use Amazon::SimpleDB::Model::CreateDomainRequest;
use Amazon::SimpleDB::Model::DeleteAttributesRequest;
use Amazon::SimpleDB::Model::DeleteDomainRequest;
use Amazon::SimpleDB::Model::DomainMetadataRequest;
use Amazon::SimpleDB::Model::GetAttributesRequest;
use Amazon::SimpleDB::Model::ListDomainsRequest;
use Amazon::SimpleDB::Model::PutAttributesRequest;
use Amazon::SimpleDB::Model::SelectRequest;



use Moose;

has 'aws_access_key' => (is => 'rw');
has 'aws_secret_access_key' => (is => 'rw');
has 'domain_name' => (is => 'rw');
has 'sdb' => (is => 'ro', lazy_build => 1);
has 'pending' => (is => 'ro', default => sub { {} });

has 'DEBUG' => (is => 'rw', default => !1);

no Moose;



REPLACE_AMAZON_SIMPLEDB_CLIENT_HTTPPOST: {
    package Amazon::SimpleDB::Client;
    use common::sense;
    use AnyEvent::HTTP;
    use HTTP::Request;
    use HTTP::Response;

    # The only mention of a time-out in Amazon::SimpleDB::Client is in
    # reference to a select operation.  I'm using the value from there
    # (5 seconds) as the default time-out for HTTP requests, as it
    # seems reasonable.  The setting is dynamic, but is used prior to
    # putting HTTP request coros into wait state so it should do the
    # expected thing if it's changed.  Caveat emptor.

    our $HTTP_REQUEST_TIMEOUT = 5;

    # Replace the _httpPost method in Amazon::SimpleDB::Client to use
    # an HTTP lib which does non-blocking requests better than
    # Coro::LWP.  This dangerously violates Amazon::SimpleDB::Client's
    # encapsulation and has some code copied from the original
    # _httpPost.  There is a chance that changes to Amazon's module
    # could break this.

    no warnings 'redefine';

    sub _httpPost {
	my ($self, $parameters) = @_;
        my $response = undef;
        http_request
            POST    => $self->{_config}{ServiceURL},
            body    => join('&', map { $_ . '=' . $self->_urlencode($parameters->{$_}, 0) } keys %{$parameters}),
            timeout => $HTTP_REQUEST_TIMEOUT,
            headers => { 'Content-Type' => 'application/x-www-form-urlencoded; charset=utf-8' },
            sub {
                my ($body, $headers) = @_;
                $response = HTTP::Response->new(@{$headers}{qw( Status Reason )});
                $response->content($body);
                $response->header($_, $headers->{$_})
                    for grep { !/[[:upper:]]/ } keys %{$headers};
            };
        # We need to put this coro to sleep until the response is returned.
        while (not defined $response) { Coro::AnyEvent::sleep 0.1 }
        return $response;
    }
}


ADD_DISPATCH_HELPER_METHODS: {
    # These methods are helpers so we can do method mapping via real
    # dispatch.  It would be nice if Amazon's library could do this
    # dispatching for us.
    sub Amazon::SimpleDB::Model::BatchPutAttributesRequest::client_request_method { 'batchPutAttributes' }
    sub Amazon::SimpleDB::Model::CreateDomainRequest::client_request_method { 'createDomain' }
    sub Amazon::SimpleDB::Model::DeleteAttributesRequest::client_request_method { 'deleteAttributes' }
    sub Amazon::SimpleDB::Model::DeleteDomainRequest::client_request_method { 'deleteDomain' }
    sub Amazon::SimpleDB::Model::DomainMetadataRequest::client_request_method { 'domainMetadata' }
    sub Amazon::SimpleDB::Model::GetAttributesRequest::client_request_method { 'getAttributes' }
    sub Amazon::SimpleDB::Model::ListDomainsRequest::client_request_method { 'listDomains' }
    sub Amazon::SimpleDB::Model::PutAttributesRequest::client_request_method { 'putAttributes' }
    sub Amazon::SimpleDB::Model::SelectRequest::client_request_method { 'select' }
}



# Debugging methods.


sub _bug_process_message {
    require Data::Dumper;
    local $Data::Dumper::Sortkeys = 1;
    local $Data::Dumper::Terse = 1;
    my $message = shift;
    my $result
        = (not defined $message)                             ? ''
        : (ref $message eq q())                              ? $message
        : (blessed($message) and $message->can('as_string')) ? scalar $message->as_string
        : blessed($message)                                  ? "$message"
        :                                                      Dumper($message)
        ;
    return $result;
}


sub _bug_message {
    my $message_array_ref = shift || [];
    my $caller_level = shift || 0;
    if (ref $message_array_ref ne 'ARRAY') {
        warn "message_array_ref is not an array ref";
        $message_array_ref = [];
    }
    $message_array_ref = [ "something is interesting" ]
        unless @{$message_array_ref};
    my $message = join((defined $, ? $, : q()), map { _bug_process_message($_) } @{$message_array_ref});
    my @caller = caller($caller_level);
    $message .= " at $caller[1] line $caller[2]\n" unless $message =~ /\n/xms;
    return $message;
}


sub bug {
    my $self = shift;
    $self->DEBUG ? print STDERR _bug_message(\@_, 1) : undef;
}



sub _build_sdb {
    my $self = shift;
    my $sdb = Amazon::SimpleDB::Client->new(
        $self->aws_access_key,
        $self->aws_secret_access_key,
    );
    return $sdb;
}


sub _normalize_sdb_request {
    my $self = shift;
    my $request = shift;

    # A scalar is interpreted as a request for an item name in the
    # canonical domain.
    return Amazon::SimpleDB::Model::GetAttributesRequest->new({
        DomainName => $self->domain_name,
        ItemName => $request,
    })
        unless ref $request;

    # A hash ref is interpreted as an argument to a call to the 'new'
    # method of the class specified by the 'RequestType' key.  This
    # key will be removed prior to call 'new', a 'DomainName' key will
    # be added if needed, and the class called may be aliased as
    # specified in the anonymous hash below.
    if (ref $request eq 'HASH') {
        # Copy the request to avoid side effects.
        my %request = (
            DomainName => $self->domain_name,
            %{$request},
        );
        my $type = delete $request{RequestType}
            or croak "missing RequestType in request";
        my $class = {
            BatchPutAttributesRequest => 'Amazon::SimpleDB::Model::BatchPutAttributesRequest',
            batchPutAttributes        => 'Amazon::SimpleDB::Model::BatchPutAttributesRequest',
            CreateDomainRequest       => 'Amazon::SimpleDB::Model::CreateDomainRequest',
            createDomain              => 'Amazon::SimpleDB::Model::CreateDomainRequest',
            DeleteAttributesRequest   => 'Amazon::SimpleDB::Model::DeleteAttributesRequest',
            deleteAttributes          => 'Amazon::SimpleDB::Model::DeleteAttributesRequest',
            DeleteDomainRequest       => 'Amazon::SimpleDB::Model::DeleteDomainRequest',
            deleteDomain              => 'Amazon::SimpleDB::Model::DeleteDomainRequest',
            DomainMetadataRequest     => 'Amazon::SimpleDB::Model::DomainMetadataRequest',
            domainMetadata            => 'Amazon::SimpleDB::Model::DomainMetadataRequest',
            GetAttributesRequest      => 'Amazon::SimpleDB::Model::GetAttributesRequest',
            getAttributes             => 'Amazon::SimpleDB::Model::GetAttributesRequest',
            ListDomainsRequest        => 'Amazon::SimpleDB::Model::ListDomainsRequest',
            listDomains               => 'Amazon::SimpleDB::Model::ListDomainsRequest',
            PutAttributesRequest      => 'Amazon::SimpleDB::Model::PutAttributesRequest',
            putAttributes             => 'Amazon::SimpleDB::Model::PutAttributesRequest',
            SelectRequest             => 'Amazon::SimpleDB::Model::SelectRequest',
            select                    => 'Amazon::SimpleDB::Model::SelectRequest',
        }->{$type} || $type;
        return $class->new(\%request);
    }

    # An Amazon::SimpleDB::Model instance is almost left alone.  The
    # only processing done is adding a DomainName if needed.  This is
    # done directly on the class, so this produces a side-effect.
    if (blessed $request and $request->isa('Amazon::SimpleDB::Model')) {
        # Amazon's class hierarchy is very unfortunate.  It would be
        # nice to handle these as a base class but that's not how it
        # was designed.
        $request->setDomainName($self->domain_name)
            if first { $request->isa("Amazon::SimpleDB::Model::$_") } qw(
                BatchPutAttributesRequest
                DeleteAttributesRequest
                GetAttributesRequest
                PutAttributesRequest
                SelectRequest
            )
            and not $request->isSetDomainName
            ;
        return $request;
    }

    croak "can't normalize '".(ref $request)."' request to an Amazon::SimpleDB::Model";
}


sub _process_request {
    my $self = shift;
    my $request = $self->_normalize_sdb_request(shift);
    my $method = $request->client_request_method
        or croak "no processing for request of type '".(ref $request)."'";
    return $self->sdb->$method($request);
}


sub add_pending {
    my $self = shift;
    $self->pending->{$_} = $_ for @_;
    return $self;
}

sub remove_pending {
    my $self = shift;
    delete $self->pending->{$_} for @_;
    return $self;
}

sub has_pending { !!%{ shift->pending } }


sub poll {
    my $self = shift;
    async {
        CHECK_LOOP: {
            # Keep polling as long as there are pending requests.
            if ($self->has_pending) {
                Coro::AnyEvent::sleep 0.1;
                redo CHECK_LOOP;
            }
            EV::unloop;
        }
    };
    EV::loop;
    return $self;
}


sub async_requests {
    my ($self, @requests) = @_;

    my $debug = $self->DEBUG;
    require Time::HiRes and Time::HiRes->import('time') if $debug;
    my ($start, $duration) = (0, 0);
    my @responses = ();
    $self->bug("starting async enqueues");
    $start = time() if $debug;
    for ($[ .. $#requests) {
        my $idx = $_;
        my $request = $requests[$idx];
        $self->bug("adding request $request");
        my $coro = async {
            my ($start, $duration) = (0, 0);
            $self->bug("starting request for $request");
            $start = time() if $debug;
            $responses[$idx] = eval { $self->_process_request($request) };
            # Store the exception instead of the response (which
            # should be undef) if there was a problem.
            $responses[$idx] = $@ if $@;
            $duration = time() - $start if $debug;
            $self->bug("completed request for $request in $duration secs");
        };
        $self->add_pending($coro);
        $coro->on_destroy(sub { $self->remove_pending($coro) });
    }
    $duration = time() - $start if $debug;
    $self->bug("completed async enqueues in $duration secs, starting coro polling");
    $self->poll;

    return \@responses;
}


sub async_get_items {
    my ($self, @items) = @_;
    my $responses = $self->async_requests(@items);
    my %items = map {
        my $item_name = $items[$_];
        my $response = $responses->[$_];
        my $attributes
            = (ref $response eq 'Amazon::SimpleDB::Model::GetAttributesResponse') ?
                  {
                      map {
                          defined $_->getName ? ($_->getName, $_->getValue) : ()
                      } @{ $response->getGetAttributesResult->getAttribute }
                  }
            :     $response
            ;
        ($item_name => $attributes);
    } $[ .. $#items;
    return \%items;
}



1;

__END__