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