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 use Redis;
   9
  10 with 'APKPM::Gearman';
  11
  12 sub prefix { 'Store_' }
  13
  14 sub process_name {
  15         my ( $self, $orig, $job_name ) = @_;
  16         warn "# process_name $orig $job_name\n";
  17         return "$orig ($job_name)";
  18 }
  19
  20 sub dbh {
  21         DBI->connect_cached('DBI:Pg:dbname=apkpm','','', {
  22                 RaiseError => 1,
  23                 AutoCommit => 1,
  24         });
  25 }
  26
  27 sub pg_insert {
  28         my ( $self, $table, $h ) = @_;
  29
  30         my $redis = Redis->new;
  31
  32         my @c;
  33
  34         if ( my $cols = $redis->get("pg.$table") ) {
  35                 @c = split(/\s+/,$cols);
  36         } else {
  37                 my $sth = $self->dbh->prepare( "select * from $table limit 1" );
  38                 $sth->execute;
  39                 @c = @{ $sth->{NAME_lc} };
  40                 $redis->set( "pg.$table" => join(' ',@c) );
  41                 $redis->expire( "pg.$table" => 5 * 60 ); # refresh every 5 min
  42         }
  43
  44         my $sql = "INSERT INTO $table (" . join(',',@c) . ') values (' . join(',', map { '?' } 0 .. $#c) . ')';
  45         warn $sql;
  46         my $sth = $self->dbh->prepare($sql);
  47
  48         my $h_lc;
  49         $h_lc->{ lc $_ } = $h->{$_} foreach keys %$h;
  50
  51         $sth->execute( map { $h_lc->{$_} } @c );
  52 }
  53
  54 sub insert : Job : Decode(d_json) : MinProcesses(0) {
  55         my ( $self, $job, $workload ) = @_;
  56         my $table = delete $workload->{_table} || die "no _table";
  57         $self->pg_insert($table => $workload);
  58 }
  59
  60 sub sql : Job : Encode(e_json) : MinProcesses(0) {
  61         my ( $self, $job, $workload ) = @_;
  62
  63         my $sth = $self->dbh->prepare($workload);
  64         my $rows = eval { $sth->execute };
  65         return { error => $@ } if $@;
  66
  67         warn "# $rows rows get_username_table $workload\n";
  68
  69         $rows = $sth->fetchall_arrayref;
  70         my @columns = @{ $sth->{NAME} };
  71
  72         # decode hash column
  73         my $hash_col;
  74         foreach ( 0 .. $#columns ) {
  75                 $hash_col = $_ if $columns[$_] eq 'h';
  76         }
  77         if ( defined $hash_col ) {
  78                 map {
  79                         $_->[$hash_col] = eval '{ ' . $_->[$hash_col] . ' }';
  80                 } @$rows
  81         }
  82
  83         return {
  84                 columns => \@columns,
  85                 rows => $rows,
  86                 hash_col => $hash_col,
  87         };
  88 }
  89
  90 1;