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 |
- |