Line 0
Link Here
|
|
|
1 |
package Koha::BackgroundJob::Broker; |
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 Encode qw(); |
20 |
use JSON; |
21 |
use Net::Stomp; |
22 |
use Try::Tiny qw( catch try ); |
23 |
use Time::HiRes; |
24 |
|
25 |
use C4::Context; |
26 |
|
27 |
my $not_found_retries = {}; |
28 |
my $max_retries = $ENV{MAX_RETRIES} || 10; |
29 |
|
30 |
=head1 NAME |
31 |
|
32 |
Koha::BackgroundJob::Broker - Class to connect to RabbitMQ |
33 |
|
34 |
=head1 API |
35 |
|
36 |
=head2 Class methods |
37 |
|
38 |
=cut |
39 |
|
40 |
sub new { |
41 |
my $class = shift; |
42 |
my $self = {}; |
43 |
return bless $self, $class; |
44 |
} |
45 |
|
46 |
=head3 connect |
47 |
|
48 |
Connect to the message broker using default guest/guest credential |
49 |
|
50 |
=cut |
51 |
|
52 |
sub connect { |
53 |
my ($self) = @_; |
54 |
my $hostname = 'localhost'; |
55 |
my $port = '61613'; |
56 |
my $config = C4::Context->config('message_broker'); |
57 |
my $credentials = { |
58 |
login => 'guest', |
59 |
passcode => 'guest', |
60 |
}; |
61 |
if ($config) { |
62 |
$hostname = $config->{hostname} if $config->{hostname}; |
63 |
$port = $config->{port} if $config->{port}; |
64 |
$credentials->{login} = $config->{username} if $config->{username}; |
65 |
$credentials->{passcode} = $config->{password} if $config->{password}; |
66 |
$credentials->{host} = $config->{vhost} if $config->{vhost}; |
67 |
} |
68 |
my $stomp = Net::Stomp->new( { hostname => $hostname, port => $port } ); |
69 |
$stomp->connect($credentials); |
70 |
$self->{conn} = $stomp; |
71 |
return $self; |
72 |
} |
73 |
|
74 |
sub subscribe { |
75 |
my ( $self, @queues ) = @_; |
76 |
|
77 |
# FIXME cf note in Koha::BackgroundJob about $namespace |
78 |
my $namespace = C4::Context->config('memcached_namespace'); |
79 |
for my $queue (@queues) { |
80 |
$self->{conn}->subscribe( |
81 |
{ |
82 |
destination => sprintf( "/queue/%s-%s", $namespace, $queue ), |
83 |
ack => 'client', |
84 |
'prefetch-count' => 1, |
85 |
} |
86 |
); |
87 |
} |
88 |
} |
89 |
|
90 |
sub receive_frame { |
91 |
my ($self) = @_; |
92 |
return $self->{conn}->receive_frame; |
93 |
} |
94 |
|
95 |
=head3 enqueue_job |
96 |
|
97 |
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. |
98 |
|
99 |
C<job_size> is the size of the job |
100 |
C<job_args> is the arguments of the job. It's a structure that will be JSON encoded. |
101 |
|
102 |
Return the job_id of the newly created job. |
103 |
|
104 |
=cut |
105 |
|
106 |
sub enqueue_job { |
107 |
my ( $self, $params ) = @_; |
108 |
|
109 |
my $job_queue = $params->{job_queue} // 'default'; |
110 |
my $body = $params->{body}; |
111 |
my $conn; |
112 |
try { |
113 |
$conn = $self->connect; |
114 |
} catch { |
115 |
warn "Cannot connect to broker " . $_; |
116 |
}; |
117 |
return unless $conn; |
118 |
|
119 |
# This namespace is wrong, it must be a vhost instead. |
120 |
# But to do so it needs to be created on the server => much more work when a new Koha instance is created. |
121 |
# Also, here we just want the Koha instance's name, but it's not in the config... |
122 |
# Picking a random id (memcached_namespace) from the config |
123 |
my $namespace = C4::Context->config('memcached_namespace'); |
124 |
my $encoded_args = Encode::encode_utf8($body); # FIXME We should better leave this to Net::Stomp? |
125 |
my $destination = sprintf( "/queue/%s-%s", $namespace, $job_queue ); |
126 |
$self->{conn}->send_with_receipt( { destination => $destination, body => $encoded_args, persistent => 'true' } ) |
127 |
or Koha::Exceptions::Exception->throw('Job has not been enqueued'); |
128 |
} |
129 |
|
130 |
sub decode_frame { |
131 |
my ( $self, $frame ) = @_; |
132 |
|
133 |
return try { |
134 |
my $body = $frame->body; |
135 |
decode_json($body); # TODO Should this be from_json? Check utf8 flag. |
136 |
} catch { |
137 |
Koha::Logger->get( { interface => 'worker' } )->warn( sprintf "Frame not processed - %s", $_ ); |
138 |
return; |
139 |
}; |
140 |
} |
141 |
|
142 |
sub get_job { |
143 |
my ( $self, $params ) = @_; |
144 |
|
145 |
my $job_id = $params->{job_id}; |
146 |
my $frame = $params->{frame}; |
147 |
|
148 |
my $job = Koha::BackgroundJobs->find($job_id); |
149 |
|
150 |
if ( $job && $job->status ne 'new' ) { |
151 |
Koha::Logger->get( { interface => 'worker' } ) |
152 |
->warn( sprintf "Job %s has wrong status %s", $job_id, $job->status ); |
153 |
|
154 |
# nack without requeue, we do not want to process this frame again |
155 |
$self->{conn}->nack( { frame => $frame, requeue => 'false' } ); |
156 |
return; |
157 |
} |
158 |
|
159 |
unless ($job) { |
160 |
if ( ++$not_found_retries->{$job_id} >= $max_retries ) { |
161 |
Koha::Logger->get( { interface => 'worker' } )->warn( sprintf "Job %s not found, no more retry", $job_id ); |
162 |
|
163 |
# nack without requeue, we do not want to process this frame again |
164 |
$self->{conn}->nack( { frame => $frame, requeue => 'false' } ); |
165 |
return; |
166 |
} |
167 |
|
168 |
Koha::Logger->get( { interface => 'worker' } )->debug( sprintf "Job %s not found, will retry later", $job_id ); |
169 |
|
170 |
# nack to force requeue |
171 |
$self->{conn}->nack( { frame => $frame, requeue => 'true' } ); |
172 |
Time::HiRes::sleep(0.5); |
173 |
return; |
174 |
} |
175 |
|
176 |
} |
177 |
|
178 |
1; |