Bug 26742: Add configuration to koha-conf.xml for message broker
[koha.git] / 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( encode_json decode_json );
20 use Carp qw( croak );
21 use Net::Stomp;
22 use Try::Tiny;
23
24 use C4::Context;
25 use Koha::DateUtils qw( dt_from_string );
26 use Koha::Exceptions;
27 use Koha::BackgroundJob::BatchUpdateBiblio;
28 use Koha::BackgroundJob::BatchUpdateAuthority;
29
30 use base qw( Koha::Object );
31
32 =head1 NAME
33
34 Koha::BackgroundJob - Koha BackgroundJob Object class
35
36 This is a base class for BackgroundJob, some methods must be subclassed.
37
38 Example of usage:
39
40 Producer:
41 my $job_id = Koha::BackgroundJob->enqueue(
42     {
43         job_type => $job_type,
44         job_size => $job_size,
45         job_args => $job_args
46     }
47 );
48
49 Consumer:
50 Koha::BackgrounJobs->find($job_id)->process;
51 See also C<misc/background_jobs_worker.pl> for a full example
52
53 =head1 API
54
55 =head2 Class methods
56
57 =head3 connect
58
59 Connect to the message broker using default guest/guest credential
60
61 =cut
62
63 sub connect {
64     my ( $self );
65     my $hostname = 'localhost';
66     my $port = '61613';
67     my $config = C4::Context->config('message_broker');
68     my $credentials = {
69         login => 'guest',
70         passcode => 'guest',
71     };
72     if ($config){
73         $hostname = $config->{hostname} if $config->{hostname};
74         $port = $config->{port} if $config->{port};
75         $credentials->{login} = $config->{username} if $config->{username};
76         $credentials->{passcode} = $config->{password} if $config->{password};
77         $credentials->{host} = $config->{vhost} if $config->{vhost};
78     }
79     my $stomp = Net::Stomp->new( { hostname => $hostname, port => $port } );
80     my $frame = $stomp->connect( $credentials );
81     unless ($frame && $frame->command eq 'CONNECTED'){
82         if ($frame){
83             warn $frame->as_string;
84         }
85         die "Cannot connect to message broker";
86     }
87     return $stomp;
88 }
89
90 =head3 enqueue
91
92 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.
93
94 C<job_size> is the size of the job
95 C<job_args> is the arguments of the job. It's a structure that will be JSON encoded.
96
97 Return the job_id of the newly created job.
98
99 =cut
100
101 sub enqueue {
102     my ( $self, $params ) = @_;
103
104     my $job_type = $self->job_type;
105     my $job_size = $params->{job_size};
106     my $job_args = $params->{job_args};
107
108     my $borrowernumber = C4::Context->userenv->{number}; # FIXME Handle non GUI calls
109     my $json_args = encode_json $job_args;
110     my $job_id;
111     $self->_result->result_source->schema->txn_do(
112         sub {
113             $self->set(
114                 {
115                     status         => 'new',
116                     type           => $job_type,
117                     size           => $job_size,
118                     data           => $json_args,
119                     enqueued_on    => dt_from_string,
120                     borrowernumber => $borrowernumber,
121                 }
122             )->store;
123
124             $job_id = $self->id;
125             $job_args->{job_id} = $job_id;
126             $json_args = encode_json $job_args;
127
128             try {
129                 my $conn = $self->connect;
130                 # This namespace is wrong, it must be a vhost instead.
131                 # But to do so it needs to be created on the server => much more work when a new Koha instance is created.
132                 # Also, here we just want the Koha instance's name, but it's not in the config...
133                 # Picking a random id (memcached_namespace) from the config
134                 my $namespace = C4::Context->config('memcached_namespace');
135                 $conn->send_with_receipt( { destination => sprintf("/queue/%s-%s", $namespace, $job_type), body => $json_args } )
136                   or Koha::Exceptions::Exception->throw('Job has not been enqueued');
137             } catch {
138                 if ( ref($_) eq 'Koha::Exceptions::Exception' ) {
139                     $_->rethrow;
140                 } else {
141                     warn sprintf "The job has not been sent to the message broker: (%s)", $_;
142                 }
143             };
144         }
145     );
146
147     return $job_id;
148 }
149
150 =head3 process
151
152 Process the job!
153
154 =cut
155
156 sub process {
157     my ( $self, $args ) = @_;
158
159     my $job_type = $self->type;
160     return $job_type eq 'batch_biblio_record_modification'
161       ? Koha::BackgroundJob::BatchUpdateBiblio->process($args)
162       : $job_type eq 'batch_authority_record_modification'
163       ? Koha::BackgroundJob::BatchUpdateAuthority->process($args)
164       : Koha::Exceptions::Exception->throw('->process called without valid job_type');
165 }
166
167 =head3 job_type
168
169 Return the job type of the job. Must be a string.
170
171 =cut
172
173 sub job_type { croak "This method must be subclassed" }
174
175 =head3 messages
176
177 Messages let during the processing of the job.
178
179 =cut
180
181 sub messages {
182     my ( $self ) = @_;
183
184     my @messages;
185     my $data_dump = decode_json $self->data;
186     if ( exists $data_dump->{messages} ) {
187         @messages = @{ $data_dump->{messages} };
188     }
189
190     return \@messages;
191 }
192
193 =head3 report
194
195 Report of the job.
196
197 =cut
198
199 sub report {
200     my ( $self ) = @_;
201
202     my $data_dump = decode_json $self->data;
203     return $data_dump->{report};
204 }
205
206 =head3 cancel
207
208 Cancel a job.
209
210 =cut
211
212 sub cancel {
213     my ( $self ) = @_;
214     $self->status('cancelled')->store;
215 }
216
217 sub _type {
218     return 'BackgroundJob';
219 }
220
221 1;