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; |
57 |
|
58 |
|
58 |
my ( $help, @queues ); |
59 |
my ( $help, @queues ); |
|
|
60 |
my $max_processes = 10; |
59 |
GetOptions( |
61 |
GetOptions( |
|
|
62 |
'm|max-processes=i' => \$max_processes, |
60 |
'h|help' => \$help, |
63 |
'h|help' => \$help, |
61 |
'queue=s' => \@queues, |
64 |
'queue=s' => \@queues, |
62 |
) || pod2usage(1); |
65 |
) || pod2usage(1); |
Lines 74-79
try {
Link Here
|
74 |
warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_; |
77 |
warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_; |
75 |
}; |
78 |
}; |
76 |
|
79 |
|
|
|
80 |
my $pm = Parallel::ForkManager->new($max_processes); |
81 |
$SIG{CHLD} = sub { $pm->reap_finished_children }; |
82 |
|
77 |
if ( $conn ) { |
83 |
if ( $conn ) { |
78 |
# FIXME cf note in Koha::BackgroundJob about $namespace |
84 |
# FIXME cf note in Koha::BackgroundJob about $namespace |
79 |
my $namespace = C4::Context->config('memcached_namespace'); |
85 |
my $namespace = C4::Context->config('memcached_namespace'); |
Lines 96-127
while (1) {
Link Here
|
96 |
# It could work in a first step, but then we will want to handle job that will be created from the message received |
102 |
# 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}); |
103 |
my $job = Koha::BackgroundJobs->find($args->{job_id}); |
98 |
|
104 |
|
99 |
process_job( $job, $args ); |
105 |
$pm->start and next; |
|
|
106 |
$job->process( { job_id => $job->id, %$args } ); |
100 |
$conn->ack( { frame => $frame } ); # FIXME depending on success? |
107 |
$conn->ack( { frame => $frame } ); # FIXME depending on success? |
|
|
108 |
$pm->finish; |
101 |
|
109 |
|
102 |
} else { |
110 |
} else { |
103 |
my $jobs = Koha::BackgroundJobs->search({ status => 'new', queue => \@queues }); |
111 |
my $jobs = Koha::BackgroundJobs->search({ status => 'new', queue => \@queues }); |
104 |
while ( my $job = $jobs->next ) { |
112 |
while ( my $job = $jobs->next ) { |
|
|
113 |
$pm->start and next; |
105 |
my $args = $job->json->decode($job->data); |
114 |
my $args = $job->json->decode($job->data); |
106 |
process_job( $job, { job_id => $job->id, %$args } ); |
115 |
$job->process( { job_id => $job->id, %$args } ); |
|
|
116 |
$pm->finish; |
107 |
} |
117 |
} |
108 |
sleep 10; |
118 |
sleep 10; |
109 |
} |
119 |
} |
110 |
} |
120 |
} |
111 |
$conn->disconnect; |
121 |
$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 |
- |