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);