|
Lines 78-84
if ( $conn ) {
Link Here
|
| 78 |
# FIXME cf note in Koha::BackgroundJob about $namespace |
78 |
# FIXME cf note in Koha::BackgroundJob about $namespace |
| 79 |
my $namespace = C4::Context->config('memcached_namespace'); |
79 |
my $namespace = C4::Context->config('memcached_namespace'); |
| 80 |
for my $queue (@queues) { |
80 |
for my $queue (@queues) { |
| 81 |
$conn->subscribe({ destination => sprintf("/queue/%s-%s", $namespace, $queue), ack => 'client' }); |
81 |
$conn->subscribe({ |
|
|
82 |
destination => sprintf("/queue/%s-%s", $namespace, $queue), |
| 83 |
ack => 'client', |
| 84 |
'activemq.prefetchSize' => 1, |
| 85 |
}); |
| 82 |
} |
86 |
} |
| 83 |
} |
87 |
} |
| 84 |
while (1) { |
88 |
while (1) { |
|
Lines 90-127
while (1) {
Link Here
|
| 90 |
} |
94 |
} |
| 91 |
|
95 |
|
| 92 |
my $body = $frame->body; |
96 |
my $body = $frame->body; |
|
|
97 |
warn "GOT $body"; |
| 93 |
my $args = decode_json($body); # TODO Should this be from_json? Check utf8 flag. |
98 |
my $args = decode_json($body); # TODO Should this be from_json? Check utf8 flag. |
| 94 |
|
99 |
|
|
|
100 |
$conn->ack( { frame => $frame } ); # FIXME depending on success? |
| 101 |
|
| 95 |
# FIXME This means we need to have create the DB entry before |
102 |
# FIXME This means we need to have create the DB entry before |
| 96 |
# It could work in a first step, but then we will want to handle job that will be created from the message received |
103 |
# It could work in a first step, but then we will want to handle job that will be created from the message received |
| 97 |
my $job = Koha::BackgroundJobs->find($args->{job_id}); |
104 |
my $job = Koha::BackgroundJobs->find($args->{job_id}); |
| 98 |
|
105 |
|
| 99 |
process_job( $job, $args ); |
106 |
warn "Starting " . $args->{job_id}; |
| 100 |
$conn->ack( { frame => $frame } ); # FIXME depending on success? |
107 |
$job->start; |
|
|
108 |
sleep 1; |
| 109 |
warn "Finishing" . $args->{job_id}; |
| 110 |
$job->finish; |
| 101 |
|
111 |
|
| 102 |
} else { |
|
|
| 103 |
my $jobs = Koha::BackgroundJobs->search({ status => 'new', queue => \@queues }); |
| 104 |
while ( my $job = $jobs->next ) { |
| 105 |
my $args = $job->json->decode($job->data); |
| 106 |
process_job( $job, { job_id => $job->id, %$args } ); |
| 107 |
} |
| 108 |
sleep 10; |
| 109 |
} |
112 |
} |
| 110 |
} |
113 |
} |
| 111 |
$conn->disconnect; |
114 |
$conn->disconnect; |
| 112 |
|
|
|
| 113 |
sub process_job { |
| 114 |
my ( $job, $args ) = @_; |
| 115 |
|
| 116 |
my $pid; |
| 117 |
if ( $pid = fork ) { |
| 118 |
wait; |
| 119 |
return; |
| 120 |
} |
| 121 |
|
| 122 |
die "fork failed!" unless defined $pid; |
| 123 |
|
| 124 |
$job->process( $args ); |
| 125 |
|
| 126 |
exit; |
| 127 |
} |