view release on metacpan or  search on metacpan

lib/Amazon/DynamoDB/20120810.pm  view on Meta::CPAN

method scan (CodeRef $code,
             AttributesToGetType :$AttributesToGet,
             KeyType :$ExclusiveStartKey,
             Int :$Limit where { $_ >= 0},
             ReturnConsumedCapacityType :$ReturnConsumedCapacity,
             ScanFilterType :$ScanFilter,
             Int :$Segment where { $_ >= 0 },
             SelectType :$Select,
             TableNameType :$TableName!,
             Int :$TotalSegments where { $_ >= 1 && $_ <= 1000000 },
             Str :$FilterExpression,
             ExpressionAttributeValuesType :$ExpressionAttributeValues,
             ExpressionAttributeNamesType :$ExpressionAttributeNames,
         ) {
    my $payload = _make_payload({
                                'AttributesToGet' => $AttributesToGet,
                                'ExclusiveStartKey' => $ExclusiveStartKey,
                                'ExpressionAttributeValues' => $ExpressionAttributeValues,
                                'ExpressionAttributeNames' => $ExpressionAttributeNames,
                                'FilterExpression' => $FilterExpression,
                                'ReturnConsumedCapacity' => $ReturnConsumedCapacity,
                                'ScanFilter' => $ScanFilter,
                                'Segment' => $Segment,
                                'Select' => $Select,
                                'TableName' => $TableName,
                                'TotalSegments' => $TotalSegments
                            });

    $self->_scan_or_query_process('Scan', $payload, $code, { ResultLimit => $Limit});
}


method make_request(Str :$target,
                    HashRef :$payload,
                ) {
    my $api_version = '20120810';
    my $host = $self->host;
    my $req = HTTP::Request->new(
        POST => (($self->ssl) ? 'https' : 'http') . '://' . $self->host . ($self->port ? (':' . $self->port) : '') . '/'
    );
    $req->header( host => $host );
    # Amazon requires ISO-8601 basic format
    my $now = time;
    my $http_date = strftime('%Y%m%dT%H%M%SZ', gmtime($now));
    my $date = strftime('%Y%m%d', gmtime($now));

    $req->protocol('HTTP/1.1');
    $req->header( 'Date' => $http_date );
    $req->header( 'x-amz-target', 'DynamoDB_'. $api_version. '.'. $target );
    $req->header( 'content-type' => 'application/x-amz-json-1.0' );
    $payload = encode_json($payload);
    $req->content($payload);
    $req->header( 'Content-Length' => length($payload));
    
    if ($self->{use_iam_role}) {
        my $creds = VM::EC2::Security::CredentialCache->get();
        defined($creds) || die("Unable to retrieve IAM role credentials");
        $self->{access_key} = $creds->accessKeyId;
        $self->{secret_key} = $creds->secretAccessKey;
        $req->header('x-amz-security-token' => $creds->sessionToken);
    }        

    my $signer = AWS::Signature4->new(-access_key => $self->access_key,
                                      -secret_key => $self->secret_key);
    
    $signer->sign($req);
    return $req;
}

method _request(HTTP::Request $req) {
    $self->implementation->request($req);
}


# Since scan and query have the same type of responses share the processing.
method _scan_or_query_process (Str $target,
                               HashRef $payload,
                               CodeRef $code,
                               HashRef $args) {
    my $finished = 0;
    my $records_seen = 0;
    my $repeat = try_repeat {
        
        # Since we're may be making more than one request in this repeat loop
        # decrease our limit of results to scan in each call by the number 
        # of records remaining that the overall request wanted ot pull.
        if (defined($args->{ResultLimit})) {
            $payload->{Limit} = $args->{ResultLimit} - $records_seen;
        }

        my $req = $self->make_request(
            target => $target,
            payload => $payload,
        );
        
        $self->_process_request(
            $req,
            sub {
                my $result = shift;
                my $data = decode_json($result);
                
                for my $entry (@{$data->{Items}}) {
                    $code->(_decode_item_attributes($entry));
                }

                $records_seen += scalar(@{$data->{Items}});
                if ((defined($args->{ResultLimit}) && $records_seen >= $args->{ResultLimit})) {
                    $finished = 1;
                } 

                if (!defined($data->{LastEvaluatedKey})) {
                    $finished = 1;
                } else {
                    if (!$finished) {
                        $payload->{ExclusiveStartKey} = $data->{LastEvaluatedKey};                    
                    }
                }