Merge branch 'devel' of h1dev:/srv/APKPM/
[APKPM.git] / lib / APKPM / Store.pm
1 package APKPM::Store;
2
3 use base qw(Gearman::Driver::Worker);
4 use Moose;
5 use Time::HiRes;
6 use Data::Dump qw(dump);
7 use DBD::Pg;
8
9 with 'APKPM::Gearman';
10
11 sub prefix { 'Store_' }
12
13 sub process_name {
14         my ( $self, $orig, $job_name ) = @_;
15         warn "# process_name $orig $job_name\n";
16         return "$orig ($job_name)";
17 }
18
19 sub dbh {
20         DBI->connect_cached('DBI:Pg:dbname=apkpm','','', {
21                 RaiseError => 1,
22                 AutoCommit => 1,
23         });
24 }
25
26 sub pg_insert {
27         my ( $self, $table, $h ) = @_;
28
29
30         my @c = keys %$h;
31
32         my $sql = "INSERT INTO $table (" . join(',',@c) . ') values (' . join(',', map { '?' } 0 .. $#c) . ')';
33         warn $sql;
34         my $sth = $self->dbh->prepare($sql);
35         $sth->execute( map { $h->{$_} } @c );
36 }
37
38 sub ADSL : Job : MinProcesses(1) : MaxProcesses(5) : Decode(d_json) {
39         my ( $self, $job, $workload ) = @_;
40
41         my $h = $workload->{ADSL} || die "no ADSL in ",dump $workload;
42         foreach my $c ( qw(ip username) ) {
43                 $h->{$c} = $workload->{$c} || die "no $c in ",dump $workload;
44         }
45
46         warn "# ADSL ", dump $h;
47
48         $self->pg_insert(adsl => $h)
49 }
50
51 sub ping : Job : MinProcesses(1) : MaxProcesses(5) : Decode(d_json) {
52         my ( $self, $job, $workload ) = @_;
53
54         $self->pg_insert(ping => $workload);
55 }
56
57 sub sql : Job : MinProcesses(1) : MaxProcesses(5) : Encode(e_json) {
58         my ( $self, $job, $workload ) = @_;
59
60         my $sth = $self->dbh->prepare($workload);
61         my $rows = $sth->execute;
62
63         warn "# $rows rows get_username_table $workload\n";
64
65         return {
66                 columns => $sth->{NAME},
67                 rows => $sth->fetchall_arrayref,
68         };
69 }
70
71 1;