data-worker.pl

   1 #!/usr/bin/perl
   2 use warnings;
   3 use strict;
   4
   5 use Data::Dump qw(dump);
   6 use DBI;
   7 use JSON::XS;
   8
   9 sub insert {
  10         my ($table,$job) = @_;
  11
  12         my $hash = decode_json $job->arg;
  13
  14         print "insert $table ",dump($hash),$/;
  15
  16         my $dbh = DBI->connect_cached('DBI:Pg:dbname=cloudstore','','', {
  17                 RaiseError => 1,
  18                 AutoCommit => 1,
  19         });
  20
  21         my @k = keys %$hash;
  22
  23         my $sql = sprintf 'INSERT INTO %s (%s) values (%s)', $table,
  24                 join(',',@k), join(',', map { '?' } @k);
  25
  26         warn "# $sql";
  27         my $sth = $dbh->prepare($sql);
  28         $sth->execute( map { $hash->{$_} } @k );
  29 };
  30
  31 use Gearman::Worker;
  32 my $worker = Gearman::Worker->new;
  33 $worker->job_servers('127.0.0.1:4730');
  34 $worker->register_function( rsync_transfer => sub { insert 'rsync_transfer' => shift } );
  35
  36 warn "$0 pid $$ waitng for jobs\n";
  37 $worker->work while 1;
  38