view release on metacpan or  search on metacpan

lib/Cassandra/Client.pm  view on Meta::CPAN

package Cassandra::Client;
our $AUTHORITY = 'cpan:TVDW';
$Cassandra::Client::VERSION = '0.21';
# ABSTRACT: Perl library for accessing Cassandra using its binary network protocol

use 5.010;
use strict;
use warnings;

use Cassandra::Client::AsyncAnyEvent;
use Cassandra::Client::AsyncEV;
use Cassandra::Client::Config;
use Cassandra::Client::Connection;
use Cassandra::Client::Metadata;
use Cassandra::Client::Policy::Queue::Default;
use Cassandra::Client::Policy::Retry::Default;
use Cassandra::Client::Policy::Retry;
use Cassandra::Client::Policy::Throttle::Default;
use Cassandra::Client::Policy::LoadBalancing::Default;
use Cassandra::Client::Pool;
use Cassandra::Client::TLSHandling;
use Cassandra::Client::Util qw/series whilst/;

use Clone 0.36 qw/clone/;
use List::Util qw/shuffle/;
use AnyEvent::XSPromises qw/deferred/;
use Time::HiRes ();
use Ref::Util 0.008 qw/is_ref/;
use Devel::GlobalDestruction 0.11;
use XSLoader;

our $XS_VERSION = ($Cassandra::Client::VERSION || '');
$XS_VERSION =~ s/\A(\d+)\.(\d+)(\d{3})\z/$1.$2_$3/;
XSLoader::load(__PACKAGE__, $XS_VERSION);

sub new {
    my ($class, %args)= @_;

    my $self= bless {
        connected         => 0,
        connect_callbacks => undef,
        shutdown          => 0,

        active_queries    => 0,
    }, $class;

    my $options= Cassandra::Client::Config->new(
        \%args
    );

    $self->{throttler}= $options->{throttler} || Cassandra::Client::Policy::Throttle::Default->new();
    $self->{retry_policy}= $options->{retry_policy} || Cassandra::Client::Policy::Retry::Default->new();
    $self->{command_queue}= $options->{command_queue} || Cassandra::Client::Policy::Queue::Default->new();
    $self->{load_balancing_policy}= $options->{load_balancing_policy} || Cassandra::Client::Policy::LoadBalancing::Default->new();

    my $async_class= $options->{anyevent} ? "Cassandra::Client::AsyncAnyEvent" : "Cassandra::Client::AsyncEV";
    my $async_io= $async_class->new(
        options => $options,
    );
    my $metadata= Cassandra::Client::Metadata->new(
        options => $options,
    );
    my $pool= Cassandra::Client::Pool->new(
        client   => $self,
        options  => $options,
        metadata => $metadata,
        async_io => $async_io,
        load_balancing_policy => $self->{load_balancing_policy},
    );
    my $tls= $options->{tls} ? Cassandra::Client::TLSHandling->new() : undef;

    $self->{options}= $options;
    $self->{async_io}= $async_io;
    $self->{metadata}= $metadata;
    $self->{pool}= $pool;
    $self->{tls}= $tls;

    return $self;
}

sub _connect {
    my ($self, $callback)= @_;
    return _cb($callback) if $self->{connected};
    return _cb($callback, 'Cannot connect: shutdown() has been called') if $self->{shutdown};

    # This is ONLY useful if the user doesn't throw away the C::C object on connect errors.
    if (!$self->{connecting} && (my $error= $self->{throttler}->should_fail())) {
        return _cb($callback, $error);
    }

    push @{$self->{connect_callbacks}||=[]}, $callback;
    if ($self->{connecting}++) {
        return;
    }

    my @contact_points= shuffle @{$self->{options}{contact_points}};
    my $last_error= "No hosts to connect to";

    my $next_connect;
    $next_connect= sub {
        my $contact_point= shift @contact_points;
        if (!$contact_point) {
            delete $self->{connecting};
            undef $next_connect;
            _cb($_, "Unable to connect to any Cassandra server. Last error: $last_error") for @{delete $self->{connect_callbacks}};
            return;
        }

        my $connection= Cassandra::Client::Connection->new(
            client => $self,
            options => $self->{options},
            host => $contact_point,
            async_io => $self->{async_io},
            metadata => $self->{metadata},
        );

        series([
            sub {
                my ($next)= @_;
                $connection->connect($next);
            },
            sub {
                my ($next)= @_;
                $self->{pool}->init($next, $connection);
            },
        ], sub {
            my $error= shift;
            $self->{throttler}->count($error);
            if ($error) {
                $last_error= "On $contact_point: $error";
                return $next_connect->();
            }

            undef $next_connect;
            $self->{connected}= 1;
            delete $self->{connecting};
            _cb($_) for @{delete $self->{connect_callbacks}};
        });
    };
    $next_connect->();

    return;
}

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

    return if $self->{shutdown};
    $self->{shutdown}= 1;
    $self->{connected}= 0;

    $self->{pool}->shutdown;

    return;
}

sub is_active {
    my ($self)= @_;
    return 0 unless $self->{connected};
    return 1;
}

sub _disconnected {
    my ($self, $connid)= @_;
    $self->{pool}->remove($connid);
    return;
}

sub _handle_topology_change {
    my ($self, $change, $ipaddress)= @_;
    if ($change eq 'NEW_NODE') {
        $self->{pool}->event_added_node($ipaddress);
    } elsif ($change eq 'REMOVED_NODE') {
        $self->{pool}->event_removed_node($ipaddress);
    } else {
        warn "Received unknown topology change: $change for $ipaddress";
    }
}

sub _handle_status_change {
    my ($self, $change, $ipaddress)= @_;
    # XXX Ignored, for now
    $self->{pool}->connect_if_needed;
}



# Query functions
sub _prepare {
    my ($self, $callback, $query)= @_;

    # Fast path: we're already done
    if ($self->{metadata}->is_prepared(\$query)) {
        return _cb($callback);
    }

    $self->_command("prepare", $callback, [ $query ]);

lib/Cassandra/Client.pm  view on Meta::CPAN

        [ "INSERT INTO my_table (a, b) VALUES (?, ?)", [ $row1_a, $row1_b ] ],
        [ "INSERT INTO my_table (a, b) VALUES (?, ?)", [ $row2_a, $row2_b ] ],
    ], { batch_type => "unlogged" });

=item $client->execute($query[, $bound_parameters[, $attributes]])

Executes a single query on Cassandra, and fetch the results (if any).

For queries that have large amounts of result rows and end up spanning multiple pages, C<each_page> is the function you need. C<execute> does not handle pagination, and may end up missing rows unless pagination is implemented by its user through the ...

    $client->execute(
        "UPDATE my_table SET column=:new_column WHERE id=:id",
        { new_column => 2, id => 5 },
        { consistency => "quorum" },
    );

The C<idempotent> attribute indicates that the query is idempotent and may be retried without harm.

=item $client->each_page($query, $bound_parameters, $attributes, $page_callback)

Executes a query and invokes C<$page_callback> with each page of the results, represented as L<Cassandra::Client::ResultSet> objects.

    # Downloads the entire table from the database, even if it's terabytes in size
    $client->each_page( "SELECT id, column FROM my_table", undef, undef, sub {
        my $page= shift;
        for my $row (@{$page->rows}) {
            say $row->[0];
        }
    });

=item $client->prepare($query)

Prepares a query on the server. C<execute> and C<each_page> already do this internally, so this method is only useful for preloading purposes (and to check whether queries even compile, I guess).

=item $client->shutdown()

Disconnect all connections and abort all current queries. After this, the C<Cassandra::Client> object considers itself shut down and must be reconstructed with C<new()>.

=item $client->wait_for_schema_agreement()

Wait until all nodes agree on the schema version. Useful after changing table or keyspace definitions.

=back

=head1 (A?)SYNCHRONOUS

It's up to the user to choose which calling style to use: synchronous, asynchronous with promises, or through returned coderefs.

=head2 Synchronous

All C<Cassandra::Client> methods are available as synchronous methods by using their normal names. For example, C<< $client->connect(); >> will block until the client has connected. Similarly, C<< $client->execute($query) >> will wait for the query r...

    my $client= Cassandra::Client->new( ... );
    $client->connect;
    $client->execute("INSERT INTO my_table (id, value) VALUES (?, ?) USING TTL ?",
        [ 1, "test", 86400 ],
        { consistency => "quorum" });

=head2 Promises

C<Cassandra::Client> methods are also available as promises (see perldoc L<AnyEvent::XSPromises>). This integrates well with other libraries that deal with promises or asynchronous callbacks. Note that for promises to work, C<AnyEvent> is required, a...

Promise variants are available by prefixing method names with C<async_>, eg. C<async_connect>, C<async_execute>, etc. The usual result of the method is passed to the promise's success handler, or to the failure handler if there was an error.

    # Asynchronously pages through the result set, processing data as it comes in.
    my $promise= $client->async_each_page("SELECT id, column FROM my_table WHERE id=?", [ 5 ], undef, sub {
        for my $row (@{shift->rows}) {
            my ($id, $column)= @$row;
            say "Row: $id $column";
        }
    })->then(sub {
        say "We finished paging through all the rows";
    }, sub {
        my $error= shift;
    });

Promises normally get resolved from event loops, so for this to work you need one. Normally you would deal with that by collecting all your promises and then waiting for that :

    use AnyEvent::XSPromises qw/collect/;
    use AnyEvent;

    my @promises= ( ... ); # See other examples
    my $condvar= AnyEvent->condvar;

    collect(@promises)->then(sub {
        $condvar->send;
    }, sub {
        my $error= shift;
        warn "Unhandled error! $error";
        $condvar->send;
    });
    $condvar->recv; # Wait for the promsie to resolve or fail

How you integrate this into your infrastructure is of course up to you, and beyond the scope of the C<Cassandra::Client> documentation.

=head2 Coderefs

These are the simplest form of asynchronous querying in C<Cassandra::Client>. Instead of dealing with complex callback resolution, the client simply returns a coderef that, once invoked, returns what the original method would have retruned.

The variants are available by prefixing method names with C<future_>, eg. C<future_connect>, C<future_execute>, etc. These methods return a coderef.

    my $coderef= $client->future_execute("INSERT INTO table (id, value) VALUES (?, ?), [ $id, $value ]);

    # Do other things
    ...

    # Wait for the query to finish
    $coderef->();

Upon errors, the coderef will die, just like the synchronous methods would. Because of this, invoking the coderef immediately after getting it is equal to using the synchronous methods :

    # This :
    $client->connect;

    # Is the same as :
    $client->future_connect->();

When used properly, coderefs can give a modest performance boost, but their real value is in the ease of use compared to promises.

=head1 CAVEATS, BUGS, TODO