File Coverage

blib/lib/DBIx/Handler.pm

Criterion	Covered	Total	%
statement	15	157	9.5
branch	0	78	0.0
condition	0	32	0.0
subroutine	5	31	16.1
pod	17	18	94.4
total	37	316	11.7

package DBIx::Handler;
use strict;
use warnings;
our $VERSION = '0.13';

use DBI 1.605;
use DBIx::TransactionManager 1.09;
use Carp ();

sub _noop {}

*connect = \&new;
sub new {
    my $class = shift;

    my $opts = scalar(@_) == 5 ? pop @_ : +{};
    bless {
        _connect_info    => [@_],
        _pid             => undef,
        _dbh             => undef,
        trace_query      => $opts->{trace_query}      || 0,
        trace_ignore_if  => $opts->{trace_ignore_if}  || \&_noop,
        result_class     => $opts->{result_class}     || undef,
        on_connect_do    => $opts->{on_connect_do}    || undef,
        on_disconnect_do => $opts->{on_disconnect_do} || undef,
        no_ping          => $opts->{no_ping}          || 0,
        dbi_class        => $opts->{dbi_class}        || "DBI",
    }, $class;
}

sub _connect {
    my $self = shift;

    my $dbh = $self->{_dbh} = $self->{dbi_class}->connect(@{$self->{_connect_info}});
    my $attr = @{$self->{_connect_info}} > 3 ? $self->{_connect_info}->[3] : {};

    if (DBI->VERSION > 1.613 && !exists $attr->{AutoInactiveDestroy}) {
        $dbh->STORE(AutoInactiveDestroy => 1);
    }

    if (!exists $attr->{RaiseError} && !exists $attr->{HandleError}) {
        $dbh->STORE(RaiseError => 1);
    }

    if ($dbh->FETCH('RaiseError') && !exists $attr->{PrintError}) {
        $dbh->STORE(PrintError => 0);
    }

    $self->{_pid} = $$;

    $self->_run_on('on_connect_do', $dbh);

    $dbh;
}

sub dbh {
    my $self = shift;
    $self->_seems_connected or $self->_connect;
}

sub _ping {
    my ($self, $dbh) = @_;
    $self->{no_ping} || $dbh->ping;
}

sub _seems_connected {
    my $self = shift;

    my $dbh = $self->{_dbh} or return;

    if ( $self->{_pid} != $$ ) {
        $dbh->STORE(InactiveDestroy => 1);
        $self->_in_txn_check;
        delete $self->{txn_manager};
        return;
    }

    unless ($dbh->FETCH('Active') && $self->_ping($dbh)) {
        $self->_in_txn_check;
        $self->_disconnect;
        return;
    }

    $dbh;
}

sub disconnect {
    my $self = shift;

    $self->_seems_connected or return;
    $self->_disconnect;
}

sub _disconnect {
    my $self = shift;
    my $dbh = delete $self->{_dbh} or return;
    delete $self->{txn_manager};
    $self->_run_on('on_disconnect_do', $dbh);
    $dbh->STORE(CachedKids => {});
    $dbh->disconnect;
}

sub _run_on {
    my ($self, $mode, $dbh) = @_;
    if ( my $on_connect_do = $self->{$mode} ) {
        if (not ref($on_connect_do)) {
            $dbh->do($on_connect_do);
        } elsif (ref($on_connect_do) eq 'CODE') {
            $on_connect_do->($dbh);
        } elsif (ref($on_connect_do) eq 'ARRAY') {
            $dbh->do($_) for @$on_connect_do;
        } else {
            Carp::croak("Invalid $mode: ".ref($on_connect_do));
        }
    }
}

sub DESTROY { $_[0]->disconnect }

sub result_class {
    my ($self, $result_class) = @_;
    $self->{result_class} = $result_class if $result_class;
    $self->{result_class};
}

sub trace_query {
    my ($self, $flag) = @_;
    $self->{trace_query} = $flag if defined $flag;
    $self->{trace_query};
}

sub trace_ignore_if {
    my ($self, $callback) = @_;
    $self->{trace_ignore_if} = $callback if defined $callback;
    $self->{trace_ignore_if};
}

sub no_ping {
    my ($self, $enable) = @_;
    $self->{no_ping} = $enable if defined $enable;
    $self->{no_ping};
}

sub query {
    my ($self, $sql, @args) = @_;

    my $bind;
    if (ref($args[0]) eq 'HASH') {
        ($sql, $bind) = $self->replace_named_placeholder($sql, $args[0]);
    }
    else {
        $bind = ref($args[0]) eq 'ARRAY' ? $args[0] : \@args;
    }

    $sql = $self->trace_query_set_comment($sql);

    my $sth;
    eval {
        $sth = $self->dbh->prepare($sql);
        $sth->execute(@{$bind || []});
    };
    if (my $error = $@) {
        Carp::croak($error);
    }

    my $result_class = $self->result_class;
    $result_class ? $result_class->new($self, $sth) : $sth;
}

sub replace_named_placeholder {
    my ($self, $sql, $args) = @_;

    my %named_bind = %{$args};
    my @bind;
    $sql =~ s{:(\w+)}{
        Carp::croak("$1 does not exists in hash") if !exists $named_bind{$1};
        if ( ref $named_bind{$1} && ref $named_bind{$1} eq "ARRAY" ) {
            push @bind, @{ $named_bind{$1} };
            my $tmp = join ',', map { '?' } @{ $named_bind{$1} };
            "($tmp)";
        } else {
            push @bind, $named_bind{$1};
            '?'
        }
    }ge;

    return ($sql, \@bind);
}

sub trace_query_set_comment {
    my ($self, $sql) = @_;
    return $sql unless $self->trace_query;

    my $i = 1;
    while ( my (@caller) = caller($i++) ) {
        next if ( $caller[0]->isa( __PACKAGE__ ) );
        next if $self->trace_ignore_if->(@caller);
        my $comment = "$caller[1] at line $caller[2]";
        $comment =~ s/\*\// /g;
        $sql = "/* $comment */ $sql";
        last;
    }

    $sql;
}

sub run {
    my ($self, $coderef) = @_;
    my $wantarray = wantarray;

    my @ret = eval {
        my $dbh = $self->dbh;
        $wantarray ? $coderef->($dbh) : scalar $coderef->($dbh);
    };
    if (my $error = $@) {
        Carp::croak($error);
    }

    $wantarray ? @ret : $ret[0];
}

# --------------------------------------------------------------------------------
# for transaction
sub txn_manager {
    my $self = shift;

    my $dbh = $self->dbh;
    $self->{txn_manager} ||= DBIx::TransactionManager->new($dbh);
}

sub in_txn {
    my $self = shift;
    return unless $self->{txn_manager};
    return $self->{txn_manager}->in_transaction;
}

sub _in_txn_check {
    my $self = shift;

    my $info = $self->in_txn;
    return unless $info;

    my $caller = $info->{caller};
    my $pid    = $info->{pid};
    Carp::confess("Detected transaction during a connect operation (last known transaction at $caller->[1] line $caller->[2], pid $pid). Refusing to proceed at");
}

sub txn_scope {
    my @caller = caller();
    $_[0]->txn_manager->txn_scope(caller => \@caller);
}

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

    my $wantarray = wantarray;
    my $txn = $self->txn_scope;

    my @ret = eval {
        my $dbh = $self->dbh;
        $wantarray ? $coderef->($dbh) : scalar $coderef->($dbh);
    };

    if (my $error = $@) {
        $txn->rollback;
        Carp::croak($error);
    } else {
        eval { $txn->commit };
        Carp::croak($@) if $@;
    }

    $wantarray ? @ret : $ret[0];
}

sub txn_begin    { $_[0]->txn_manager->txn_begin    }
sub txn_rollback { $_[0]->txn_manager->txn_rollback }
sub txn_commit   { $_[0]->txn_manager->txn_commit   }

1;

__END__