|
Lines 53-59
use Try::Tiny;
Link Here
|
| 53 |
use Pod::Usage; |
53 |
use Pod::Usage; |
| 54 |
use Getopt::Long; |
54 |
use Getopt::Long; |
| 55 |
|
55 |
|
| 56 |
use Koha::Logger; |
|
|
| 57 |
use Koha::BackgroundJobs; |
56 |
use Koha::BackgroundJobs; |
| 58 |
|
57 |
|
| 59 |
my ( $help, @queues ); |
58 |
my ( $help, @queues ); |
|
Lines 94-139
while (1) {
Link Here
|
| 94 |
next; # will reconnect automatically |
93 |
next; # will reconnect automatically |
| 95 |
} |
94 |
} |
| 96 |
|
95 |
|
|
|
96 |
if ( $frame->command ne 'MESSAGE' ) { |
| 97 |
warn "Frame is not a message we should process"; |
| 98 |
warn $frame->body; |
| 99 |
next; |
| 100 |
} |
| 101 |
|
| 97 |
my $job; |
102 |
my $job; |
|
|
103 |
my $job_id; |
| 98 |
try { |
104 |
try { |
| 99 |
my $body = $frame->body; |
105 |
my $body = $frame->body; |
| 100 |
my $args = decode_json($body); # TODO Should this be from_json? Check utf8 flag. |
106 |
my $args = decode_json($body); # TODO Should this be from_json? Check utf8 flag. |
| 101 |
|
107 |
|
| 102 |
# FIXME This means we need to have create the DB entry before |
108 |
# FIXME This means we need to have create the DB entry before |
| 103 |
# It could work in a first step, but then we will want to handle job that will be created from the message received |
109 |
# It could work in a first step, but then we will want to handle job that will be created from the message received |
|
|
110 |
$job_id = $args->{job_id}; |
| 104 |
$job = Koha::BackgroundJobs->find($args->{job_id}); |
111 |
$job = Koha::BackgroundJobs->find($args->{job_id}); |
| 105 |
|
112 |
|
| 106 |
process_job( $job, $args ); |
113 |
$job->process( $args ); |
| 107 |
} catch { |
114 |
} catch { |
| 108 |
Koha::Logger->get->warn(sprintf "Job and/or frame not processed - %s", $_); |
115 |
Koha::Logger->get->warn(sprintf "Job and/or frame not processed - %s", $_); |
| 109 |
} finally { |
116 |
} finally { |
| 110 |
$job->status('failed')->store if $job && @_; |
117 |
$job->status('failed')->store if $job && @_; |
| 111 |
$conn->ack( { frame => $frame } ); |
118 |
$conn->ack( { frame => $frame } ); |
| 112 |
}; |
119 |
}; |
| 113 |
|
|
|
| 114 |
} else { |
120 |
} else { |
| 115 |
my $jobs = Koha::BackgroundJobs->search({ status => 'new', queue => \@queues }); |
121 |
my $jobs = Koha::BackgroundJobs->search({ status => 'new', queue => \@queues }); |
| 116 |
while ( my $job = $jobs->next ) { |
122 |
while ( my $job = $jobs->next ) { |
| 117 |
my $args = $job->json->decode($job->data); |
123 |
my $args = $job->json->decode($job->data); |
| 118 |
process_job( $job, { job_id => $job->id, %$args } ); |
124 |
$job->process( { job_id => $job->id, %$args } ); |
| 119 |
} |
125 |
} |
| 120 |
sleep 10; |
126 |
sleep 10; |
| 121 |
} |
127 |
} |
| 122 |
} |
128 |
} |
| 123 |
$conn->disconnect; |
129 |
$conn->disconnect; |
| 124 |
|
|
|
| 125 |
sub process_job { |
| 126 |
my ( $job, $args ) = @_; |
| 127 |
|
| 128 |
my $pid; |
| 129 |
if ( $pid = fork ) { |
| 130 |
wait; |
| 131 |
return; |
| 132 |
} |
| 133 |
|
| 134 |
die "fork failed!" unless defined $pid; |
| 135 |
|
| 136 |
$job->process( $args ); |
| 137 |
|
| 138 |
exit; |
| 139 |
} |
| 140 |
- |