view release on metacpan or  search on metacpan

lib/Catmandu/Store/DBI.pm  view on Meta::CPAN

use Catmandu::Util qw(require_package);
use DBI;
use Catmandu::Store::DBI::Bag;
use Moo;
use namespace::clean;

our $VERSION = "0.0702";

with 'Catmandu::Store';
with 'Catmandu::Transactional';

has data_source => (
    is       => 'ro',
    required => 1,
    trigger  => sub {
        my $ds = $_[0]->{data_source};
        $ds = $ds =~ /^DBI:/i ? $ds : "DBI:$ds";
        $_[0]->{data_source} = $ds;
    },
);
has username => (is => 'ro', default => sub {''});
has password => (is => 'ro', default => sub {''});
has timeout => (is => 'ro', predicate => 1);
has reconnect_after_timeout => (is => 'ro');
has handler                 => (is => 'lazy');
has _in_transaction         => (is => 'rw', writer => '_set_in_transaction',);
has _connect_time           => (is => 'rw', writer => '_set_connect_time');
has _dbh => (is => 'lazy', builder => '_build_dbh', writer => '_set_dbh',);

sub handler_namespace {
    'Catmandu::Store::DBI::Handler';
}

sub _build_handler {
    my ($self) = @_;
    my $driver = $self->dbh->{Driver}{Name} // '';
    my $ns = $self->handler_namespace;
    my $pkg;
    if ($driver =~ /pg/i) {
        $pkg = 'Pg';
    }
    elsif ($driver =~ /sqlite/i) {
        $pkg = 'SQLite';
    }
    elsif ($driver =~ /mysql/i) {
        $pkg = 'MySQL';
    }
    else {
        Catmandu::NotImplemented->throw(
            'Only Pg, SQLite and MySQL are supported.');
    }
    require_package($pkg, $ns)->new;
}

sub _build_dbh {
    my ($self) = @_;
    my $opts = {
        AutoCommit                       => 1,
        RaiseError                       => 1,
        mysql_auto_reconnect             => 1,
        mysql_enable_utf8                => 1,
        pg_utf8_strings                  => 1,
        sqlite_use_immediate_transaction => 1,
        sqlite_unicode                   => 1,
    };
    my $dbh
        = DBI->connect($self->data_source, $self->username, $self->password,
        $opts,);
    $self->_set_connect_time(time);
    $dbh;
}

sub dbh {
    my ($self)       = @_;
    my $dbh          = $self->_dbh;
    my $connect_time = $self->_connect_time;
    my $driver = $dbh->{Driver}{Name} // '';

    # MySQL has builtin option mysql_auto_reconnect
    if (   $driver !~ /mysql/i
        && $self->has_timeout
        && time - $connect_time > $self->timeout)
    {
        if ($self->reconnect_after_timeout || !$dbh->ping) {

            # ping failed, so try to reconnect
            $dbh->disconnect;
            $dbh = $self->_build_dbh;
            $self->_set_dbh($dbh);
        }
        else {
            $self->_set_connect_time(time);
        }
    }

    $dbh;
}

sub transaction {
    my ($self, $sub) = @_;

    if ($self->_in_transaction) {
        return $sub->();
    }

    my $dbh = $self->dbh;
    my @res;

    eval {
        $self->_set_in_transaction(1);
        $dbh->begin_work;
        @res = $sub->();
        $dbh->commit;
        $self->_set_in_transaction(0);
        1;
    } or do {
        my $err = $@;
        eval {$dbh->rollback};
        $self->_set_in_transaction(0);
        die $err;
    };