misc/zebraqueue_daemon.pl

   1 #!/usr/bin/perl
   2
   3 # daemon to watch the zebraqueue and update zebra as needed
   4
   5 use strict;
   6 BEGIN {
   7     # find Koha's Perl modules
   8     # test carefully before changing this
   9     use FindBin;
  10     eval { require "$FindBin::Bin/kohalib.pl" };
  11 }
  12 use POE qw(Wheel::SocketFactory Wheel::ReadWrite Filter::Stream Driver::SysRW);
  13 use Unix::Syslog qw(:macros);
  14
  15 use C4::Context;
  16 use C4::Biblio;
  17 use C4::Search;
  18 use C4::AuthoritiesMarc;
  19 use XML::Simple;
  20 use utf8;
  21
  22
  23 my $dbh=C4::Context->dbh;
  24 my $ident = "Koha Zebraqueue ";
  25
  26 my $debug = 1;
  27 Unix::Syslog::openlog $ident, LOG_PID, LOG_LOCAL0;
  28
  29 Unix::Syslog::syslog LOG_INFO, "Starting Zebraqueue log at " . scalar localtime(time) . "\n";
  30
  31 sub handler_start {
  32
  33     # Starts session. Only ever called once only really used to set an alias
  34     # for the POE kernel
  35     my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
  36
  37     my $time = localtime(time);
  38     Unix::Syslog::syslog LOG_INFO, "$time POE Session ", $session->ID, " has started.\n";
  39
  40     # check status
  41 #    $kernel->yield('status_check');
  42     $kernel->yield('sleep');
  43 }
  44
  45 sub handler_sleep {
  46
  47     # can be used to slow down loop execution if needed
  48     my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
  49
  50     sleep 1;
  51     $kernel->yield('status_check');
  52 }
  53
  54 sub handler_check {
  55         # check if we need to do anything, at the moment just checks the zebraqueue, it could check other things too
  56         my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
  57         my $dbh=C4::Context->dbh;
  58         my $sth = $dbh->prepare("SELECT count(*) AS opcount FROM zebraqueue WHERE done = 0");
  59     $sth->execute;
  60         my $data = $sth->fetchrow_hashref();
  61         if ($data->{'opcount'} > 0){
  62                 Unix::Syslog::syslog LOG_INFO, "$data->{'opcount'} operations waiting to be run\n";
  63                 $sth->finish();
  64                 $kernel->yield('do_ops');
  65         }
  66         else {
  67                 $sth->finish();
  68                 $kernel->yield('sleep');
  69         }
  70 }
  71
  72 sub zebraop {
  73         # execute operations waiting in the zebraqueue
  74         my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
  75         my $dbh=C4::Context->dbh;
  76         my $readsth=$dbh->prepare("SELECT id,biblio_auth_number,operation,server FROM zebraqueue WHERE done=0");
  77         $readsth->execute();
  78         Unix::Syslog::syslog LOG_INFO, "Executing zebra operations\n";
  79         while (my $data = $readsth->fetchrow_hashref()){
  80                 eval {
  81                 my $ok = 0;
  82                 if ($data->{'operation'} =~ /delete/ ){
  83                         # 1st read the record in zebra, we have to get it from zebra as its no longer in the db
  84                         my $Zconn=C4::Context->Zconn($data->{'server'}, 0, 1,'','xml');
  85                         my $query = $Zconn->search_pqf( '@attr 1=Local-Number '.$data->{'biblio_auth_number'});
  86                         # then, delete the record
  87                         $ok=zebrado($query->record(0)->render(),$data->{'operation'},$data->{'server'},$data->{'biblio_auth_number'});
  88                 }
  89                 else {
  90                         # it is an update
  91                         # get the XML
  92                         my $marcxml;
  93                         if ($data->{'server'} eq "biblioserver") {
  94                                 my $marc = GetMarcBiblio($data->{'biblio_auth_number'});
  95                                 $marcxml = $marc->as_xml_record() if $marc;
  96                         }
  97                         elsif ($data->{'server'} eq "authorityserver") {
  98                                 $marcxml =C4::AuthoritiesMarc::GetAuthorityXML($data->{'biblio_auth_number'});
  99                         }
 100                         # check it's XML, just in case
 101                         eval {
 102                                 my $hashed=XMLin($marcxml);
 103                         }; ### is it a proper xml? broken xml may crash ZEBRA- slow but safe
 104                         ## it's Broken XML-- Should not reach here-- but if it does -lets protect ZEBRA
 105                         if ($@){
 106                              Unix::Syslog::syslog LOG_ERR, "$@";
 107                                 my $delsth=$dbh->prepare("UPDATE zebraqueue SET done=1 WHERE id =?");
 108                                 $delsth->execute($data->{'id'});
 109                                 next;
 110                         }
 111                         # ok, we have everything, do the operation in zebra !
 112                         $ok=zebrado($marcxml,$data->{'operation'},$data->{'server'},$data->{'biblio_auth_number'});
 113                 }
 114                 if ($ok == 1){
 115                         $dbh=C4::Context->dbh;
 116                         my $delsth;
 117                         # if it's a deletion, we can delete every request on this biblio : in case the user
 118                         # did a modif (or item deletion) just before biblio deletion, there are some specialUpdate
 119                         # that are pending and can't succeed, as we don't have the XML anymore
 120                         # so, delete everything for this biblionumber
 121                         if ($data->{'operation'} eq 'delete_record') {
 122                                 $delsth =$dbh->prepare("UPDATE zebraqueue SET done=1 WHERE biblio_auth_number =?");
 123                                 $delsth->execute($data->{'biblio_auth_number'});
 124                                 # if it's not a deletion, delete every pending specialUpdate for this biblionumber
 125                                 # in case the user add biblio, then X items, before this script runs
 126                                 # this avoid indexing X+1 times where just 1 is enough.
 127                         } else {
 128                                 $delsth =$dbh->prepare("UPDATE zebraqueue SET done=1 WHERE biblio_auth_number =? and operation='specialUpdate'");
 129                                 $delsth->execute($data->{'biblio_auth_number'});
 130                         }
 131                 }
 132                         };
 133                 if ($@){
 134                         Unix::Syslog::syslog LOG_ERR, "$@";
 135                 }
 136         }
 137         $readsth->finish();
 138         $kernel->yield('status_check');
 139 }
 140
 141 sub zebrado {
 142
 143     ###Accepts a $server variable thus we can use it to update  biblios, authorities or other zebra dbs
 144     my ($record,$op,$server,$biblionumber)=@_;
 145
 146     my @port;
 147
 148     my $tried=0;
 149     my $recon=0;
 150     my $reconnect=0;
 151 #    $record=Encode::encode("UTF-8",$record);
 152     my $shadow=$server."shadow";
 153
 154     $op = 'recordDelete' if $op eq 'delete_record';
 155 reconnect:
 156
 157     my $Zconn=C4::Context->Zconn($server, 0, 1);
 158     if ($record){
 159         my $Zpackage = $Zconn->package();
 160         $Zpackage->option(action => $op);
 161         $Zpackage->option(record => $record);
 162 #           $Zpackage->option(recordIdOpaque => $biblionumber) if $biblionumber;
 163 retry:
 164         $Zpackage->send("update");
 165         my($error, $errmsg, $addinfo, $diagset) = $Zconn->error_x();
 166         if ($error==10007 && $tried<3) {## timeout --another 30 looonng seconds for this update
 167             sleep 1;    ##  wait a sec!
 168             $tried=$tried+1;
 169             goto "retry";
 170         }elsif ($error==2 && $tried<2) {## timeout --temporary zebra error !whatever that means
 171             sleep 2;    ##  wait two seconds!
 172             $tried=$tried+1;
 173             goto "retry";
 174         }elsif($error==10004 && $recon==0){##Lost connection -reconnect
 175             sleep 1;    ##  wait a sec!
 176             $recon=1;
 177             $Zpackage->destroy();
 178             $Zconn->destroy();
 179             goto "reconnect";
 180         }elsif ($error){
 181             $Zpackage->destroy();
 182             $Zconn->destroy();
 183             return 0;
 184         }
 185         $Zpackage->send('commit');
 186     return 1;
 187     }
 188     return 0;
 189 }
 190
 191
 192 sub handler_stop {
 193     my $heap = $_[HEAP];
 194     my $time = localtime(time);
 195     Unix::Syslog::syslog LOG_INFO, "$time Session ", $_[SESSION]->ID, " has stopped.\n";
 196     delete $heap->{session};
 197 }
 198
 199 POE::Session->create(
 200     inline_states => {
 201         _start       => \&handler_start,
 202         sleep        => \&handler_sleep,
 203                 status_check => \&handler_check,
 204                 do_ops       => \&zebraop,
 205         _stop        => \&handler_stop,
 206     },
 207 );
 208
 209 # start the kernel
 210 $poe_kernel->run();
 211
 212 Unix::Syslog::closelog;
 213
 214 exit;