EV-Pg

 view release on metacpan or  search on metacpan

lib/EV/Pg.pm  view on Meta::CPAN

    PGRES_NONFATAL_ERROR PGRES_FATAL_ERROR PGRES_COPY_BOTH
    PGRES_SINGLE_TUPLE PGRES_PIPELINE_SYNC PGRES_PIPELINE_ABORTED
    PGRES_TUPLES_CHUNK
)];

$EXPORT_TAGS{conn} = [qw(CONNECTION_OK CONNECTION_BAD)];

$EXPORT_TAGS{transaction} = [qw(
    PQTRANS_IDLE PQTRANS_ACTIVE PQTRANS_INTRANS
    PQTRANS_INERROR PQTRANS_UNKNOWN
)];

$EXPORT_TAGS{pipeline} = [qw(
    PQ_PIPELINE_OFF PQ_PIPELINE_ON PQ_PIPELINE_ABORTED
)];

$EXPORT_TAGS{verbosity} = [qw(
    PQERRORS_TERSE PQERRORS_DEFAULT PQERRORS_VERBOSE PQERRORS_SQLSTATE
)];

$EXPORT_TAGS{context} = [qw(
    PQSHOW_CONTEXT_NEVER PQSHOW_CONTEXT_ERRORS PQSHOW_CONTEXT_ALWAYS
)];

$EXPORT_TAGS{trace} = [qw(
    PQTRACE_SUPPRESS_TIMESTAMPS PQTRACE_REGRESS_MODE
)];

{
    my %seen;
    @EXPORT_OK = grep { !$seen{$_}++ } map { @$_ } values %EXPORT_TAGS;
    $EXPORT_TAGS{all} = \@EXPORT_OK;
}

*q          = \&query;
*qp         = \&query_params;
*qx         = \&query_prepared;
*prep       = \&prepare;
*reconnect  = \&reset;
*disconnect = \&finish;
*flush      = \&send_flush_request if defined &send_flush_request;
*sync       = \&pipeline_sync;
*quote      = \&escape_literal;
*quote_id   = \&escape_identifier;
*errstr     = \&error_message;
*txn_status = \&transaction_status;
*pid        = \&backend_pid;

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

    my $loop = delete $args{loop} || EV::default_loop;
    my $self = $class->_new($loop);

    $self->on_error(delete $args{on_error} // sub { die @_ });
    $self->on_connect(delete $args{on_connect})   if exists $args{on_connect};
    $self->on_notify(delete $args{on_notify})     if exists $args{on_notify};
    $self->on_notice(delete $args{on_notice})     if exists $args{on_notice};
    $self->on_drain(delete $args{on_drain})       if exists $args{on_drain};

    my $keep_alive      = delete $args{keep_alive};
    my $conninfo        = delete $args{conninfo};
    my $conninfo_params = delete $args{conninfo_params};
    my $expand_dbname   = delete $args{expand_dbname};

    if (my @unknown = sort keys %args) {
        Carp::carp("EV::Pg->new: unknown argument(s): @unknown");
    }

    $self->keep_alive(1) if $keep_alive;

    if (defined $conninfo_params) {
        $self->connect_params($conninfo_params, $expand_dbname ? 1 : 0);
    } elsif (defined $conninfo) {
        $self->connect($conninfo);
    }

    $self;
}

1;

__END__

=head1 NAME

EV::Pg - asynchronous PostgreSQL client using libpq and EV

=head1 SYNOPSIS

    use v5.10;
    use EV;
    use EV::Pg;

    my $pg = EV::Pg->new(
        conninfo   => 'dbname=mydb',
        on_error   => sub { die "PG error: $_[0]\n" },
    );
    $pg->on_connect(sub {
        $pg->query_params(
            'select $1::int + $2::int', [10, 20],
            sub {
                my ($rows, $err) = @_;
                die $err if $err;
                say $rows->[0][0];  # 30
                EV::break;
            },
        );
    });
    EV::run;

=head1 DESCRIPTION

EV::Pg is a non-blocking PostgreSQL client built on top of libpq and
the L<EV> event loop.  It drives the libpq async API (C<PQsendQuery>,
C<PQconsumeInput>, C<PQgetResult>) through C<ev_io> watchers on the
libpq socket, so the event loop never blocks on database I/O.

Features: parameterized queries, prepared statements, pipeline mode,
single-row and chunked rows (libpq E<gt>= 17), COPY IN/OUT,
LISTEN/NOTIFY, async cancel (libpq E<gt>= 17), structured error
fields, protocol tracing, and notice handling.

=head1 CALLBACKS

Query callbacks always receive a single positional argument on success
and C<(undef, $error_message)> on error, so

    my ($result, $err) = @_;

lib/EV/Pg.pm  view on Meta::CPAN

C<""> -- always an empty string (these commands return no row count).

=item describe_prepared / describe_portal

C<\%meta> -- hashref with C<nfields>, C<nparams>, and (when non-zero)
C<fields> (arrayref of C<< {name, type} >> hashes) and C<paramtypes>
(arrayref of OIDs).

=item COPY

C<"COPY_IN">, C<"COPY_OUT">, or C<"COPY_BOTH"> -- a string tag
identifying the COPY direction.

=item pipeline_sync

C<1>.

=back

Exceptions thrown inside callbacks are caught and reported via C<warn>
so that one bad callback does not derail the rest of the queue.

=head1 CONSTRUCTOR

=head2 new

    my $pg = EV::Pg->new(%args);

Returns a new EV::Pg object.  If C<conninfo> or C<conninfo_params> is
supplied, an asynchronous connect starts immediately; otherwise call
C<connect> later.

Recognized arguments:

=over

=item conninfo

libpq connection string passed to C<connect>.

=item conninfo_params

Hashref of connection parameters (e.g.
C<< { host => 'localhost', dbname => 'mydb', port => 5432 } >>),
passed to C<connect_params>.  Mutually exclusive with C<conninfo>.

=item expand_dbname

When true together with C<conninfo_params>, the C<dbname> value is
itself parsed as a connection string -- so
C<< dbname => 'postgresql://host/db?sslmode=require' >> works.

=item on_connect

Fires once with no arguments when the handshake completes.

=item on_error

Fires as C<($error_message)> on connection-level errors.  Defaults to
C<sub { die @_ }>; pass an explicit handler to keep the loop alive.

=item on_notify

Fires as C<($channel, $payload, $backend_pid)> for LISTEN/NOTIFY
messages.

=item on_notice

Fires as C<($message)> for server NOTICE/WARNING messages.

=item on_drain

Fires with no arguments when the libpq send buffer has been fully
flushed during a COPY -- use it to resume sending after
C<put_copy_data> returned 0.

=item keep_alive

When true, the connection keeps C<EV::run> alive even with an empty
callback queue.  See L</keep_alive>.

=item loop

An L<EV> loop object.  Defaults to C<EV::default_loop>.

=back

Unknown arguments produce a C<carp> warning and are otherwise ignored.

=head1 CONNECTION METHODS

=head2 connect

    $pg->connect($conninfo);

Starts an asynchronous connection from a libpq connection string.
C<on_connect> fires on success, C<on_error> on failure.

=head2 connect_params

    $pg->connect_params(\%params);
    $pg->connect_params(\%params, $expand_dbname);

Like C<connect> but takes a hashref of keyword/value parameters.  When
C<$expand_dbname> is true, the C<dbname> entry may itself be a
connection string or URI.

=head2 reset

    $pg->reset;

Drops the current connection and reconnects with the same parameters.
Pending callbacks fire with C<(undef, "connection reset")> first.
Alias: C<reconnect>.

=head2 finish

    $pg->finish;

Closes the connection.  Pending callbacks fire with
C<(undef, "connection finished")>.  Alias: C<disconnect>.

=head2 is_connected

    my $bool = $pg->is_connected;

True if the handshake has completed and the connection is ready for
queries.  False during connect, after C<finish>, and after a fatal
error.

=head2 status

    my $st = $pg->status;

libpq connection status: C<CONNECTION_OK> or C<CONNECTION_BAD>.
Returns C<CONNECTION_BAD> when not connected.

=head1 QUERY METHODS

=head2 query

lib/EV/Pg.pm  view on Meta::CPAN

Switches the most recently sent query into single-row mode.  Must be
called immediately after a send method (C<query>, C<query_params>,
...) and before the event loop delivers any results -- a 0 return
means no query was in the right async state and should be treated as
a programmer error rather than a runtime condition.

The query callback then fires once per row with a single-row C<\@rows>
(e.g. C<[[$col0, $col1, ...]]>), and once more at the end with an
empty C<\@rows> as the completion sentinel.

=head2 set_chunked_rows_mode

    my $ok = $pg->set_chunked_rows_mode($chunk_size);

Like C<set_single_row_mode> but delivers up to C<$chunk_size> rows per
callback (requires libpq E<gt>= 17), reducing per-callback overhead for
large result sets.  Same call-timing constraint and same trailing
empty-rows completion sentinel.

=head2 close_prepared

    $pg->close_prepared($name, sub { my ($result, $err) = @_; });

Closes (deallocates) a prepared statement at protocol level (requires
libpq E<gt>= 17).  The callback receives an empty string (C<"">) on
success.  Works in pipeline mode, unlike C<DEALLOCATE> SQL.

=head2 close_portal

    $pg->close_portal($name, sub { my ($result, $err) = @_; });

Closes a portal at protocol level (requires libpq E<gt>= 17).
The callback receives an empty string (C<"">) on success.

=head2 cancel

    my $err = $pg->cancel;

Sends a cancel request using the legacy C<PQcancel> API.  B<Blocks>
the event loop for one network round trip; prefer C<cancel_async> on
libpq E<gt>= 17.  Returns C<undef> on success or an error string on
failure.

=head2 cancel_async

    $pg->cancel_async(sub { my ($r, $err) = @_; });

Sends a non-blocking cancel request using the C<PQcancelConn> API
(requires libpq E<gt>= 17).  The callback receives C<(1)> on success
or C<(undef, $errmsg)> on failure.  Croaks if a cancel is already in
progress.

=head2 pending_count

    my $n = $pg->pending_count;

Number of callbacks currently in the queue (queries sent but not yet
delivered).

=head2 keep_alive

    $pg->keep_alive(1);
    my $bool = $pg->keep_alive;

When true, the read watcher keeps C<EV::run> alive even when the
callback queue is empty.  Required when waiting for server-side
C<NOTIFY> events via C<on_notify> -- without this flag the loop would
exit as soon as the C<LISTEN> query completes.  Getter/setter.

=head2 skip_pending

    $pg->skip_pending;

Drops every queued callback, invoking each with
C<(undef, "skipped")>.  Any in-flight server results are drained and
discarded; the connection remains usable for new queries.

=head1 PIPELINE METHODS

Pipeline mode lets you send multiple queries without waiting for
individual results, then receive the results in order after a sync
point.  Inside a pipeline you must use C<query_params> or
C<query_prepared> -- C<query> is rejected.

=head2 enter_pipeline

    $pg->enter_pipeline;

Switches the connection into pipeline mode.  Croaks if there are
unfinished results outstanding.

=head2 exit_pipeline

    $pg->exit_pipeline;

Returns to normal mode.  Croaks if the pipeline is not idle.

=head2 pipeline_sync

    $pg->pipeline_sync(sub { my ($r, $err) = @_; });

Sends a pipeline sync point.  The callback fires with C<(1)> after all
preceding queries in the batch have completed, or
C<(undef, $errmsg)> if the connection drops first.  Alias: C<sync>.

=head2 send_pipeline_sync

    $pg->send_pipeline_sync(sub { my ($r, $err) = @_; });

Like C<pipeline_sync> but does B<not> flush the send buffer (requires
libpq E<gt>= 17).  Useful for batching multiple sync points before a
single manual flush via C<send_flush_request>.

=head2 send_flush_request

    $pg->send_flush_request;

Asks the server to deliver results for queries sent so far -- the
manual companion to C<send_pipeline_sync>.  Alias: C<flush>.

=head2 pipeline_status

    my $st = $pg->pipeline_status;

One of C<PQ_PIPELINE_OFF>, C<PQ_PIPELINE_ON>, or