1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
=head1 NAME
AnyEvent::DBI::Slave - implement AnyEvent::DBI child/server processes
=head1 SYNOPSIS
# this module is normally loaded automatically
=head1 DESCRIPTION
This module contains the code that implements the DBI server part of
C<AnyEvent::DBI>. It is normally loaded automatically into each child
process, but can be loaded explicitly to save memory or startup time
(search for C<AnyEvent::DBI::Slave> in the L<AnyEvent::DBI> manpage).
=cut
package AnyEvent::DBI::Slave;
use common::sense;
use DBI ();
use Convert::Scalar ();
use CBOR::XS ();
use AnyEvent ();
our $VERSION = '3.04';
# this is the forked server code, could/should be bundled as it's own file
our $DBH;
our $STH;
sub req_pid {
[1, $$]
}
sub req_open {
my (undef, $dbi, $user, $pass, %attr) = @{+shift};
$DBH = DBI->connect ($dbi, $user, $pass, \%attr) or die $DBI::errstr;
[1, 1]
}
sub req_attr {
my (undef, $attr_name, @attr_val) = @{+shift};
$DBH->{$attr_name} = $attr_val[0]
if @attr_val;
[1, $DBH->{$attr_name}]
}
sub req_exec {
my (undef, $st, @args) = @{+shift};
$STH = $DBH->prepare_cached ($st, undef, 1)
or die [$DBI::errstr];
my $rv = $STH->execute (@args)
or die [$STH->errstr];
[1, $STH->{NUM_OF_FIELDS} ? $STH->fetchall_arrayref : undef, $rv]
}
sub req_stattr {
my (undef, $attr_name) = @{+shift};
[1, $STH->{$attr_name}]
}
sub req_begin_work {
[1, $DBH->begin_work || die [$DBI::errstr]]
}
sub req_commit {
[1, $DBH->commit || die [$DBI::errstr]]
}
sub req_rollback {
[1, $DBH->rollback || die [$DBI::errstr]]
}
sub req_func {
my (undef, $arg_string, $function) = @{+shift};
my @args = eval $arg_string;
die "error evaling \$dbh->func() arg_string: $@"
if $@;
my $rc = $DBH->func (@args, $function);
return [1, $rc, $DBI::err, $DBI::errstr];
}
sub serve($$) {
my ($fork_fh, $version, $fh) = @_;
$0 = "dbi slave";
close $fork_fh;
if ($VERSION != $version) {
Convert::Scalar::write_all $fh, CBOR::XS::encode_cbor
[undef, "AnyEvent::DBI version mismatch ($VERSION vs. $version)"];
return;
}
eval {
my $cbor = new CBOR::XS;
my $rbuf;
while (Convert::Scalar::extend_read $fh, $rbuf, 16000) {
for my $req ($cbor->incr_parse_multiple ($rbuf)) {
my $wbuf = eval { CBOR::XS::encode_cbor $req->[0]($req) };
$wbuf = CBOR::XS::encode_cbor [undef, ref $@ ? ("$@->[0]", $@->[1]) : ("$@", 1)]
if $@;
Convert::Scalar::write_all $fh, $wbuf
or die "unable to write results";
}
}
};
}
=head1 SEE ALSO
L<AnyEvent::DBI>.
=head1 AUTHOR AND CONTACT
Marc Lehmann <schmorp@schmorp.de> (current maintainer)
http://home.schmorp.de/
Adam Rosenstein <adam@redcondor.com>
http://www.redcondor.com/
=cut
1
|
