| Line 0
          
      
      
        Link Here | 
            
              |  |  | 1 | package Koha::OAI::Harvester; | 
            
              | 2 |  | 
            
              | 3 | # Copyright 2017 Prosentient Systems | 
            
              | 4 | # | 
            
              | 5 | # This file is part of Koha. | 
            
              | 6 | # | 
            
              | 7 | # Koha is free software; you can redistribute it and/or modify it | 
            
              | 8 | # under the terms of the GNU General Public License as published by | 
            
              | 9 | # the Free Software Foundation; either version 3 of the License, or | 
            
              | 10 | # (at your option) any later version. | 
            
              | 11 | # | 
            
              | 12 | # Koha is distributed in the hope that it will be useful, but | 
            
              | 13 | # WITHOUT ANY WARRANTY; without even the implied warranty of | 
            
              | 14 | # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | 
            
              | 15 | # GNU General Public License for more details. | 
            
              | 16 | # | 
            
              | 17 | # You should have received a copy of the GNU General Public License | 
            
              | 18 | # along with Koha; if not, see <http://www.gnu.org/licenses>. | 
            
              | 19 | # | 
            
              | 20 |  | 
            
              | 21 | use Modern::Perl; | 
            
              | 22 | use POE qw(Component::JobQueue); | 
            
              | 23 | use JSON; | 
            
              | 24 | use Sereal::Encoder; | 
            
              | 25 | use Sereal::Decoder; | 
            
              | 26 | use IO::Handle; | 
            
              | 27 | use File::Copy; | 
            
              | 28 | use File::Path qw(make_path remove_tree); | 
            
              | 29 | use DateTime; | 
            
              | 30 | use DateTime::Format::Strptime; | 
            
              | 31 |  | 
            
              | 32 | use C4::Context; | 
            
              | 33 | use Koha::OAI::Harvester::ImportQueues; | 
            
              | 34 |  | 
            
              | 35 | =head1 NAME | 
            
              | 36 |  | 
            
              | 37 | Koha::OAI::Harvester | 
            
              | 38 |  | 
            
              | 39 | =head1 SYNOPSIS | 
            
              | 40 |  | 
            
              | 41 | my $harvester = Koha::OAI::Harvester->spawn({ | 
            
              | 42 |     Downloader => $downloader, | 
            
              | 43 |     DownloaderWorkers => $download_workers, | 
            
              | 44 |     Importer => $importer, | 
            
              | 45 |     ImporterWorkers => $import_workers, | 
            
              | 46 |     ImportQueuePoll => $import_poll, | 
            
              | 47 |     logger => $logger, | 
            
              | 48 |     state_file => $statefile, | 
            
              | 49 |     spooldir => $spooldir, | 
            
              | 50 | }); | 
            
              | 51 |  | 
            
              | 52 | =head2 METHODS | 
            
              | 53 |  | 
            
              | 54 | =cut | 
            
              | 55 |  | 
            
              | 56 | my $day_granularity = DateTime::Format::Strptime->new( | 
            
              | 57 |     pattern   => '%F', | 
            
              | 58 | ); | 
            
              | 59 | my $seconds_granularity = DateTime::Format::Strptime->new( | 
            
              | 60 |     pattern   => '%FT%TZ', | 
            
              | 61 | ); | 
            
              | 62 |  | 
            
              | 63 | =head2 new | 
            
              | 64 |  | 
            
              | 65 | Create object | 
            
              | 66 |  | 
            
              | 67 | =cut | 
            
              | 68 |  | 
            
              | 69 | sub new { | 
            
              | 70 |     my ($class, $args) = @_; | 
            
              | 71 |     $args = {} unless defined $args; | 
            
              | 72 |     return bless ($args, $class); | 
            
              | 73 | } | 
            
              | 74 |  | 
            
              | 75 | =head2 spawn | 
            
              | 76 |  | 
            
              | 77 |     Creates POE sessions for handling tasks, downloading, and importing | 
            
              | 78 |  | 
            
              | 79 | =cut | 
            
              | 80 |  | 
            
              | 81 | sub spawn { | 
            
              | 82 |     my ($class, $args) = @_; | 
            
              | 83 |     my $self = $class->new($args); | 
            
              | 84 |     my $downloader = $self->{Downloader}; | 
            
              | 85 |     my $importer = $self->{Importer}; | 
            
              | 86 |     my $download_worker_limit = ( $self->{DownloaderWorkers} && int($self->{DownloaderWorkers}) ) ? $self->{DownloaderWorkers} : 1; | 
            
              | 87 |     my $import_worker_limit = ( $self->{ImporterWorkers} && int($self->{ImporterWorkers}) ) ? $self->{ImporterWorkers} : 1; | 
            
              | 88 |     my $import_queue_poll = ( $self->{ImportQueuePoll} && int($self->{ImportQueuePoll}) ) ? $self->{ImportQueuePoll} : 5; | 
            
              | 89 |  | 
            
              | 90 |     #NOTE: This job queue should always be created before the | 
            
              | 91 |     #harvester so that you can start download jobs immediately | 
            
              | 92 |     #upon spawning the harvester. | 
            
              | 93 |     POE::Component::JobQueue->spawn( | 
            
              | 94 |         Alias         => 'oai-downloader', | 
            
              | 95 |         WorkerLimit   => $download_worker_limit, | 
            
              | 96 |         Worker        => sub { | 
            
              | 97 |             my ($postback, $task) = @_; | 
            
              | 98 |             if ($downloader){ | 
            
              | 99 |                 if ($task->{status} eq "active"){ | 
            
              | 100 |                     $downloader->run({ | 
            
              | 101 |                         postback => $postback, | 
            
              | 102 |                         task => $task, | 
            
              | 103 |                    }); | 
            
              | 104 |                 } | 
            
              | 105 |             } | 
            
              | 106 |         }, | 
            
              | 107 |         Passive => {}, | 
            
              | 108 |     ); | 
            
              | 109 |  | 
            
              | 110 |     POE::Session->create( | 
            
              | 111 |         object_states => [ | 
            
              | 112 |             $self => { | 
            
              | 113 |                 _start => "on_start", | 
            
              | 114 |                 get_task => "get_task", | 
            
              | 115 |                 list_tasks => "list_tasks", | 
            
              | 116 |                 create_task => "create_task", | 
            
              | 117 |                 start_task => "start_task", | 
            
              | 118 |                 stop_task => "stop_task", | 
            
              | 119 |                 delete_task => "delete_task", | 
            
              | 120 |                 repeat_task => "repeat_task", | 
            
              | 121 |                 register => "register", | 
            
              | 122 |                 deregister => "deregister", | 
            
              | 123 |                 restore_state => "restore_state", | 
            
              | 124 |                 save_state => "save_state", | 
            
              | 125 |                 is_task_finished => "is_task_finished", | 
            
              | 126 |                 does_task_repeat => "does_task_repeat", | 
            
              | 127 |                 download_postback => "download_postback", | 
            
              | 128 |                 reset_imports_status => "reset_imports_status", | 
            
              | 129 |             }, | 
            
              | 130 |         ], | 
            
              | 131 |     ); | 
            
              | 132 |  | 
            
              | 133 |     POE::Component::JobQueue->spawn( | 
            
              | 134 |         Alias         => 'oai-importer', | 
            
              | 135 |         WorkerLimit   => $import_worker_limit, | 
            
              | 136 |         Worker        => sub { | 
            
              | 137 |             my $meta_postback = shift; | 
            
              | 138 |  | 
            
              | 139 |             #NOTE: We need to only retrieve queue items for active tasks. Otherwise, | 
            
              | 140 |             #the importer will just spin its wheels on inactive tasks and do nothing. | 
            
              | 141 |             my $active_tasks = $poe_kernel->call("harvester","list_tasks","active"); | 
            
              | 142 |             my @active_uuids = map { $_->{uuid} } @$active_tasks; | 
            
              | 143 |  | 
            
              | 144 |             my $rs = Koha::OAI::Harvester::ImportQueues->new->search({ | 
            
              | 145 |                 uuid => \@active_uuids, | 
            
              | 146 |                 status => "new" | 
            
              | 147 |             },{ | 
            
              | 148 |                 order_by => { -asc => 'id' }, | 
            
              | 149 |                 rows => 1, | 
            
              | 150 |             }); | 
            
              | 151 |             my $result = $rs->single; | 
            
              | 152 |             if ($result){ | 
            
              | 153 |                 $result->status("wip"); | 
            
              | 154 |                 $result->update; | 
            
              | 155 |                 my $task = { | 
            
              | 156 |                     id => $result->id, | 
            
              | 157 |                     uuid => $result->uuid, | 
            
              | 158 |                     result => $result->result, | 
            
              | 159 |                 }; | 
            
              | 160 |  | 
            
              | 161 |                 my $postback = $meta_postback->($task); | 
            
              | 162 |                 $importer->run({ | 
            
              | 163 |                     postback => $postback, | 
            
              | 164 |                     task => $task, | 
            
              | 165 |                 }); | 
            
              | 166 |             } | 
            
              | 167 |         }, | 
            
              | 168 |         Active => { | 
            
              | 169 |             PollInterval => $import_queue_poll, | 
            
              | 170 |             AckAlias => undef, | 
            
              | 171 |             AckState => undef, | 
            
              | 172 |         }, | 
            
              | 173 |     ); | 
            
              | 174 |  | 
            
              | 175 |     return; | 
            
              | 176 | } | 
            
              | 177 |  | 
            
              | 178 | =head2 on_start | 
            
              | 179 |  | 
            
              | 180 |     Internal method that starts the harvester | 
            
              | 181 |  | 
            
              | 182 | =cut | 
            
              | 183 |  | 
            
              | 184 | sub on_start { | 
            
              | 185 |     my ($self, $kernel, $heap) = @_[OBJECT, KERNEL,HEAP]; | 
            
              | 186 |     $kernel->alias_set('harvester'); | 
            
              | 187 |     $heap->{scoreboard} = {}; | 
            
              | 188 |  | 
            
              | 189 |     #Reset any 'wip' imports to 'new' so they can be re-tried. | 
            
              | 190 |     $kernel->call("harvester","reset_imports_status"); | 
            
              | 191 |  | 
            
              | 192 |     #Restore state from state file | 
            
              | 193 |     $kernel->call("harvester","restore_state"); | 
            
              | 194 | } | 
            
              | 195 |  | 
            
              | 196 | =head2 download_postback | 
            
              | 197 |  | 
            
              | 198 |     Internal method | 
            
              | 199 |  | 
            
              | 200 | =cut | 
            
              | 201 |  | 
            
              | 202 | #NOTE: This isn't really implemented at the moment, as it's not really necessary. | 
            
              | 203 | sub download_postback { | 
            
              | 204 |     my ($kernel, $request_packet, $response_packet) = @_[KERNEL, ARG0, ARG1]; | 
            
              | 205 |     my $message = $response_packet->[0]; | 
            
              | 206 | } | 
            
              | 207 |  | 
            
              | 208 | =head2 deregister | 
            
              | 209 |  | 
            
              | 210 |     Remove the worker session from the harvester's in-memory scoreboard, | 
            
              | 211 |     unset the downloading flag if downloading is completed. | 
            
              | 212 |  | 
            
              | 213 | =cut | 
            
              | 214 |  | 
            
              | 215 | sub deregister { | 
            
              | 216 |     my ($self, $kernel, $heap, $session, $sender, $type) = @_[OBJECT, KERNEL,HEAP,SESSION,SENDER,ARG0]; | 
            
              | 217 |  | 
            
              | 218 |     my $scoreboard = $heap->{scoreboard}; | 
            
              | 219 |  | 
            
              | 220 |     my $logger = $self->{logger}; | 
            
              | 221 |     $logger->debug("Start deregistering $sender as $type task"); | 
            
              | 222 |  | 
            
              | 223 |     my $task_uuid = delete $scoreboard->{session}->{$sender}; | 
            
              | 224 |     #NOTE: If you don't check each step of the hashref, autovivication can lead to surprises. | 
            
              | 225 |     if ($task_uuid){ | 
            
              | 226 |         if ($scoreboard->{task}->{$task_uuid}){ | 
            
              | 227 |             if ($scoreboard->{task}->{$task_uuid}->{$type}){ | 
            
              | 228 |                 delete $scoreboard->{task}->{$task_uuid}->{$type}->{$sender}; | 
            
              | 229 |             } | 
            
              | 230 |         } | 
            
              | 231 |     } | 
            
              | 232 |  | 
            
              | 233 |     my $task = $heap->{tasks}->{$task_uuid}; | 
            
              | 234 |     if ($task && $task->{status} && ($task->{status} eq "active") ){ | 
            
              | 235 |         #NOTE: Each task only has 1 download session, so we can now set/unset flags for the task. | 
            
              | 236 |         #NOTE: We should unset the downloading flag, if we're not going to repeat the task. | 
            
              | 237 |         if ($type eq "download"){ | 
            
              | 238 |             my $task_repeats = $kernel->call("harvester","does_task_repeat",$task_uuid); | 
            
              | 239 |             if ($task_repeats){ | 
            
              | 240 |                 my $interval = $task->{interval}; | 
            
              | 241 |  | 
            
              | 242 |                 $task->{effective_from} = delete $task->{effective_until}; | 
            
              | 243 |                 $task->{download_timer} = $kernel->delay_set("repeat_task", $interval, $task_uuid); | 
            
              | 244 |             } | 
            
              | 245 |             else { | 
            
              | 246 |                 $task->{downloading} = 0; | 
            
              | 247 |                 $kernel->call("harvester","save_state"); | 
            
              | 248 |                 $kernel->call("harvester","is_task_finished",$task_uuid); | 
            
              | 249 |             } | 
            
              | 250 |         } | 
            
              | 251 |         elsif ($type eq 'import'){ | 
            
              | 252 |             $kernel->call("harvester","is_task_finished",$task_uuid); | 
            
              | 253 |         } | 
            
              | 254 |     } | 
            
              | 255 |     $logger->debug("End deregistering $sender as $type task"); | 
            
              | 256 | } | 
            
              | 257 |  | 
            
              | 258 |  | 
            
              | 259 | =head2 is_task_finished | 
            
              | 260 |  | 
            
              | 261 |     This event handler checks if the task has finished downloading and importing record. | 
            
              | 262 |     If it is finished downloading and importing, the task is deleted from the harvester. | 
            
              | 263 |  | 
            
              | 264 |     This only applies to non-repeating tasks. | 
            
              | 265 |  | 
            
              | 266 | =cut | 
            
              | 267 |  | 
            
              | 268 | sub is_task_finished { | 
            
              | 269 |     my ($self, $kernel, $heap, $session, $uuid) = @_[OBJECT, KERNEL,HEAP,SESSION,ARG0]; | 
            
              | 270 |     my $task = $kernel->call("harvester","get_task",$uuid); | 
            
              | 271 |     if ($task && (! $task->{downloading}) ){ | 
            
              | 272 |         my $count = $self->get_import_count_for_task($uuid); | 
            
              | 273 |         if ( ! $count ) { | 
            
              | 274 |             #Clear this task out of the harvester as it's finished. | 
            
              | 275 |             $kernel->call("harvester","delete_task",$uuid); | 
            
              | 276 |             return 1; | 
            
              | 277 |         } | 
            
              | 278 |     } | 
            
              | 279 |     return 0; | 
            
              | 280 | } | 
            
              | 281 |  | 
            
              | 282 | =head2 register | 
            
              | 283 |  | 
            
              | 284 |     Internal method for registering a task on the harvester's scoreboard | 
            
              | 285 |  | 
            
              | 286 | =cut | 
            
              | 287 |  | 
            
              | 288 | sub register { | 
            
              | 289 |     my ($self, $kernel, $heap, $session, $sender, $type, $task_uuid) = @_[OBJECT, KERNEL,HEAP,SESSION,SENDER,ARG0,ARG1]; | 
            
              | 290 |     my $logger = $self->{logger}; | 
            
              | 291 |  | 
            
              | 292 |     my $scoreboard = $heap->{scoreboard}; | 
            
              | 293 |  | 
            
              | 294 |  | 
            
              | 295 |     if ($type && $task_uuid){ | 
            
              | 296 |         $logger->debug("Registering $sender as $type for $task_uuid"); | 
            
              | 297 |  | 
            
              | 298 |         my $task = $heap->{tasks}->{$task_uuid}; | 
            
              | 299 |         if ($task){ | 
            
              | 300 |  | 
            
              | 301 |             if ($type){ | 
            
              | 302 |                 #Register the task uuid with the session id as a key for later recall | 
            
              | 303 |                 $scoreboard->{session}->{$sender} = $task_uuid; | 
            
              | 304 |  | 
            
              | 305 |                 #Register the session id as a certain type of session for a task | 
            
              | 306 |                 $scoreboard->{task}->{$task_uuid}->{$type}->{$sender} = 1; | 
            
              | 307 |  | 
            
              | 308 |                 if ($type eq "download"){ | 
            
              | 309 |                     $task->{downloading} = 1; | 
            
              | 310 |  | 
            
              | 311 |                     my $task_repeats = $kernel->call("harvester","does_task_repeat",$task_uuid); | 
            
              | 312 |                     if ($task_repeats){ | 
            
              | 313 |  | 
            
              | 314 |                         #NOTE: Set an effective until, so we know we're not getting records any newer than | 
            
              | 315 |                         #this moment. | 
            
              | 316 |                         my $dt = DateTime->now(); | 
            
              | 317 |                         if ($dt){ | 
            
              | 318 |                             #NOTE: Ideally, I'd like to make sure that we can use 'seconds' granularity, but | 
            
              | 319 |                             #it's valid for 'from' to be null, so it's impossible to know from the data whether | 
            
              | 320 |                             #or not the repository will support the seconds granularity. | 
            
              | 321 |                             #NOTE: Ideally, it would be good to use either 'day' granularity or 'seconds' granularity, | 
            
              | 322 |                             #but at the moment the interval is expressed as seconds only. | 
            
              | 323 |                             $dt->set_formatter($seconds_granularity); | 
            
              | 324 |                             $task->{effective_until} = "$dt"; | 
            
              | 325 |                         } | 
            
              | 326 |                     } | 
            
              | 327 |  | 
            
              | 328 |                     $kernel->call("harvester","save_state"); | 
            
              | 329 |                 } | 
            
              | 330 |             } | 
            
              | 331 |         } | 
            
              | 332 |     } | 
            
              | 333 | } | 
            
              | 334 |  | 
            
              | 335 | =head2 does_task_repeat | 
            
              | 336 |  | 
            
              | 337 |     Internal method for checking if a task is supposed to repeat after it finishes | 
            
              | 338 |  | 
            
              | 339 | =cut | 
            
              | 340 |  | 
            
              | 341 | sub does_task_repeat { | 
            
              | 342 |     my ($self, $kernel, $heap, $session, $uuid) = @_[OBJECT, KERNEL,HEAP,SESSION,ARG0]; | 
            
              | 343 |     my $task = $kernel->call("harvester","get_task",$uuid); | 
            
              | 344 |     if ($task){ | 
            
              | 345 |         my $interval = $task->{interval}; | 
            
              | 346 |         my $parameters = $task->{parameters}; | 
            
              | 347 |         if ($parameters){ | 
            
              | 348 |             my $oai_pmh = $parameters->{oai_pmh}; | 
            
              | 349 |             if ($oai_pmh){ | 
            
              | 350 |                 if ( $interval && ($oai_pmh->{verb} eq "ListRecords") && (! $oai_pmh->{until}) ){ | 
            
              | 351 |                     return 1; | 
            
              | 352 |                 } | 
            
              | 353 |             } | 
            
              | 354 |         } | 
            
              | 355 |     } | 
            
              | 356 |     return 0; | 
            
              | 357 | } | 
            
              | 358 |  | 
            
              | 359 | =head2 reset_imports_status | 
            
              | 360 |  | 
            
              | 361 |     Internal method that resets the import queue from wip to new | 
            
              | 362 |  | 
            
              | 363 | =cut | 
            
              | 364 |  | 
            
              | 365 | sub reset_imports_status { | 
            
              | 366 |     my ($self, $kernel, $heap, $session) = @_[OBJECT, KERNEL,HEAP,SESSION]; | 
            
              | 367 |  | 
            
              | 368 |     my $rs = Koha::OAI::Harvester::ImportQueues->new->search({ | 
            
              | 369 |                 status => "wip", | 
            
              | 370 |     }); | 
            
              | 371 |     $rs->update({ | 
            
              | 372 |         status => "new", | 
            
              | 373 |     }); | 
            
              | 374 | } | 
            
              | 375 |  | 
            
              | 376 | =head2 restore_state | 
            
              | 377 |  | 
            
              | 378 |     Method to restore the harvester to a pre-existing state. | 
            
              | 379 |  | 
            
              | 380 | =cut | 
            
              | 381 |  | 
            
              | 382 | sub restore_state { | 
            
              | 383 |     my ($self, $kernel, $heap, $session) = @_[OBJECT, KERNEL,HEAP,SESSION]; | 
            
              | 384 |  | 
            
              | 385 |     my $state_file = $self->{state_file}; | 
            
              | 386 |     if ($state_file){ | 
            
              | 387 |         my $state_backup = "$state_file~"; | 
            
              | 388 |  | 
            
              | 389 |         #NOTE: If there is a state backup, it means we crashed while saving the state. Otherwise, | 
            
              | 390 |         #let's try the regular state file if it exists. | 
            
              | 391 |         my $file_to_restore = ( -f $state_backup ) ? $state_backup : ( ( -f $state_file ) ? $state_file : undef ); | 
            
              | 392 |         if ( $file_to_restore ){ | 
            
              | 393 |             my $opened = open( my $fh, '<', $file_to_restore ) or die "Couldn't open state: $!"; | 
            
              | 394 |             if ($opened){ | 
            
              | 395 |                 local $/; | 
            
              | 396 |                 my $in = <$fh>; | 
            
              | 397 |                 my $decoder = Sereal::Decoder->new; | 
            
              | 398 |                 my $state = $decoder->decode($in); | 
            
              | 399 |                 if ($state){ | 
            
              | 400 |                     if ($state->{tasks}){ | 
            
              | 401 |                         #Restore tasks from our saved state | 
            
              | 402 |                         $heap->{tasks} = $state->{tasks}; | 
            
              | 403 |                         foreach my $uuid ( keys %{$heap->{tasks}} ){ | 
            
              | 404 |                             my $task = $heap->{tasks}->{$uuid}; | 
            
              | 405 |  | 
            
              | 406 |                             #If tasks were still downloading, restart the task | 
            
              | 407 |                             if ( ($task->{status} && $task->{status} eq "active") && $task->{downloading} ){ | 
            
              | 408 |                                 $task->{status} = "new"; | 
            
              | 409 |                                 $kernel->call("harvester","start_task",$task->{uuid}); | 
            
              | 410 |                             } | 
            
              | 411 |                         } | 
            
              | 412 |                     } | 
            
              | 413 |                 } | 
            
              | 414 |             } | 
            
              | 415 |         } | 
            
              | 416 |     } | 
            
              | 417 | } | 
            
              | 418 |  | 
            
              | 419 | =head2 save_state | 
            
              | 420 |  | 
            
              | 421 |     Method to save the existing state of the harvester | 
            
              | 422 |  | 
            
              | 423 | =cut | 
            
              | 424 |  | 
            
              | 425 | sub save_state { | 
            
              | 426 |     my ($self, $kernel, $heap, $session) = @_[OBJECT, KERNEL,HEAP,SESSION]; | 
            
              | 427 |     my $state_file = $self->{state_file}; | 
            
              | 428 |     my $state_backup = "$state_file~"; | 
            
              | 429 |  | 
            
              | 430 |     #Make a backup of existing state record | 
            
              | 431 |     my $moved = move($state_file,$state_backup); | 
            
              | 432 |  | 
            
              | 433 |     my $opened = open(my $fh, ">", $state_file) or die "Couldn't save state: $!"; | 
            
              | 434 |     if ($opened){ | 
            
              | 435 |         $fh->autoflush(1); | 
            
              | 436 |         my $tasks = $heap->{tasks}; | 
            
              | 437 |         my $harvester_state = { | 
            
              | 438 |             tasks => $tasks, | 
            
              | 439 |         }; | 
            
              | 440 |         my $encoder = Sereal::Encoder->new; | 
            
              | 441 |         my $out = $encoder->encode($harvester_state); | 
            
              | 442 |         local $\; | 
            
              | 443 |         my $printed = print $fh $out; | 
            
              | 444 |         if ($printed){ | 
            
              | 445 |             close $fh; | 
            
              | 446 |             unlink($state_backup); | 
            
              | 447 |             return 1; | 
            
              | 448 |         } | 
            
              | 449 |     } | 
            
              | 450 |     return 0; | 
            
              | 451 | } | 
            
              | 452 |  | 
            
              | 453 | =head2 get_task | 
            
              | 454 |  | 
            
              | 455 |     This event handler returns a task from a harvester using the task's | 
            
              | 456 |     uuid as an argument. | 
            
              | 457 |  | 
            
              | 458 | =cut | 
            
              | 459 |  | 
            
              | 460 | sub get_task { | 
            
              | 461 |     my ($self, $kernel, $heap, $session, $uuid, $sender) = @_[OBJECT, KERNEL,HEAP,SESSION,ARG0, SENDER]; | 
            
              | 462 |  | 
            
              | 463 |     if ( ! $uuid && $sender ){ | 
            
              | 464 |         my $scoreboard = $heap->{scoreboard}; | 
            
              | 465 |         my $uuid_by_session = $scoreboard->{session}->{$sender}; | 
            
              | 466 |         if ($uuid_by_session){ | 
            
              | 467 |             $uuid = $uuid_by_session; | 
            
              | 468 |         } | 
            
              | 469 |     } | 
            
              | 470 |  | 
            
              | 471 |     my $tasks = $heap->{tasks}; | 
            
              | 472 |     if ($tasks && $uuid){ | 
            
              | 473 |         my $task = $tasks->{$uuid}; | 
            
              | 474 |         if ($task){ | 
            
              | 475 |             return $task; | 
            
              | 476 |         } | 
            
              | 477 |     } | 
            
              | 478 |     return 0; | 
            
              | 479 | } | 
            
              | 480 |  | 
            
              | 481 | =head2 get_import_count_for_task | 
            
              | 482 |  | 
            
              | 483 |     This gets the count of the number of imports exist for a certain task | 
            
              | 484 |  | 
            
              | 485 | =cut | 
            
              | 486 |  | 
            
              | 487 | sub get_import_count_for_task { | 
            
              | 488 |     my ($self,$uuid) = @_; | 
            
              | 489 |     my $count = undef; | 
            
              | 490 |     if ($uuid){ | 
            
              | 491 |         my $items = Koha::OAI::Harvester::ImportQueues->new->search({ | 
            
              | 492 |             uuid => $uuid, | 
            
              | 493 |         }); | 
            
              | 494 |         $count = $items->count; | 
            
              | 495 |     } | 
            
              | 496 |     return $count; | 
            
              | 497 | } | 
            
              | 498 |  | 
            
              | 499 | =head2 list_tasks | 
            
              | 500 |  | 
            
              | 501 |     This event handler returns a list of tasks that have been submitted | 
            
              | 502 |     to the harvester. It returns data like uuid, status, parameters, | 
            
              | 503 |     number of pending imports, etc. | 
            
              | 504 |  | 
            
              | 505 | =cut | 
            
              | 506 |  | 
            
              | 507 | sub list_tasks { | 
            
              | 508 |     my ($self, $kernel, $heap, $session, $status) = @_[OBJECT, KERNEL,HEAP,SESSION, ARG0]; | 
            
              | 509 |     my @tasks = (); | 
            
              | 510 |     foreach my $uuid (sort keys %{$heap->{tasks}}){ | 
            
              | 511 |         my $task = $heap->{tasks}->{$uuid}; | 
            
              | 512 |         my $items = Koha::OAI::Harvester::ImportQueues->new->search({ | 
            
              | 513 |             uuid => $uuid, | 
            
              | 514 |         }); | 
            
              | 515 |         my $count = $items->count // 0; | 
            
              | 516 |         $task->{pending_imports} = $count; | 
            
              | 517 |         if ( ( ! $status ) || ( $status && $status eq $task->{status} ) ){ | 
            
              | 518 |             push(@tasks, $task); | 
            
              | 519 |         } | 
            
              | 520 |  | 
            
              | 521 |     } | 
            
              | 522 |     return \@tasks; | 
            
              | 523 | } | 
            
              | 524 |  | 
            
              | 525 | =head2 create_task | 
            
              | 526 |  | 
            
              | 527 |     This event handler creates a spool directory for the task's imports. | 
            
              | 528 |     It also adds it to the harvester's memory and then saves memory to | 
            
              | 529 |     a persistent datastore. | 
            
              | 530 |  | 
            
              | 531 |     Newly created tasks have a status of "new". | 
            
              | 532 |  | 
            
              | 533 | =cut | 
            
              | 534 |  | 
            
              | 535 | sub create_task { | 
            
              | 536 |     my ($self, $kernel, $heap, $session, $incoming_task) = @_[OBJECT, KERNEL,HEAP,SESSION,ARG0]; | 
            
              | 537 |     my $logger = $self->{logger}; | 
            
              | 538 |     if ($incoming_task){ | 
            
              | 539 |         my $uuid = $incoming_task->{uuid}; | 
            
              | 540 |         if ( ! $heap->{tasks}->{$uuid} ){ | 
            
              | 541 |  | 
            
              | 542 |             #Step One: assign a spool directory to this task | 
            
              | 543 |             my $spooldir = $self->{spooldir} // "/tmp"; | 
            
              | 544 |             my $task_spooldir = "$spooldir/$uuid"; | 
            
              | 545 |             if ( ! -d $task_spooldir ){ | 
            
              | 546 |                 my $made_spool_directory = make_path($task_spooldir); | 
            
              | 547 |                 if ( ! $made_spool_directory ){ | 
            
              | 548 |                     if ($logger){ | 
            
              | 549 |                         $logger->warn("Unable to make task-specific spool directory at '$task_spooldir'"); | 
            
              | 550 |                     } | 
            
              | 551 |                     return 0; | 
            
              | 552 |                 } | 
            
              | 553 |             } | 
            
              | 554 |             $incoming_task->{spooldir} = $task_spooldir; | 
            
              | 555 |  | 
            
              | 556 |             #Step Two: assign new status | 
            
              | 557 |             $incoming_task->{status} = "new"; | 
            
              | 558 |  | 
            
              | 559 |             #Step Three: add task to harvester's memory | 
            
              | 560 |             $heap->{tasks}->{ $uuid } = $incoming_task; | 
            
              | 561 |  | 
            
              | 562 |             #Step Four: save state | 
            
              | 563 |             $kernel->call($session,"save_state"); | 
            
              | 564 |             return 1; | 
            
              | 565 |         } | 
            
              | 566 |     } | 
            
              | 567 |     return 0; | 
            
              | 568 | } | 
            
              | 569 |  | 
            
              | 570 | =head2 start_task | 
            
              | 571 |  | 
            
              | 572 |     This event handler marks a task as active in the harvester's memory, | 
            
              | 573 |     save the memory to a persistent datastore, then enqueues the task, | 
            
              | 574 |     so that it can be directed to the next available download worker. | 
            
              | 575 |  | 
            
              | 576 |     Newly started tasks have a status of "active". | 
            
              | 577 |  | 
            
              | 578 | =cut | 
            
              | 579 |  | 
            
              | 580 | sub start_task { | 
            
              | 581 |     my ($self, $session,$kernel,$heap,$uuid) = @_[OBJECT, SESSION,KERNEL,HEAP,ARG0]; | 
            
              | 582 |     my $task = $heap->{tasks}->{$uuid}; | 
            
              | 583 |     if ($task){ | 
            
              | 584 |         if ( $task->{status} ne "active" ){ | 
            
              | 585 |  | 
            
              | 586 |             #Clear any pre-existing error states | 
            
              | 587 |             delete $task->{error} if $task->{error}; | 
            
              | 588 |  | 
            
              | 589 |             #Step One: mark task as active | 
            
              | 590 |             $task->{status} = "active"; | 
            
              | 591 |  | 
            
              | 592 |             #Step Two: save state | 
            
              | 593 |             $kernel->call("harvester","save_state"); | 
            
              | 594 |  | 
            
              | 595 |             #Step Three: enqueue task | 
            
              | 596 |             $kernel->post("oai-downloader",'enqueue','download_postback', $task); | 
            
              | 597 |  | 
            
              | 598 |             return 1; | 
            
              | 599 |         } | 
            
              | 600 |     } | 
            
              | 601 |     return 0; | 
            
              | 602 | } | 
            
              | 603 |  | 
            
              | 604 | =head2 repeat_task | 
            
              | 605 |  | 
            
              | 606 |     This method re-queues a task for downloading | 
            
              | 607 |  | 
            
              | 608 | =cut | 
            
              | 609 |  | 
            
              | 610 | sub repeat_task { | 
            
              | 611 |     my ($self, $session,$kernel,$heap,$uuid) = @_[OBJECT, SESSION,KERNEL,HEAP,ARG0]; | 
            
              | 612 |     my $task = $heap->{tasks}->{$uuid}; | 
            
              | 613 |     if ($task){ | 
            
              | 614 |         my $interval = $task->{interval}; | 
            
              | 615 |         if ($task->{downloading} && $interval){ | 
            
              | 616 |             $kernel->post("oai-downloader",'enqueue','download_postback', $task); | 
            
              | 617 |         } | 
            
              | 618 |     } | 
            
              | 619 | } | 
            
              | 620 |  | 
            
              | 621 | =head2 stop_task | 
            
              | 622 |  | 
            
              | 623 |     This event handler prevents new workers from spawning, kills | 
            
              | 624 |     existing workers, and stops pending imports from being imported. | 
            
              | 625 |  | 
            
              | 626 |     Newly stopped tasks have a status of "stopped". | 
            
              | 627 |  | 
            
              | 628 | =cut | 
            
              | 629 |  | 
            
              | 630 | sub stop_task { | 
            
              | 631 |     my ($self, $kernel, $heap, $session, $sender, $task_uuid) = @_[OBJECT, KERNEL,HEAP,SESSION,SENDER,ARG0]; | 
            
              | 632 |  | 
            
              | 633 |     my $task = $heap->{tasks}->{$task_uuid}; | 
            
              | 634 |  | 
            
              | 635 |     if ($task && $task->{status} && $task->{status} ne "stopped" ){ | 
            
              | 636 |  | 
            
              | 637 |         #Step One: deactivate task, so no new workers can be started | 
            
              | 638 |         $task->{status} = "stopped"; | 
            
              | 639 |         #NOTE: You could also clear timers for new downloads, but that's probably unnecessary because of this step. | 
            
              | 640 |  | 
            
              | 641 |         #Step Two: kill workers | 
            
              | 642 |         my $scoreboard = $heap->{scoreboard}; | 
            
              | 643 |         my $session_types = $scoreboard->{task}->{$task_uuid}; | 
            
              | 644 |         if ($session_types){ | 
            
              | 645 |             foreach my $type ( keys %$session_types ){ | 
            
              | 646 |                 my $sessions = $session_types->{$type}; | 
            
              | 647 |                 if ($sessions){ | 
            
              | 648 |                     foreach my $session (keys %$sessions){ | 
            
              | 649 |                         if ($session){ | 
            
              | 650 |                             $kernel->signal($session, "cancel"); | 
            
              | 651 |                         } | 
            
              | 652 |                     } | 
            
              | 653 |                 } | 
            
              | 654 |             } | 
            
              | 655 |             #Remove the task uuid from the task key of the scoreboard | 
            
              | 656 |             delete $scoreboard->{task}->{$task_uuid}; | 
            
              | 657 |             #NOTE: The task uuid will still exist under the session key, | 
            
              | 658 |             #but the sessions will deregister themselves and clean that up for you. | 
            
              | 659 |         } | 
            
              | 660 |  | 
            
              | 661 |         #Step Three: stop pending imports for this task | 
            
              | 662 |         my $items = Koha::OAI::Harvester::ImportQueues->new->search({ | 
            
              | 663 |             uuid => $task_uuid, | 
            
              | 664 |         }); | 
            
              | 665 |         my $rows_updated = $items->update({ | 
            
              | 666 |             status => "stopped", | 
            
              | 667 |         }); | 
            
              | 668 |  | 
            
              | 669 |         #Step Four: save state | 
            
              | 670 |         $kernel->call("harvester","save_state"); | 
            
              | 671 |         return 1; | 
            
              | 672 |     } | 
            
              | 673 |     return 0; | 
            
              | 674 | } | 
            
              | 675 |  | 
            
              | 676 | =head2 delete_task | 
            
              | 677 |  | 
            
              | 678 |     Deleted tasks are stopped, pending imports are deleted from the | 
            
              | 679 |     database and file system, and then the task is removed from the harvester. | 
            
              | 680 |  | 
            
              | 681 | =cut | 
            
              | 682 |  | 
            
              | 683 | sub delete_task { | 
            
              | 684 |     my ($self, $kernel, $heap, $session, $task_uuid) = @_[OBJECT, KERNEL,HEAP,SESSION,ARG0]; | 
            
              | 685 |  | 
            
              | 686 |     my $task = $heap->{tasks}->{$task_uuid}; | 
            
              | 687 |     if ($task){ | 
            
              | 688 |         #Step One: stop task | 
            
              | 689 |         $kernel->call($session,"stop_task",$task_uuid); | 
            
              | 690 |  | 
            
              | 691 |         #Step Two: delete pending imports in database | 
            
              | 692 |         my $items = Koha::OAI::Harvester::ImportQueues->new->search({ | 
            
              | 693 |             uuid => $task_uuid, | 
            
              | 694 |         }); | 
            
              | 695 |         if ($items){ | 
            
              | 696 |             my $rows_deleted = $items->delete; | 
            
              | 697 |             #NOTE: shows 0E0 instead of 0 | 
            
              | 698 |         } | 
            
              | 699 |  | 
            
              | 700 |         #Step Three: remove task specific spool directory and files within it | 
            
              | 701 |         my $spooldir = $task->{spooldir}; | 
            
              | 702 |         if ($spooldir){ | 
            
              | 703 |             my $files_deleted = remove_tree($spooldir, { safe => 1 }); | 
            
              | 704 |         } | 
            
              | 705 |  | 
            
              | 706 |         delete $heap->{tasks}->{$task_uuid}; | 
            
              | 707 |  | 
            
              | 708 |         #Step Four: save state | 
            
              | 709 |         $kernel->call("harvester","save_state"); | 
            
              | 710 |         return 1; | 
            
              | 711 |     } | 
            
              | 712 |     return 0; | 
            
              | 713 | } | 
            
              | 714 |  | 
            
              | 715 | 1; |