[ViewVC] Diff of: cvs/AnyEvent-DBI/DBI.pm

Comparing AnyEvent-DBI/DBI.pm (file contents):
Revision 1.3 by root, Fri Jun 6 16:13:07 2008 UTC vs.
Revision 1.10 by root, Tue Jun 2 16:16:03 2009 UTC

…		…
3	AnyEvent::DBI - asynchronous DBI access	3	AnyEvent::DBI - asynchronous DBI access
4		4
5	=head1 SYNOPSIS	5	=head1 SYNOPSIS
6		6
7	use AnyEvent::DBI;	7	use AnyEvent::DBI;
		8
		9	my $cv = AnyEvent->condvar;
		10
		11	my $dbh = new AnyEvent::DBI "DBI:SQLite:dbname=test.db", "", "";
		12
		13	$dbh->exec ("select * from test where num=?", 10, sub {
		14	my ($rows, $rv) = @_;
		15
		16	print "@$_\n"
		17	for @$rows;
		18
		19	$cv->broadcast;
		20	});
		21
		22	# asynchronously do sth. else here
		23
		24	$cv->wait;
8		25
9	=head1 DESCRIPTION	26	=head1 DESCRIPTION
10		27
11	This module is an L<AnyEvent> user, you need to make sure that you use and	28	This module is an L<AnyEvent> user, you need to make sure that you use and
12	run a supported event loop.	29	run a supported event loop.
13		30
14	This module implements asynchronous DBI access my forking or executing	31	This module implements asynchronous DBI access by forking or executing
15	separate "DBI-Server" processes and sending them requests.	32	separate "DBI-Server" processes and sending them requests.
16		33
17	It means that you can run DBI requests in parallel to other tasks.	34	It means that you can run DBI requests in parallel to other tasks.
18		35
19	The overhead for very simple statements ("select 0") is somewhere	36	The overhead for very simple statements ("select 0") is somewhere
20	around 120% to 200% (single/dual core CPU) compared to an explicit	37	around 120% to 200% (dual/single core CPU) compared to an explicit
21	prepare_cached/execute/fetchrow_arrayref/finish combination.	38	prepare_cached/execute/fetchrow_arrayref/finish combination.
22		39
23	=cut	40	=cut
24		41
25	package AnyEvent::DBI;	42	package AnyEvent::DBI;
…		…
35	use DBI ();	52	use DBI ();
36		53
37	use AnyEvent ();	54	use AnyEvent ();
38	use AnyEvent::Util ();	55	use AnyEvent::Util ();
39		56
		57	use Errno qw(:POSIX);
		58	use Fcntl qw(F_SETFD);
		59	use POSIX qw(sysconf _SC_OPEN_MAX);
		60
40	our $VERSION = '1.0';	61	our $VERSION = '1.2';
		62	my $fd_max = 1023; # default
		63	eval { $fd_max = sysconf _SC_OPEN_MAX - 1; };
41		64
42	# this is the forked server code	65	# this is the forked server code
43		66
44	our $DBH;	67	our $DBH;
45		68
46	sub req_open {	69	sub req_open {
47	my (undef, $dbi, $user, $pass, %attr) = @{+shift};	70	my (undef, $dbi, $user, $pass, %attr) = @{+shift};
48		71
49	$DBH = DBI->connect ($dbi, $user, $pass, \%attr);	72	$DBH = DBI->connect ($dbi, $user, $pass, \%attr) or die $DBI::errstr;
50		73
51	[1]	74	[1]
52	}	75	}
53		76
54	sub req_exec {	77	sub req_exec {
55	my (undef, $st, @args) = @{+shift};	78	my (undef, $st, @args) = @{+shift};
56
57	my $sth = $DBH->prepare_cached ($st, undef, 1);	79	my $sth = $DBH->prepare_cached ($st, undef, 1)
		80	or die [$DBI::errstr];
58		81
59	$sth->execute (@args)	82	my $rv = $sth->execute (@args)
60	or die $sth->errstr;	83	or die [$sth->errstr];
61		84
62	[$sth->fetchall_arrayref]	85	[1, $sth->{NUM_OF_FIELDS} ? $sth->fetchall_arrayref : undef, { rv => $rv }]
		86	}
		87
		88	sub req_attr {
		89	my (undef, $attr_name, $attr_val) = @{+shift};
		90
		91	if (defined $attr_val) {
		92	$DBH->{$attr_name} = $attr_val;
		93	}
		94
		95	[1, $DBH->{$attr_name}]
		96	}
		97
		98	sub req_begin_work {
		99	[scalar $DBH->begin_work or die $DBI::errstr]
		100	}
		101
		102	sub req_commit {
		103	[scalar $DBH->commit or die $DBI::errstr]
		104	}
		105
		106	sub req_rollback {
		107	[scalar $DBH->rollback or die $DBI::errstr]
		108	}
		109
		110	sub req_func {
		111	my (undef, $arg_string, $function) = @{+shift};
		112	my @args = eval $arg_string;
		113
		114	if ($@) {
		115	die "Bad func() arg string: $@";
		116	}
		117
		118	my $rv = $DBH->func (@args, $function);
		119	return [$rv . $DBH->err];
63	}	120	}
64		121
65	sub serve {	122	sub serve {
66	my ($fh) = @_;	123	my ($fileno) = @_;
		124
		125	open my $fh, ">>&=$fileno"
		126	or die "Couldn't open service socket: $!";
67		127
68	no strict;	128	no strict;
69		129
70	eval {	130	eval {
71	my $rbuf;	131	my $rbuf;
…		…
82		142
83	my $req = Storable::thaw substr $rbuf, 4;	143	my $req = Storable::thaw substr $rbuf, 4;
84	substr $rbuf, 0, $len + 4, ""; # remove length + request	144	substr $rbuf, 0, $len + 4, ""; # remove length + request
85		145
86	my $wbuf = eval { pack "L/a*", Storable::freeze $req->[0]($req) };	146	my $wbuf = eval { pack "L/a*", Storable::freeze $req->[0]($req) };
87
88	$wbuf = pack "L/a*", Storable::freeze [undef, "$@"]	147	$wbuf = pack "L/a*", Storable::freeze [undef, ref $@ ? "$@->[0]" : $@ , ref $@ ? $@->[1] : 1]
89	if $@;	148	if $@;
90		149
91	for (my $ofs = 0; $ofs < length $wbuf; ) {	150	for (my $ofs = 0; $ofs < length $wbuf; ) {
92	$ofs += (syswrite $fh, substr $wbuf, $ofs	151	$ofs += (syswrite $fh, substr $wbuf, $ofs
93	or die "unable to write results");	152	or die "unable to write results");
94	}	153	}
95	}	154	}
96	}	155	}
97	};	156	};
98		157
		158	if (AnyEvent::WIN32) {
99	kill 9, $$; # no other way on the broken windows platform	159	kill 9, $$; # no other way on the broken windows platform
		160	# and the above doesn't even work on windows, it seems the only
		161	# way to is to leak memory and kill 9 from the parent. yay.
		162	}
		163
		164	require POSIX;
		165	POSIX::_exit 0;
		166	# and the above kills the parent process on windows
		167	}
		168
		169	sub start_server {
		170	serve shift @ARGV;
100	}	171	}
101		172
102	=head2 METHODS	173	=head2 METHODS
103		174
104	=over 4	175	=over 4
…		…
126		197
127	When an error occurs, then this callback will be invoked. On entry, C<$@>	198	When an error occurs, then this callback will be invoked. On entry, C<$@>
128	is set to the error message. C<$filename> and C<$line> is where the	199	is set to the error message. C<$filename> and C<$line> is where the
129	original request was submitted.	200	original request was submitted.
130		201
131	If this callback returns and this was a fatal error (C<$fatal> is true)	202	If the fatal argument is true then the database connection shuts down and your
132	then AnyEvent::DBI die's, otherwise it calls the original request callback	203	database handle becomes invalid. All of your queued request callbacks are
133	without any arguments.	204	called without any arguments.
134		205
135	If omitted, then C<die> will be called on any errors, fatal or not.	206	If omitted, then C<die> will be called on any errors, fatal or not.
136		207
		208	The C<$dbh> argument is always a weak reference to the AnyEvent::DBI object.
		209
		210	=item on_connect => $callback->($dbh)
		211
		212	If you supply an on_connect callback, then this callback will be invoked after
		213	the database connection is attempted. If the connection succeeds, C<$dbh>
		214	contains a weak reference to the AnyEvent::DBI object. If the connection fails
		215	for any reason, no arguments are passed to the callback and C<$@> contains
		216	$DBI::errstr.
		217
		218	Regardless of whether on_connect is supplied, connect errors will result in
		219	on_error being called. However, if no on_connect callback is supplied, then
		220	connection errors are considered fatal. The client will die() and the on_error
		221	callback will be called with C<$fatal> true. When on_connect is supplied,
		222	connect error are not fatal and AnyEvent::DBI will not die(). You still
		223	cannot, however, use the $dbh object you recived from new() to make requests.
		224
		225	=item exec_server => 1
		226
		227	If you supply an exec_server argument, then the DBI server process will call
		228	something like:
		229
		230	exec "$^X -MAnyEvent::DBI -e AnyEvent::DBI::start_server"
		231
		232	after forking. This will provide the cleanest possible interpreter for your
		233	database server. There are special provisions to include C<-Mblib> if the
		234	current interpreter is running with blib.
		235
		236	If you do not supply the exec_server argument (or supply it with a false value)
		237	then the traditional method of starting the server within the same forked
		238	interpreter context is used. The forked interpreter will try to clean itself
		239	up by calling POSIX::close on all filedescriptors except STDIN, STDOUT, and
		240	STDERR (and the socket it uses to communicate with the cilent, of course).
		241
		242	=item timeout => seconds
		243
		244	If you supply a timeout parameter (floating point number of seconds), then a
		245	timer is started any time the DBI handle expects a response from the server.
		246	This includes connection setup as well as requests made to the backend. The
		247	timeout spans the duration from the moment the first data is written (or queued
		248	to be written) until all expected responses are returned, but is postponed for
		249	"timeout" seconds each time more data is returned from the server. If the
		250	timer ever goes off then a fatal error is generated. If you have an on_error
		251	handler installed, then it will be called, otherwise your program will die().
		252
		253	When altering your databases with timeouts it is wise to use transactions. If
		254	you quit due to timeout while performing insert, update or schema-altering
		255	commands you can end up not knowing if the action was submitted to the
		256	database, complicating recovery.
		257
		258	Timeout errors are always fatal.
		259
137	=back	260	=back
		261
		262	Any additional key-value pairs will be rolled into a hash reference and passed
		263	as the final argument to the DBI->connect(...) call. For example, to supress
		264	errors on STDERR and send them instead to an AnyEvent::Handle you could do:
		265
		266	$dbh = new AnyEvent::DBI
		267	"DBI:mysql:test;mysql_read_default_file=/root/.my.cnf", "", "",
		268	PrintError => 0,
		269	on_error => sub { $log_handle->push_write("DBI Error: $@ at $_[1]:$_[2]\n"); }
138		270
139	=cut	271	=cut
140		272
141	# stupid Storable autoloading, total loss-loss situation	273	# stupid Storable autoloading, total loss-loss situation
142	Storable::thaw Storable::freeze [];	274	Storable::thaw Storable::freeze [];
…		…
145	my ($class, $dbi, $user, $pass, %arg) = @_;	277	my ($class, $dbi, $user, $pass, %arg) = @_;
146		278
147	socketpair my $client, my $server, &Socket::AF_UNIX, &Socket::SOCK_STREAM, &Socket::PF_UNSPEC	279	socketpair my $client, my $server, &Socket::AF_UNIX, &Socket::SOCK_STREAM, &Socket::PF_UNSPEC
148	or croak "unable to create dbi communicaiton pipe: $!";	280	or croak "unable to create dbi communicaiton pipe: $!";
149		281
		282	my %dbi_args = ( %arg ) ;
		283	delete @dbi_args{qw( on_connect on_error timeout exec_server )};
		284
150	my $self = bless \%arg, $class;	285	my $self = bless \%arg, $class;
151
152	$self->{fh} = $client;	286	$self->{fh} = $client;
153		287
154	Scalar::Util::weaken (my $wself = $self);	288	Scalar::Util::weaken (my $wself = $self);
155		289
156	AnyEvent::Util::fh_nonblocking $client, 1;	290	AnyEvent::Util::fh_nonblocking $client, 1;
157		291
158	my $rbuf;	292	my $rbuf;
159	my @caller = (caller)[1,2]; # the "default" caller	293	my @caller = (caller)[1,2]; # the "default" caller
160		294
161	$self->{rw} = AnyEvent->io (fh => $client, poll => "r", cb => sub {	295	$self->{rw} = AnyEvent->io (fh => $client, poll => "r", cb => sub {
		296	return unless $wself;
162	my $len = sysread $client, $rbuf, 65536, length $rbuf;	297	my $len = sysread $client, $rbuf, 65536, length $rbuf;
		298	my $err = $!;
163		299
164	if ($len > 0) {	300	if ($len > 0) {
		301	# we received data, so reset the timer
		302	delete $wself->{timer};
		303	if ($wself->{timeout}) {
		304	$wself->{timer} = AnyEvent->timer (
		305	after => $wself->{timeout},
		306	cb => sub { $wself && $wself->_timedout },
		307	);
		308	}
165		309
166	while () {	310	while () {
167	my $len = unpack "L", $rbuf;	311	my $len = unpack "L", $rbuf;
168		312
169	# full request available?	313	# full response available?
170	last unless $len && $len + 4 <= length $rbuf;	314	last unless $len && $len + 4 <= length $rbuf;
171		315
172	my $res = Storable::thaw substr $rbuf, 4;	316	my $res = Storable::thaw substr $rbuf, 4;
173	substr $rbuf, 0, $len + 4, ""; # remove length + request	317	substr $rbuf, 0, $len + 4, ""; # remove length + request
174		318
		319	last unless $wself;
175	my $req = shift @{ $wself->{queue} };	320	my $req = shift @{ $wself->{queue} };
176		321
177	if (defined $res->[0]) {	322	if (defined $res->[0]) {
		323	$res->[0] = $wself;
178	$req->[0](@$res);	324	$req->[0](@$res);
179	} else {	325	} else {
180	my $cb = shift @$req;	326	my $cb = shift @$req;
181	$wself->_error ($res->[1], @$req);	327	$@=$res->[1];
182	$cb->();	328	$cb->();
		329	if ($wself) { # cb() could have deleted it
		330	$wself->_error ($res->[1], @$req, $res->[2]); # error, request record, is_fatal
		331	}
		332	}
		333
		334	# no more queued requests, so cancel timeout
		335	if ($wself) {
		336	delete $wself->{timer}
		337	unless @{ $wself->{queue} };
183	}	338	}
184	}	339	}
185		340
186	} elsif (defined $len) {	341	} elsif (defined $len) {
187	$wself->_error ("unexpected eof", @caller, 1);	342	$wself->_error ("unexpected eof", @caller, 1);
188	} else {	343	} else {
		344	return if $err == EAGAIN;
189	$wself->_error ("read error: $!", @caller, 1);	345	$wself->_error ("read error: $err", @caller, 1);
190	}	346	}
191	});	347	});
192		348
193	$self->{ww_cb} = sub {	349	$self->{ww_cb} = sub {
		350	return unless $wself;
194	my $len = syswrite $client, $wself->{wbuf}	351	my $len = syswrite $client, $wself->{wbuf}
195	or return delete $wself->{ww};	352	or return delete $wself->{ww};
196		353
197	substr $wself->{wbuf}, 0, $len, "";	354	substr $wself->{wbuf}, 0, $len, "";
198	};	355	};
…		…
200	my $pid = fork;	357	my $pid = fork;
201		358
202	if ($pid) {	359	if ($pid) {
203	# parent	360	# parent
204	close $server;	361	close $server;
205
206	} elsif (defined $pid) {	362	} elsif (defined $pid) {
207	# child	363	# child
208	close $client;	364	my $serv_fno = fileno $server;
209	@_ = $server;
210	goto &serve;
211		365
		366	if ($self->{exec_server}) {
		367	fcntl $server, F_SETFD, 0; # don't close the server side
		368	exec "$^X -MAnyEvent::DBI -e AnyEvent::DBI::start_server $serv_fno";
		369	POSIX::_exit 124;
		370	} else {
		371	($_ != $serv_fno) && POSIX::close $_
		372	for $^F+1..$fd_max;
		373	serve $serv_fno;
		374	POSIX::_exit 0; # not reachable
		375	}
212	} else {	376	} else {
213	croak "fork: $!";	377	croak "fork: $!";
214	}	378	}
215		379
216	$self->_req (sub { }, (caller)[1,2], 1, req_open => $dbi, $user, $pass);	380	$self->{child_pid} = $pid;
		381	# set a connect timeout
		382	if ($self->{timeout}) {
		383	$self->{timer} = AnyEvent->timer (
		384	after => $self->{timeout},
		385	cb => sub { $wself && $wself->_timedout },
		386	);
		387	}
		388	$self->_req (
		389	($self->{on_connect} ? $self->{on_connect} : sub { }),
		390	(caller)[1,2],
		391	req_open => $dbi, $user, $pass, %dbi_args
		392	);
217		393
218	$self	394	$self
		395	}
		396
		397	sub _server_pid {
		398	shift->{child_pid}
		399	}
		400
		401	sub kill_child {
		402	my $self = shift;
		403	my $child_pid = delete $self->{child_pid};
		404	if ($child_pid) {
		405	# send SIGKILL in two seconds
		406	my $murder_timer = AnyEvent->timer (
		407	after => 2,
		408	cb => sub {
		409	kill 9, $child_pid;
		410	},
		411	);
		412
		413	# reap process
		414	my $kid_watcher;
		415	$kid_watcher = AnyEvent->child (
		416	pid => $child_pid ,
		417	cb => sub {
		418	# just hold on to this so it won't go away
		419	undef $kid_watcher;
		420	# cancel SIGKILL
		421	undef $murder_timer;
		422	},
		423	);
		424
		425	# SIGTERM = the beginning of the end
		426	kill TERM => $child_pid;
		427	}
		428	}
		429
		430	sub DESTROY {
		431	shift->kill_child;
219	}	432	}
220		433
221	sub _error {	434	sub _error {
222	my ($self, $error, $filename, $line, $fatal) = @_;	435	my ($self, $error, $filename, $line, $fatal) = @_;
223		436
		437	if ($fatal) {
224	delete $self->{rw};	438	delete $self->{rw};
225	delete $self->{ww};	439	delete $self->{ww};
226	delete $self->{fh};	440	delete $self->{fh};
		441	delete $self->{timer};
		442
		443	# for fatal errors call all enqueued callbacks with error
		444	while (my $req = shift @{$self->{queue}}) {
		445	$@ = $error;
		446	$req->[0]->();
		447	}
		448	$self->kill_child;
		449	}
227		450
228	$@ = $error;	451	$@ = $error;
229		452
		453	if ($self->{on_error}) {
230	$self->{on_error}($self, $filename, $line, $fatal)	454	$self->{on_error}($self, $filename, $line, $fatal)
231	if $self->{on_error};	455	} else {
232
233	die "$error at $filename, line $line\n";	456	die "$error at $filename, line $line\n";
		457	}
		458	}
		459
		460	=item $dbh->on_error ($cb->($dbh, $filename, $line, $fatal))
		461
		462	Sets (or clears, with C<undef>) the on_error handler.
		463
		464	=cut
		465
		466	sub on_error {
		467	$_[0]{on_error} = $_[1];
		468	}
		469
		470	=item $dbh->on_connect ($cb->($dbh))
		471
		472	Sets (or clears, with C<undef>) the on_connect handler.
		473
		474	=cut
		475
		476	sub on_connect {
		477	$_[0]{on_connect} = $_[1];
		478	}
		479
		480	=item $dbh->timeout ($seconds)
		481
		482	Sets (or clears, with C<undef>) the database timeout. Useful to extend the
		483	timeout when you are about to make a really long query.
		484
		485	=cut
		486
		487	sub timeout {
		488	my ($self, $timeout) = @_;
		489
		490	if ($timeout) {
		491	$self->{timeout} = $timeout;
		492	# reschedule timer if one was running
		493	if ($self->{timer}) {
		494	Scalar::Util::weaken (my $wself = $self);
		495	$self->{timer} = AnyEvent->timer (
		496	after => $self->{timeout},
		497	cb => sub { $wself && $wself->_timedout },
		498	);
		499	}
		500	} else {
		501	delete @{%$self}[qw(timer timeout)];
		502	}
		503	}
		504
		505	sub _timedout {
		506	my ($self) = @_;
		507
		508	my $req = shift @{ $self->{queue} };
		509
		510	if ($req) {
		511	my $cb = shift @$req;
		512	$@ = 'TIMEOUT';
		513	$cb->();
		514	$self->_error ('TIMEOUT', @$req, 1); # timeouts are always fatal
		515	} else {
		516	# shouldn't be possible to timeout without a pending request
		517	$self->_error ('TIMEOUT', 'NO_PENDING_WTF', 0, 1);
		518	}
234	}	519	}
235		520
236	sub _req {	521	sub _req {
237	my ($self, $cb, $filename, $line, $fatal) = splice @_, 0, 5, ();	522	my ($self, $cb, $filename, $line) = splice @_, 0, 4, ();
238		523
		524	if (!$self->{fh}) {
		525	my $err = $@ = 'NO DATABASE CONNECTION';
		526	$cb->();
		527	$self->_error ($err, $filename, $line, 1);
		528	return;
		529	}
		530
239	push @{ $self->{queue} }, [$cb, $filename, $line, $fatal];	531	push @{ $self->{queue} }, [$cb, $filename, $line ];
		532
		533	if ($self->{timeout} && !$self->{timer}) {
		534	Scalar::Util::weaken (my $wself = $self);
		535	$self->{timer} = AnyEvent->timer (
		536	after => $self->{timeout},
		537	cb => sub { $wself && $wself->_timedout },
		538	);
		539	}
240		540
241	$self->{wbuf} .= pack "L/a*", Storable::freeze \@_;	541	$self->{wbuf} .= pack "L/a*", Storable::freeze \@_;
242		542
243	unless ($self->{ww}) {	543	unless ($self->{ww}) {
244	my $len = syswrite $self->{fh}, $self->{wbuf};	544	my $len = syswrite $self->{fh}, $self->{wbuf};
…		…
248	$self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $self->{ww_cb})	548	$self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $self->{ww_cb})
249	if length $self->{wbuf};	549	if length $self->{wbuf};
250	}	550	}
251	}	551	}
252		552
253	=item $dbh->exec ("statement", @args, $cb->($rows, %extra))	553	=item $dbh->exec ("statement", @args, $cb->($dbh, \@rows, \%metadata))
254		554
255	Executes the given SQL statement with placeholders replaced by	555	Executes the given SQL statement with placeholders replaced by
256	C<@args>. The statement will be prepared and cached on the server side, so	556	C<@args>. The statement will be prepared and cached on the server side, so
257	using placeholders is compulsory.	557	using placeholders is compulsory.
258		558
259	The callback will be called with the result of C<fetchall_arrayref> as	559	The callback will be called with a weakened AnyEvent::DBI object as the first
260	first argument and possibly a hash reference with additional information.	560	argument and the result of C<fetchall_arrayref> as (or C<undef> if the
		561	statement wasn't a select statement) as the second argument. Third argument is
		562	a hash reference holding metadata about the request. Currently, the only key
		563	defined is C<$metadata->{rv}> holding the return value of
		564	C<execute>. Additional metadata might be added.
261		565
262	If an error occurs and the C<on_error> callback returns, then no arguments	566	If an error occurs and the C<on_error> callback returns, then no arguments
263	will be passed and C<$@> contains the error message.	567	will be passed and C<$@> contains the error message.
264		568
		569	=item $dbh->attr (attr_name, [ $attr_value ], $cb->($dbh, $new_value))
		570
		571	An accessor for the handle attributes, such as AutoCommit, RaiseError,
		572	PrintError, etc. If you provide an $attr_value, then the given attribute will
		573	be set to that value.
		574
		575	The callback will be passed the database handle and the
		576	attribute's value if successful. If accessing the attribute fails, then no
		577	arguments are passed to your callback, and $@ contains a description of the
		578	problem instead.
		579
		580	=item $dbh->begin_work ($cb->($dbh))
		581
		582	=item $dbh->commit ($cb->($dbh))
		583
		584	=item $dbh->rollback ($cb->($dbh))
		585
		586	The begin_work, commit, and rollback methods exopose the equivelant transaction
		587	control methods of the DBI. If something goes wrong, you will get no $dbh in
		588	your callaback, and will instead have an error to examine in $@.
		589
		590	=item $dbh->func ('string_which_yields_args_when_evaled', $func_name, $cb->($dbh, $result, $handle_error))
		591
		592	This gives access to database driver private methods. Because they are not
		593	standard you cannot always depend on the value of $result or $handle_error.
		594	Check the documentation for your specific driver/function combination to see
		595	what it returns.
		596
		597	Note that the first argument will be eval'ed to produce the argument list to
		598	the func() method. This must be done because the searialization protocol
		599	between the AnyEvent::DBI server process and your program does not support the
		600	passage of closures.
		601
		602	Here's an example to extend the query language in SQLite so it supports an
		603	intstr() function:
		604
		605	$cv = AnyEvent->condvar;
		606	$dbh->func(
		607	q{
		608	'instr',
		609	2,
		610	sub {
		611	my ($string, $search) = @_;
		612	return index $string, $search;
		613	},
		614	},
		615	'create_function',
		616	sub {return $cv->send($@) unless $_[0];$cv->send(undef,@_[1,2]);}
		617	);
		618	my ($err,$result,$handle_err) = $cv->recv();
		619	die "EVAL failed: $err" if $err;
		620	# otherwise, we can ignore $result and $handle_err for this particular func
		621
265	=cut	622	=cut
266		623
267	sub exec {	624	for my $cmd_name (qw(exec attr begin_work commit rollback func)) {
		625	eval 'sub ' . $cmd_name . '{
268	my $cb = pop;	626	my $cb = pop;
269	splice @_, 1, 0, $cb, (caller)[1,2], 0, "req_exec";	627	splice @_, 1, 0, $cb, (caller)[1,2], "req_' . $cmd_name . '";
270
271	goto &_req;	628	goto &_req;
		629	}';
272	}	630	}
273		631
274	=back	632	=back
275		633
276	=head1 SEE ALSO	634	=head1 SEE ALSO
277		635
278	L<AnyEvent>, L<DBI>.	636	L<AnyEvent>, L<DBI>.
279		637
280	=head1 AUTHOR	638	=head1 AUTHOR
281		639
282	Marc Lehmann <schmorp@schmorp.de>	640	Marc Lehmann <schmorp@schmorp.de>
283	http://home.schmorp.de/	641	http://home.schmorp.de/
		642
		643	Adam Rosenstein <adam@redcondor.com>
		644	http://www.redcondor.com/
284		645
285	=cut	646	=cut
286		647
287	1	648	1;
288		649

Diff Legend

-–
+Removed lines
-+
+Added lines
-<
+Changed lines
->
+Changed lines

Comparing AnyEvent-DBI/DBI.pm (file contents): Revision 1.3 by root, Fri Jun 6 16:13:07 2008 UTC vs. Revision 1.10 by root, Tue Jun 2 16:16:03 2009 UTC

Diff Legend

Comparing AnyEvent-DBI/DBI.pm (file contents):
Revision 1.3 by root, Fri Jun 6 16:13:07 2008 UTC vs.
Revision 1.10 by root, Tue Jun 2 16:16:03 2009 UTC