AnyEvent-Pg-Pool-Multiserver

 view release on metacpan or  search on metacpan

lib/AnyEvent/Pg/Pool/Multiserver.pm  view on Meta::CPAN

use strict;
use warnings;
use utf8;
use v5.10;

use Carp qw( croak carp );
use AnyEvent;
use AnyEvent::Pg::Pool;
use Future;
use Params::Validate qw( validate_with );

use fields qw(
  pool
  local
);

use Class::XSAccessor {
  getters => {
    local => 'local'
  },
};

sub new {
  my $class  = shift;
  my $params = {@_};

  my $self = fields::new( $class );

  $params = $self->_validate_new( $params );

  my $pool = {};

  foreach my $server ( @{ $params->{servers} } ) {
    my $dbh = AnyEvent::Pg::Pool->new(
      $server->{conn},
      connection_retries => 10,
      connection_delay   => 1,
      size               => 4,
      on_error           => sub { carp 'Some error'; },

lib/AnyEvent/Pg/Pool/Multiserver.pm  view on Meta::CPAN

    croak 'server_id must be unique' if $pool->{ $server->{id} };
    $pool->{ $server->{id} } = { dbh => $dbh, name => $server->{name}, id => $server->{id} };
  }

  $self->{pool} = $pool;
  $self->{local} = $params->{local};

  return $self;
}

sub _validate_new {
  my __PACKAGE__ $self = shift;
  my $params = shift;

  $params = validate_with(
    params => $params,
    spec => {
      servers => 1,
      local   => 1,
    },
  );

  return $params;
}

sub selectall_arrayref {
  my __PACKAGE__ $self = shift;
  my $params = {@_};

  $params = $self->_validate_selectall_arrayref( $params );

  my @futures = ();

  my @pool = ();
  if ( defined $params->{server_id} ) {
    push @pool, $params->{server_id};
  }
  else {
    @pool = keys %{ $self->{pool} };
  }

lib/AnyEvent/Pg/Pool/Multiserver.pm  view on Meta::CPAN

      scalar( @results ) ? [ @results ] : undef,
      scalar( @errors ) ? [ @errors ] : undef,
    );

    undef $main_future;
  } );

  return;
}

sub _validate_selectall_arrayref {
  my __PACKAGE__ $self = shift;
  my $params = shift;

  $params = validate_with(
    params => $params,
    spec => {
      query     => 1,
      args      => 0,
      cb        => 1,
      server_id => 0,
      cb_server => 0,
    },
  );

lib/AnyEvent/Pg/Pool/Multiserver.pm  view on Meta::CPAN

    $result = [ $id, $res->cmdRows ];
  }

  return $result;
}

sub selectrow_hashref {
  my __PACKAGE__ $self = shift;
  my $params = {@_};

  $params = $self->_validate_selectrow_hashref( $params );

  my $future = $self->_get_future_push_query(
    query     => $params->{query},
    args      => $params->{args},
    server    => $self->{pool}{ $params->{server_id} },
    cb_server => $params->{cb_server},
    type      => 'selectrow_hashref',
  );

  $future->on_done( sub {

lib/AnyEvent/Pg/Pool/Multiserver.pm  view on Meta::CPAN

        error       => $error,
      } );
    }

    undef $future;
  } );

  return;
}

sub _validate_selectrow_hashref {
  my __PACKAGE__ $self = shift;
  my $params = shift;

  $params = validate_with(
    params => $params,
    spec => {
      query     => 1,
      args      => 0,
      cb        => 1,
      server_id => 1,
      cb_server => 0,
    },
  );

  return $params;
}

sub selectrow_array {
  my __PACKAGE__ $self = shift;
  my $params = {@_};

  $params = $self->_validate_selectrow_array( $params );

  my $future = $self->_get_future_push_query(
    query     => $params->{query},
    args      => $params->{args},
    server    => $self->{pool}{ $params->{server_id} },
    cb_server => $params->{cb_server},
    type      => 'selectrow_array',
  );

  $future->on_done( sub {

lib/AnyEvent/Pg/Pool/Multiserver.pm  view on Meta::CPAN

        error       => $error,
      } );
    }

    undef $future;
  } );

  return;
}

sub _validate_selectrow_array {
  my __PACKAGE__ $self = shift;
  my $params = shift;

  $params = validate_with(
    params => $params,
    spec => {
      query     => 1,
      args      => 0,
      cb        => 1,
      server_id => 1,
      cb_server => 0,
    },
  );

  return $params;
}

sub do {
  my __PACKAGE__ $self = shift;
  my $params = {@_};

  $params = $self->_validate_do( $params );

  my $future = $self->_get_future_push_query(
    query     => $params->{query},
    args      => $params->{args},
    server    => $self->{pool}{ $params->{server_id} },
    cb_server => $params->{cb_server},
    type      => 'do',
  );

  $future->on_done( sub {

lib/AnyEvent/Pg/Pool/Multiserver.pm  view on Meta::CPAN

        error       => $error,
      } );
    }

    undef $future;
  } );

  return;
}

sub _validate_do {
  my __PACKAGE__ $self = shift;
  my $params = shift;

  $params = validate_with(
    params => $params,
    spec => {
      query     => 1,
      args      => 0,
      cb        => 1,
      server_id => 1,
      cb_server => 0,
    },
  );