Koha/BackgroundJob.pm

   1 package Koha::BackgroundJob;
   2
   3 # This file is part of Koha.
   4 #
   5 # Koha is free software; you can redistribute it and/or modify it
   6 # under the terms of the GNU General Public License as published by
   7 # the Free Software Foundation; either version 3 of the License, or
   8 # (at your option) any later version.
   9 #
  10 # Koha is distributed in the hope that it will be useful, but
  11 # WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13 # GNU General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU General Public License
  16 # along with Koha; if not, see <http://www.gnu.org/licenses>.
  17
  18 use Modern::Perl;
  19 use JSON qw( decode_json encode_json );
  20 use Encode qw( encode_utf8 );
  21 use Carp qw( croak );
  22 use Net::Stomp;
  23 use Try::Tiny qw( catch try );
  24
  25 use C4::Context;
  26 use Koha::DateUtils qw( dt_from_string );
  27 use Koha::Exceptions;
  28 use Koha::Plugins;
  29 use Koha::Exceptions::BackgroundJob;
  30
  31 use base qw( Koha::Object );
  32
  33 =head1 NAME
  34
  35 Koha::BackgroundJob - Koha BackgroundJob Object class
  36
  37 This is a base class for BackgroundJob, some methods must be subclassed.
  38
  39 Example of usage:
  40
  41 Producer:
  42 my $job_id = Koha::BackgroundJob->enqueue(
  43     {
  44         job_type => $job_type,
  45         job_size => $job_size,
  46         job_args => $job_args
  47     }
  48 );
  49
  50 Consumer:
  51 Koha::BackgrounJobs->find($job_id)->process;
  52 See also C<misc/background_jobs_worker.pl> for a full example
  53
  54 =head1 API
  55
  56 =head2 Class methods
  57
  58 =head3 connect
  59
  60 Connect to the message broker using default guest/guest credential
  61
  62 =cut
  63
  64 sub connect {
  65     my ( $self );
  66     my $hostname = 'localhost';
  67     my $port = '61613';
  68     my $config = C4::Context->config('message_broker');
  69     my $credentials = {
  70         login => 'guest',
  71         passcode => 'guest',
  72     };
  73     if ($config){
  74         $hostname = $config->{hostname} if $config->{hostname};
  75         $port = $config->{port} if $config->{port};
  76         $credentials->{login} = $config->{username} if $config->{username};
  77         $credentials->{passcode} = $config->{password} if $config->{password};
  78         $credentials->{host} = $config->{vhost} if $config->{vhost};
  79     }
  80     my $stomp = Net::Stomp->new( { hostname => $hostname, port => $port } );
  81     $stomp->connect( $credentials );
  82     return $stomp;
  83 }
  84
  85 =head3 enqueue
  86
  87 Enqueue a new job. It will insert a new row in the DB table and notify the broker that a new job has been enqueued.
  88
  89 C<job_size> is the size of the job
  90 C<job_args> is the arguments of the job. It's a structure that will be JSON encoded.
  91
  92 Return the job_id of the newly created job.
  93
  94 =cut
  95
  96 sub enqueue {
  97     my ( $self, $params ) = @_;
  98
  99     my $job_type = $self->job_type;
 100     my $job_size = $params->{job_size};
 101     my $job_args = $params->{job_args};
 102     my $job_queue = $params->{job_queue} // 'default';
 103
 104     my $borrowernumber = (C4::Context->userenv) ? C4::Context->userenv->{number} : undef;
 105     my $json_args = encode_json $job_args;
 106
 107     $self->set(
 108         {
 109             status         => 'new',
 110             type           => $job_type,
 111             queue          => $job_queue,
 112             size           => $job_size,
 113             data           => $json_args,
 114             enqueued_on    => dt_from_string,
 115             borrowernumber => $borrowernumber,
 116         }
 117     )->store;
 118
 119     $job_args->{job_id} = $self->id;
 120
 121     my $conn;
 122     try {
 123         $conn = $self->connect;
 124     } catch {
 125         warn "Cannot connect to broker " . $_;
 126     };
 127     return unless $conn;
 128
 129     $json_args = encode_json $job_args;
 130     try {
 131         # This namespace is wrong, it must be a vhost instead.
 132         # But to do so it needs to be created on the server => much more work when a new Koha instance is created.
 133         # Also, here we just want the Koha instance's name, but it's not in the config...
 134         # Picking a random id (memcached_namespace) from the config
 135         my $namespace = C4::Context->config('memcached_namespace');
 136         $conn->send_with_receipt( { destination => sprintf("/queue/%s-%s", $namespace, $job_queue), body => $json_args } )
 137           or Koha::Exceptions::Exception->throw('Job has not been enqueued');
 138     } catch {
 139         $self->status('failed')->store;
 140         if ( ref($_) eq 'Koha::Exceptions::Exception' ) {
 141             $_->rethrow;
 142         } else {
 143             warn sprintf "The job has not been sent to the message broker: (%s)", $_;
 144         }
 145     };
 146
 147     return $self->id;
 148 }
 149
 150 =head3 process
 151
 152 Process the job!
 153
 154 =cut
 155
 156 sub process {
 157     my ( $self, $args ) = @_;
 158
 159     return {} if ref($self) ne 'Koha::BackgroundJob';
 160
 161     my $derived_class = $self->_derived_class;
 162
 163     $args ||= {};
 164
 165     return $derived_class->process( $args );
 166 }
 167
 168 =head3 start
 169
 170     $self->start;
 171
 172 Marks the job as started.
 173
 174 =cut
 175
 176 sub start {
 177     my ($self) = @_;
 178
 179     Koha::Exceptions::BackgroundJob::InconsistentStatus->throw(
 180         current_status  => $self->status,
 181         expected_status => 'new'
 182     ) unless $self->status eq 'new';
 183
 184     return $self->set(
 185         {
 186             started_on => \'NOW()',
 187             progress   => 0,
 188             status     => 'started',
 189         }
 190     )->store;
 191 }
 192
 193 =head3 step
 194
 195     $self->step;
 196
 197 Makes the job record a step has taken place.
 198
 199 =cut
 200
 201 sub step {
 202     my ($self) = @_;
 203
 204     Koha::Exceptions::BackgroundJob::InconsistentStatus->throw(
 205         current_status  => $self->status,
 206         expected_status => 'started'
 207     ) unless $self->status eq 'started';
 208
 209     # reached the end of the tasks already
 210     Koha::Exceptions::BackgroundJob::StepOutOfBounds->throw()
 211         unless $self->progress < $self->size;
 212
 213     return $self->progress( $self->progress + 1 )->store;
 214 }
 215
 216 =head3 finish
 217
 218     $self->finish;
 219
 220 Makes the job record as finished. If the job status is I<cancelled>, it is kept.
 221
 222 =cut
 223
 224 sub finish {
 225     my ( $self, $data ) = @_;
 226
 227     $self->status('finished') unless $self->status eq 'cancelled';
 228
 229     return $self->set(
 230         {
 231             ended_on => \'NOW()',
 232             data     => encode_json($data),
 233         }
 234     )->store;
 235 }
 236
 237 =head3 decoded_data
 238
 239     my $job_data = $self->decoded_data;
 240
 241 Returns the decoded JSON contents from $self->data.
 242
 243 =cut
 244
 245 sub decoded_data {
 246     my ($self) = @_;
 247
 248     return $self->data ? decode_json( $self->data ) : undef;
 249 }
 250
 251 =head3 set_encoded_data
 252
 253     $self->set_encoded_data( $data );
 254
 255 Serializes I<$data> as a JSON string and sets the I<data> attribute with it.
 256
 257 =cut
 258
 259 sub set_encoded_data {
 260     my ( $self, $data ) = @_;
 261
 262     return $self->data( $data ? encode_json($data) : undef );
 263 }
 264
 265 =head3 job_type
 266
 267 Return the job type of the job. Must be a string.
 268
 269 =cut
 270
 271 sub job_type { croak "This method must be subclassed" }
 272
 273 =head3 messages
 274
 275 Messages let during the processing of the job.
 276
 277 =cut
 278
 279 sub messages {
 280     my ( $self ) = @_;
 281
 282     my @messages;
 283     my $data_dump = decode_json encode_utf8 $self->data;
 284     if ( exists $data_dump->{messages} ) {
 285         @messages = @{ $data_dump->{messages} };
 286     }
 287
 288     return \@messages;
 289 }
 290
 291 =head3 report
 292
 293 Report of the job.
 294
 295 =cut
 296
 297 sub report {
 298     my ( $self ) = @_;
 299
 300     my $data_dump = decode_json encode_utf8 $self->data;
 301     return $data_dump->{report} || {};
 302 }
 303
 304 =head3 additional_report
 305
 306 Build additional variables for the job detail view.
 307
 308 =cut
 309
 310 sub additional_report {
 311     my ( $self ) = @_;
 312
 313     return {} if ref($self) ne 'Koha::BackgroundJob';
 314
 315     my $derived_class = $self->_derived_class;
 316
 317     return $derived_class->additional_report;
 318 }
 319
 320 =head3 cancel
 321
 322 Cancel a job.
 323
 324 =cut
 325
 326 sub cancel {
 327     my ( $self ) = @_;
 328     $self->status('cancelled')->store;
 329 }
 330
 331 =head2 Internal methods
 332
 333 =head3 _derived_class
 334
 335 =cut
 336
 337 sub _derived_class {
 338     my ( $self ) = @_;
 339     my $job_type = $self->type;
 340
 341     my $class = $self->type_to_class_mapping->{$job_type};
 342
 343     Koha::Exception->throw($job_type . ' is not a valid job_type')
 344         unless $class;
 345
 346     eval "require $class";
 347     return $class->_new_from_dbic( $self->_result );
 348 }
 349
 350 =head3 type_to_class_mapping
 351
 352     my $mapping = Koha::BackgrounJob->new->type_to_class_mapping;
 353
 354 Returns the available types to class mappings.
 355
 356 =cut
 357
 358 sub type_to_class_mapping {
 359     my ($self) = @_;
 360
 361     my $plugins_mapping = $self->plugin_types_to_classes;
 362
 363     return ($plugins_mapping)
 364       ? { %{ $self->core_types_to_classes }, %$plugins_mapping }
 365       : $self->core_types_to_classes;
 366 }
 367
 368 =head3 core_types_to_classes
 369
 370     my $mappings = Koha::BackgrounJob->new->core_types_to_classes
 371
 372 Returns the core background jobs types to class mappings.
 373
 374 =cut
 375
 376 sub core_types_to_classes {
 377     return {
 378         batch_authority_record_deletion     => 'Koha::BackgroundJob::BatchDeleteAuthority',
 379         batch_authority_record_modification => 'Koha::BackgroundJob::BatchUpdateAuthority',
 380         batch_biblio_record_deletion        => 'Koha::BackgroundJob::BatchDeleteBiblio',
 381         batch_biblio_record_modification    => 'Koha::BackgroundJob::BatchUpdateBiblio',
 382         batch_item_record_deletion          => 'Koha::BackgroundJob::BatchDeleteItem',
 383         batch_item_record_modification      => 'Koha::BackgroundJob::BatchUpdateItem',
 384         batch_hold_cancel                   => 'Koha::BackgroundJob::BatchCancelHold',
 385         update_elastic_index                => 'Koha::BackgroundJob::UpdateElasticIndex',
 386         update_holds_queue_for_biblios      => 'Koha::BackgroundJob::BatchUpdateBiblioHoldsQueue',
 387     };
 388 }
 389
 390 =head3 plugin_types_to_classes
 391
 392     my $mappings = Koha::BackgroundJob->new->plugin_types_to_classes
 393
 394 Returns the plugin-defined background jobs types to class mappings.
 395
 396 =cut
 397
 398 sub plugin_types_to_classes {
 399     my ($self) = @_;
 400
 401     unless ( exists $self->{_plugin_mapping} ) {
 402         my @plugins = Koha::Plugins->new()->GetPlugins( { method => 'background_tasks', } );
 403
 404         foreach my $plugin (@plugins) {
 405
 406             my $tasks    = $plugin->background_tasks;
 407             my $metadata = $plugin->get_metadata;
 408
 409             unless ( $metadata->{namespace} ) {
 410                 Koha::Logger->get->warn(
 411                         q{A plugin includes the 'background_tasks' method, }
 412                       . q{but doesn't provide the required 'namespace' }
 413                       . qq{method ($plugin->{class})} );
 414                 next;
 415             }
 416
 417             my $namespace = $metadata->{namespace};
 418
 419             foreach my $type ( keys %{$tasks} ) {
 420                 my $class = $tasks->{$type};
 421
 422                 # skip if conditions not met
 423                 next unless $type and $class;
 424
 425                 my $key = "plugin_$namespace" . "_$type";
 426
 427                 $self->{_plugin_mapping}->{$key} = $tasks->{$type};
 428             }
 429         }
 430     }
 431
 432     return $self->{_plugin_mapping};
 433 }
 434
 435 =head3 _type
 436
 437 =cut
 438
 439 sub _type {
 440     return 'BackgroundJob';
 441 }
 442
 443 1;