view release on metacpan or  search on metacpan

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

  $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 {
    my ( $server, $result, $error ) = $future->get();

    if ( !$error ) {
      $params->{cb}->( $result, undef );
    }
    else {
      $params->{cb}->( undef, {
        server_name => $server->{name},
        server_id   => $server->{id},
        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,
    },
  );

  return $params;
}

1;

=head1 NAME

AnyEvent::Pg::Pool::Multiserver - Asyncronious multiserver requests to Postgresql with AnyEvent::Pg

=head1 SYNOPSIS

  my $servers = [
    {
      id   => 1,
      name => 'remote 1',
      conn => 'host=remote1 port=5432 dbname=mydb user=myuser password=mypass',
    },
    {
      id   => 2,
      name => 'remote 2',
      conn => 'host=remote2 port=5432 dbname=mydb user=myuser password=mypass',
    },
  ];
  my $pool = AnyEvent::Pg::Pool::Multiserver->new( servers => $servers, local => 1 );

  # multi-server request

  $pool->selectall_arrayref(
    query  => 'SELECT val FROM ( SELECT 1 AS val ) tmp WHERE tmp.val = $1;',
    args   => [ 1 ],
    cb     => sub {
      my $results = shift;
      my $errors  = shift;

      if ( $errors ) {
        foreach my $srv ( @$errors ) {
          say "err $srv->{error} with $srv->{server_name} $srv->{server_id}";
        }
      }

      if ( $results ) {
        foreach my $val ( @$results ) {
          say "server_id=$val->{_server_id} value=$val->{val}";
        }
      }
    },
  );

  # single-server request

  $pool->selectall_arrayref(
    query     => 'SELECT val FROM ( SELECT 1 AS val ) tmp WHERE tmp.val = $1;',
    args      => [ 1 ],
    server_id => 1,
    cb        => sub { ... },
  );

  # multi-server request with sub-callbacks to some data manipulation
  # and may be to make another request to current server

  # main request | server_1 select -> ... select end -> cb_server call -> subrequests to current server | wait both   | global callback
  #              | server_2 select -> ... select end -> cb_server call -> subrequests to current server | subrequests |

  $pool->selectall_arrayref(
    query  => 'SELECT val FROM ( SELECT 1 AS val ) tmp WHERE tmp.val = $1;',
    args   => [ 1 ],
    cb     => sub { ... },
    cb_server => sub {
      my $params = { @_ };

      my $result_of_main_request = $params->{result};

      # Now we can do some sub-request to current server

      # And MUST call cb
      $params->{cb}->();
    },
  );

  # single-server request to select row in arrayref