|
Lines 52-62
use JSON qw( decode_json );
Link Here
|
| 52 |
use Try::Tiny qw( catch try ); |
52 |
use Try::Tiny qw( catch try ); |
| 53 |
use Pod::Usage; |
53 |
use Pod::Usage; |
| 54 |
use Getopt::Long; |
54 |
use Getopt::Long; |
|
|
55 |
use Parallel::ForkManager; |
| 55 |
|
56 |
|
| 56 |
use Koha::BackgroundJobs; |
57 |
use Koha::BackgroundJobs; |
|
|
58 |
use C4::Context; |
| 57 |
|
59 |
|
| 58 |
my ( $help, @queues ); |
60 |
my ( $help, @queues ); |
|
|
61 |
|
| 62 |
my $max_processes = $ENV{MAX_PROCESSES}; |
| 63 |
$max_processes ||= C4::Context->config('background_jobs_worker')->{max_processes} if C4::Context->config('background_jobs_worker'); |
| 64 |
$max_processes ||= 1; |
| 65 |
|
| 59 |
GetOptions( |
66 |
GetOptions( |
|
|
67 |
'm|max-processes=i' => \$max_processes, |
| 60 |
'h|help' => \$help, |
68 |
'h|help' => \$help, |
| 61 |
'queue=s' => \@queues, |
69 |
'queue=s' => \@queues, |
| 62 |
) || pod2usage(1); |
70 |
) || pod2usage(1); |
|
Lines 74-79
try {
Link Here
|
| 74 |
warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_; |
82 |
warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_; |
| 75 |
}; |
83 |
}; |
| 76 |
|
84 |
|
|
|
85 |
my $pm = Parallel::ForkManager->new($max_processes); |
| 86 |
$SIG{CHLD} = sub { $pm->reap_finished_children }; |
| 87 |
|
| 77 |
if ( $conn ) { |
88 |
if ( $conn ) { |
| 78 |
# FIXME cf note in Koha::BackgroundJob about $namespace |
89 |
# FIXME cf note in Koha::BackgroundJob about $namespace |
| 79 |
my $namespace = C4::Context->config('memcached_namespace'); |
90 |
my $namespace = C4::Context->config('memcached_namespace'); |
|
Lines 97-127
while (1) {
Link Here
|
| 97 |
my $job = Koha::BackgroundJobs->find($args->{job_id}); |
108 |
my $job = Koha::BackgroundJobs->find($args->{job_id}); |
| 98 |
|
109 |
|
| 99 |
$conn->ack( { frame => $frame } ); # Acknowledge the message was recieved |
110 |
$conn->ack( { frame => $frame } ); # Acknowledge the message was recieved |
| 100 |
process_job( $job, $args ); |
111 |
$pm->start and next; |
|
|
112 |
$job->process( { job_id => $job->id, %$args } ); |
| 113 |
$pm->finish; |
| 101 |
|
114 |
|
| 102 |
} else { |
115 |
} else { |
| 103 |
my $jobs = Koha::BackgroundJobs->search({ status => 'new', queue => \@queues }); |
116 |
my $jobs = Koha::BackgroundJobs->search({ status => 'new', queue => \@queues }); |
| 104 |
while ( my $job = $jobs->next ) { |
117 |
while ( my $job = $jobs->next ) { |
|
|
118 |
$pm->start and next; |
| 105 |
my $args = $job->json->decode($job->data); |
119 |
my $args = $job->json->decode($job->data); |
| 106 |
process_job( $job, { job_id => $job->id, %$args } ); |
120 |
$job->process( { job_id => $job->id, %$args } ); |
|
|
121 |
$pm->finish; |
| 107 |
} |
122 |
} |
| 108 |
sleep 10; |
123 |
sleep 10; |
| 109 |
} |
124 |
} |
| 110 |
} |
125 |
} |
| 111 |
$conn->disconnect; |
126 |
$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 |
} |
| 128 |
- |