AnyEvent-DBD-Pg

 view release on metacpan or  search on metacpan

README  view on Meta::CPAN

    AnyEvent::DBD::Pg - AnyEvent interface to DBD::Pg's async interface

SYNOPSIS
        use AnyEvent::DBD::Pg;

        my $adb = AnyEvent::DBD::Pg->new('dbi:Pg:dbname=test', user => 'pass', {
            pg_enable_utf8 => 1,
            pg_server_prepare => 0,
            quote_char => '"',
            name_sep => ".",
        }, debug => 1);

        $adb->queue_size( 4 );
        $adb->debug( 1 );

        $adb->connect;

        $adb->selectcol_arrayref("select pg_sleep( 0.1 ), 1", { Columns => [ 1 ] }, sub {
            my $rc = shift or return warn;
            my $res = shift;
            warn "Got <$adb->{qd}> = $rc / @{$res}";
            $adb->selectrow_hashref("select data,* from tx limit 2", {}, sub {
                my $rc = shift or return warn;
                warn "Got $adb->{qd} = $rc [@_]";

ex/sample.pl  view on Meta::CPAN

	use lib::abs '../lib';
	use EV;
	
	use AnyEvent::DBD::Pg;
	
	my $adb = AnyEvent::DBD::Pg->new('dbi:Pg:dbname=test', user => 'pass', {
		pg_enable_utf8 => 1,
		pg_server_prepare => 0,
		quote_char => '"',
		name_sep => ".",
	}, debug => 1);
	
	$adb->queue_size( 4 );
	$adb->debug( 1 );
	
	$adb->connect;
	
	$adb->selectcol_arrayref("select pg_sleep( 0.1 ), 1", { Columns => [ 1 ] }, sub {
		my $rc = shift or return warn;
		my $res = shift;
		warn "Got <$adb->{qd}> = $rc / @{$res}";
		$adb->selectrow_hashref("select data,* from tx limit 2", {}, sub {
			my $rc = shift or return warn;
			warn "Got $adb->{qd} = $rc [@_]";

lib/AnyEvent/DBD/Pg.pm  view on Meta::CPAN

=head1 SYNOPSIS

	use AnyEvent::DBD::Pg;
	
	my $adb = AnyEvent::DBD::Pg->new('dbi:Pg:dbname=test', user => 'pass', {
		pg_enable_utf8 => 1,
		pg_server_prepare => 0,
		quote_char => '"',
		name_sep => ".",
	}, debug => 1);
	
	$adb->queue_size( 4 );
	$adb->debug( 1 );
	
	$adb->connect;
	
	$adb->selectcol_arrayref("select pg_sleep( 0.1 ), 1", { Columns => [ 1 ] }, sub {
		my $rc = shift or return warn;
		my $res = shift;
		warn "Got <$adb->{qd}> = $rc / @{$res}";
		$adb->selectrow_hashref("select data,* from tx limit 2", {}, sub {
			my $rc = shift or return warn;
			warn "Got $adb->{qd} = $rc [@_]";

lib/AnyEvent/DBD/Pg.pm  view on Meta::CPAN

	$self->{queue} = [];
	#$self->{current};
	#$self->connect;
	$self->{querynum}  = 0;
	$self->{queuetime} = 0;
	$self->{querytime} = 0;
	$self;
}

BEGIN {
	for my $method (qw( cnn queue_size debug )) {
		*$method = sub { @_ > 1 ? $_[0]->{$method} = $_[1] : $_[0]->{$method} }
	}
}

sub connect {
	my $self = shift;
	my ($dsn,$user,$pass,$args) = @{ $self->{cnn} };
	local $args->{RaiseError} = 0;
	local $args->{PrintError} = 0;
	

lib/AnyEvent/DBD/Pg.pm  view on Meta::CPAN

	close $fn2;
	close $fn3;
	if( $self->{db} = DBI->connect($dsn,$user,$pass,$args) ) {
		open my $fn3, '>','/dev/null';
		if (fileno $fn3 == $next) {
			$self->{fh} = $candidate;
		} else {
			die sprintf "Bad descriptor definition implementation: got too many fds: [ %d -> %d -> %d <> %d -> ? -> %d ]\n",
				fileno($fn1), $candidate,$next, fileno($fn3);
		}
		warn "Connection to $dsn established\n" if $self->{debug} > 2;
		$self->{lasttry} = undef;
		$self->{gone} = undef;
		return $self->{db}->ping;
	} else {
		$self->{gone} = time unless defined $self->{gone};
		$self->{lasttry} = time;
		warn "Connection to $dsn failed: ".DBI->errstr;
		return 0;
	}
}

lib/AnyEvent/DBD/Pg.pm  view on Meta::CPAN

	defined $fetchmethod or croak "Method $method not implemented yet";
	ref (my $cb = pop) eq 'CODE' or croak "need callback";
	if ($self->{db}->{pg_async_status} == 1 or $self->{current} ) {
		if ( @{ $self->{queue} } >= $self->{queue_size} - 1 ) {
			my $c = 1;
			my $counter = ++$self->{querynum};
			local $@ = "Query $_[0] run out of queue size $self->{queue_size}";
			printf STDERR "\e[036;1mQ$counter\e[0m. [\e[03${c};1m%0.4fs\e[0m] < \e[03${c};1m%s\e[0m > ".("\e[031;1mQuery run out of queue size\e[0m")."\n", 0 , $_[0];
			return $cb->();
		} else {
			warn "Query $_[0] pushed to queue\n" if $self->{debug} > 1;
			push @{ $self->{queue} }, [time(), $method, @_,$cb];
			return;
		}
	}
	my $query = shift;
	my $args = shift || {};
	$args->{pg_async} = PG_ASYNC;
	my $counter = ++$self->{querynum};
	warn "prepare call <$query>( @_ ), async status = ".$self->{db}->{pg_async_status} if $self->{debug} > 2;
	$self->{current} = [$query,@_];
	$self->{current_start} = time();
	
	weaken $self;
	$self or return;
	my ($st,$w,$t,$check);
	my @watchers;
	push @watchers, sub {
		$self and $st or warn("no self"), @watchers = (), return 1;
		#warn "check status=$self->{db}->{pg_async_status}\n";
		if($self->{db}->{pg_async_status} and $st->pg_ready()) {
			undef $w;
			local $@;
			my $res = $st->pg_result;
			my $run = time - $self->{current_start};
			$self->{querytime} += $run;
			my ($diag,$DIE);
			if ($self->{debug}) {
				$diag = $self->{current}[0];
				my @bind = @{ $self->{current} };
				shift @bind;
				$diag =~ s{\?}{ "'".shift(@bind)."'" }sge;
			} else {
				$diag = $self->{current}[0];
			}
			if (!$res) {
				$DIE = $self->{db}->errstr;
			}
			local $self->{qd} = $diag;
			if ($self->{debug}) {
				my $c = $run < 0.01 ? '2' : $run < 0.1 ? '3' : '1';
				my $x = $DIE ? '1' : '6';
				printf STDERR "\e[036;1mQ$counter\e[0m. [\e[03${c};1m%0.4fs\e[0m] < \e[03${x};1m%s\e[0m > ".($DIE ? "\e[031;1m$DIE\e[0m" : '')."\n", $run , $diag;
			}
			local $self->{queuing} = @{ $self->{queue} };
			if ($res) {
				if (ref $fetchmethod) {
					$cb->($res, $st->$fetchmethod($args));
				} else {
					$cb->($res, $st->$fetchmethod);