File Coverage

blib/lib/Database/Async.pm

Criterion	Covered	Total	%
statement	66	125	52.8
branch	8	24	33.3
condition	2	15	13.3
subroutine	21	37	56.7
pod	15	20	75.0
total	112	221	50.6

package Database::Async;
# ABSTRACT: database interface for use with IO::Async
use strict;
use warnings;

our $VERSION = '0.016';

use parent qw(Database::Async::DB IO::Async::Notifier);

=head1 NAME

Database::Async - provides a database abstraction layer for L

=head1 SYNOPSIS

 # Just looking up one thing?
 my ($id) = $db->query(
  q{select id from some_table where name = ?},
  bind => ['some name']
 )->single
  # This is an example, so we want the result immediately - in
  # real async code, you'd rarely call Future->get, but would
  # typically use `->then` or `->on_done` instead
  ->get;
 # or, with Future::AsyncAwait, try:
 my ($id) = await $db->query(
  q{select id from some_table where name = ?},
  bind => ['some name']
 )->single;

 # Simple query
 $db->query(q{select id, some_data from some_table})
    ->row_hashrefs
    ->each(sub {
        printf "ID %d, data %s\n", $_->{id}, $_->{some_data};
    })
    # If you want to complete the full query, don't forget to call
    # ->get or ->retain here!
    ->retain;

 # Transactions
 $db->transaction(sub {
  my ($tx) = @_;
 })->commit
  # This returns a Future, so if you want to wait for it to complete,
  # call `->get` (throws an exception if something goes wrong)
  # or `->await` (just waits for it to succeed or fail, but ignores
  # the result).
 ->get;

=head1 DESCRIPTION

Database support for L. This is the base API, see L
and subclasses for specific database functionality.

B.

L provides a basic API for interacting with a database, but this is
very low level and uses a synchronous design. See L if you're
familiar with L and want an interface that follows it more closely.

Typically a database only allows a single query to run at a time.
Other queries will be queued.

Set up a pool of connections to provide better parallelism:

    my $dbh = Database::Async->new(
        uri  => 'postgresql://write@maindb/dbname?sslmode=require',
        pool => {
            max => 4,
        },
    );

Queries and transactions will then automatically be distributed
among these connections. However, note that:

=over 4

=item * all queries within a transaction will be made on the same connection

=item * ordering guarantees are weaker: queries will be started in
order on the next available connection

=back

With a single connection, you could expect:

    Future->needs_all(
     $dbh->do(q{insert into x ...}),
     $dbh->do(q{select from x ...})
    );

to insert the rows first, then return them in the C call. B.

=head2 Pool configuration

The following parameters are currently accepted for defining the pool:

=over 4

=item * C - minimum number of total connections to maintain, defaults to 0

=item * C - maximum permitted active connections, default is 1

=item * C - how to iterate through the available URIs, options include
C and C (default, round-robin behaviour).

=item * C - algorithm for managing connection timeouts or failures. The default
is an exponential backoff with 10ms initial delay, 30s maximum, resetting on successful
connection.

=back

See L for more details.

=head2 DBI

The interface is not the same as L, but here are some approximate equivalents for
common patterns:

=head3 selectall_hashref

In L:

 print $_->{id} . "\n" for
  $dbh->selectall_hashref(
   q{select * from something where id = ?},
   undef,
   $id
  )->@*;

In L:

 print $_->{id} . "\n" for
  $db->query(
   q{select * from something where id = ?},
   bind => [
    $id
   ])->row_hashrefs
     ->as_arrayref
     ->@*

In L:

 my $sth = $dbh->prepare(q{select * from something where id = ?});
 for my $id (1, 2, 3) {
  $sth->bind(0, $id, 'bigint');
  $sth->execute;
  while(my $row = $sth->fetchrow_hashref) {
   print $row->{name} . "\n";
  }
 }

In L:

 my $sth = $db->prepare(q{select * from something where id = ?});
 (Future::Utils::fmap_void  {
  my ($id) = @_;
  $sth->bind(0, $id, 'bigint')
   ->then(sub { $sth->execute })
   ->then(sub {
    $sth->row_hashrefs
     ->each(sub {
      print $_->{name} . "\n";
     })->completed
   })
 } foreach => [1, 2, 3 ])->get;

=cut

use mro;
no indirect;

use Future::AsyncAwait;
use Syntax::Keyword::Try;

use URI;
use URI::db;
use Module::Load ();
use Scalar::Util qw(blessed);

use Database::Async::Engine;
use Database::Async::Pool;
use Database::Async::Query;
use Database::Async::StatementHandle;
use Database::Async::Transaction;

use Log::Any qw($log);

=head1 METHODS

=cut

=head2 transaction

Resolves to a L which will yield a L
instance once ready.

=cut

async sub transaction {
    my ($self, @args) = @_;
    Scalar::Util::weaken(
        $self->{transactions}[@{$self->{transactions}}] =
            my $txn = Database::Async::Transaction->new(
                database => $self,
                @args
            )
    );
    await $txn->begin;
    return $txn;
}

=head2 txn

Executes code within a transaction. This is meant as a shorter form of
the common idiom

 $db->transaction
    ->then(sub {
     my ($txn) = @_;
     Future->call($code)
      ->then(sub {
       $txn->commit
      })->on_fail(sub {
       $txn->rollback
      });
    })

The code must return a L, and the transaction will only be committed
if that L resolves cleanly.

Returns a L which resolves once the transaction is committed.

=cut

async sub txn {
    my ($self, $code, @args) = @_;
    my $txn = await $self->transaction;
    try {
        my @data = await Future->call(
            $code => ($txn, @args)
        );
        await $txn->commit;
        return @data;
    } catch {
        my $exception = $@;
        try {
            await $txn->rollback;
        } catch {
            $log->warnf("exception %s in rollback", $@);
        }
        die $exception;
    }
}

=head1 METHODS - Internal

You're welcome to call these, but they're mostly intended
for internal usage, and the API B change in future versions.

=cut

=head2 uri

Returns the configured L for populating database instances.

=cut

sub uri { shift->{uri} }

=head2 pool

Returns the L instance.

=cut

sub pool {
    my ($self) = @_;
    $self->{pool} //= Database::Async::Pool->new(
        $self->pool_args
    )
}

=head2 pool_args

Returns a list of standard pool constructor arguments.

=cut

sub pool_args {
    my ($self) = @_;
    return (
        request_engine => $self->curry::weak::request_engine,
        uri            => $self->uri,
    );
}

=head2 configure

Applies configuration, see L for details.

Supports the following named parameters:

=over 4

=item * C - the endpoint to use when connecting a new engine instance

=item * C - the parameters to pass when instantiating a new L

=item * C - parameters for setting up the pool, or a L instance

=item * C - default encoding to apply to parameters, queries and results, defaults to C

=back

=cut

my %encoding_map = (
    'utf8'    => 'UTF-8',
    'utf-8'   => 'UTF-8',
    'UTF8'    => 'UTF-8',
    'unicode' => 'UTF-8',
);

sub configure {
    my ($self, %args) = @_;

    if(my $encoding = delete $args{encoding}) {
        $self->{encoding} = $encoding_map{$encoding} // $encoding;
    }

    if(my $uri = delete $args{uri}) {
        # This could be any type of object. We make
        # the assumption here that it safely serialises
        # to a standard URI. Some of the database
        # engines provide such a standard (e.g. PostgreSQL).
        # Others may not...
        $self->{uri} = URI->new("$uri");
    }
    if(exists $args{engine}) {
        $self->{engine_parameters} = delete $args{engine};
    }
    if(exists $args{type}) {
        $self->{type} = delete $args{type};
    }
    if(my $pool = delete $args{pool}) {
        if(blessed $pool) {
            $self->{pool} = $pool;
        } else {
            $self->{pool} = Database::Async::Pool->new(
                $self->pool_args,
                %$pool,
            );
        }
    }
    $self->next::method(%args);
}

sub encoding { shift->{encoding} }

=head2 ryu

A L instance, used for requesting sources, sinks and timers.

=cut

sub ryu {
    my ($self) = @_;
    $self->{ryu} //= do {
        $self->add_child(
            my $ryu = Ryu::Async->new
        );
        $ryu
    }
}

=head2 new_source

Instantiates a new L.

=cut

sub new_source { shift->ryu->source }

=head2 new_sink

Instantiates a new L.

=cut

sub new_sink { shift->ryu->sink }

=head2 new_future

Instantiates a new L.

=cut

sub new_future { shift->loop->new_future }

=head1 METHODS - Internal, engine-related

=cut

=head2 request_engine

Attempts to instantiate and connect to a new L
subclass. Returns a L which should resolve to a new
L instance when ready to use.

=cut

async sub request_engine {
    my ($self) = @_;
    $log->tracef('Requesting new engine');
    my $engine = $self->engine_instance;
    $log->tracef('Connecting');
    return await $engine->connect;
}

=head2 engine_instance

Loads the appropriate engine class and attaches to the loop.

=cut

sub engine_instance {
    my ($self) = @_;
    my $uri = $self->uri;
    my $type = $self->{type} // $uri->scheme;
    die 'unknown database type ' . $type
        unless my $engine_class = $Database::Async::Engine::ENGINE_MAP{$type};
    Module::Load::load($engine_class) unless $engine_class->can('new');
    $log->tracef('Instantiating new %s', $engine_class);
    my %param = (
        %{$self->{engine_parameters} || {}},
        (defined($uri) ? (uri => $uri) : ()),
        db => $self,
    );

    # Only recent engine versions support this parameter
    if(my $encoding = $self->encoding) {
        if($engine_class->can('encoding')) {
            $param{encoding} = $self->encoding;
        } else {
            # If we're given this parameter, let's not ignore it silently
            die 'Database engine ' . $engine_class . ' does not support encoding parameter, try upgrading that module from CPAN or remove the encoding configuration in Database::Async';
        }
    }

    $self->add_child(
        my $engine = $engine_class->new(%param)
    );
    $engine;
}

=head2 engine_ready

Called by L instances when the engine is
ready for queries.

=cut

sub engine_ready {
    my ($self, $engine) = @_;
    $self->pool->queue_ready_engine($engine);
}

sub engine_disconnected {
    my ($self, $engine) = @_;
    $self->pool->unregister_engine($engine);
}

sub db { shift }

=head2 queue_query

Assign the given query to the next available engine instance.

=cut

async sub queue_query {
    my ($self, $query) = @_;
    $log->tracef('Queuing query %s', $query);
    my $engine = await $self->pool->next_engine;
    $log->tracef('Query %s about to run on %s', $query, $engine);
    return await $engine->handle_query($query);
}

sub diagnostics {
    my ($self) = @_;
}

sub notification {
    my ($self, $engine, $channel, $data) = @_;
    $log->tracef('Database notifies us via %s of %s', $channel, $data);
    $self->notification_source($channel)->emit($data);
}

sub notification_source {
    my ($self, $name) = @_;
    $self->{notification_source}{$name} //= $self->new_source;
}

1;

__END__