View | Details | Raw Unified | Return to bug 27783
Collapse All | Expand All

(-)a/Koha/BackgroundJob.pm (-1 / +3 lines)
Lines 97-102 sub enqueue { Link Here
97
    my $job_type = $self->job_type;
97
    my $job_type = $self->job_type;
98
    my $job_size = $params->{job_size};
98
    my $job_size = $params->{job_size};
99
    my $job_args = $params->{job_args};
99
    my $job_args = $params->{job_args};
100
    my $job_queue = $params->{job_queue} // 'default';
100
101
101
    my $borrowernumber = (C4::Context->userenv) ? C4::Context->userenv->{number} : undef;
102
    my $borrowernumber = (C4::Context->userenv) ? C4::Context->userenv->{number} : undef;
102
    my $json_args = encode_json $job_args;
103
    my $json_args = encode_json $job_args;
Lines 105-110 sub enqueue { Link Here
105
        {
106
        {
106
            status         => 'new',
107
            status         => 'new',
107
            type           => $job_type,
108
            type           => $job_type,
109
            queue          => $job_queue,
108
            size           => $job_size,
110
            size           => $job_size,
109
            data           => $json_args,
111
            data           => $json_args,
110
            enqueued_on    => dt_from_string,
112
            enqueued_on    => dt_from_string,
Lines 129-135 sub enqueue { Link Here
129
        # Also, here we just want the Koha instance's name, but it's not in the config...
131
        # Also, here we just want the Koha instance's name, but it's not in the config...
130
        # Picking a random id (memcached_namespace) from the config
132
        # Picking a random id (memcached_namespace) from the config
131
        my $namespace = C4::Context->config('memcached_namespace');
133
        my $namespace = C4::Context->config('memcached_namespace');
132
        $conn->send_with_receipt( { destination => sprintf("/queue/%s-%s", $namespace, $job_type), body => $json_args } )
134
        $conn->send_with_receipt( { destination => sprintf("/queue/%s-%s", $namespace, $job_queue), body => $json_args } )
133
          or Koha::Exceptions::Exception->throw('Job has not been enqueued');
135
          or Koha::Exceptions::Exception->throw('Job has not been enqueued');
134
    } catch {
136
    } catch {
135
        $self->status('failed')->store;
137
        $self->status('failed')->store;
(-)a/misc/background_jobs_worker.pl (-34 / +18 lines)
Lines 21-46 background_jobs_worker.pl - Worker script that will process background jobs Link Here
21
21
22
=head1 SYNOPSIS
22
=head1 SYNOPSIS
23
23
24
./background_jobs_worker.pl [--job-type]
24
./background_jobs_worker.pl [--job-queue QUEUE]
25
25
26
=head1 DESCRIPTION
26
=head1 DESCRIPTION
27
27
28
This script will connect to the Stomp server (RabbitMQ) and subscribe to the different destination queues available.
28
This script will connect to the Stomp server (RabbitMQ) and subscribe to the queues passed in parameter (or the 'default' queue),
29
You can specify some queues only (using --job-type) if you want to run several workers that will handle their own jobs.
29
or if a Stomp server is not active it will poll the database every 10s for new jobs in the passed queue.
30
31
You can specify some queues only (using --job-queue, which is repeatable) if you want to run several workers that will handle their own jobs.
30
32
31
=head1 OPTIONS
33
=head1 OPTIONS
32
34
33
=over
35
=over
34
36
35
=item B<--job-type>
37
=item B<--job-queue>
36
38
37
Give the job types this worker will process.
39
Repeatable. Give the job queues this worker will process.
38
40
39
The different values available are:
41
The different values available are:
40
42
41
    batch_biblio_record_modification
43
    default
42
    batch_authority_record_modification
44
    index
43
    update_elastic_index
44
45
45
=back
46
=back
46
47
Lines 54-67 use Getopt::Long; Link Here
54
55
55
use Koha::BackgroundJobs;
56
use Koha::BackgroundJobs;
56
57
57
my ( $help, @job_types );
58
my ( $help, @queues );
58
GetOptions(
59
GetOptions(
59
    'h|help' => \$help,
60
    'h|help' => \$help,
60
    'job-type:s' => \@job_types,
61
    'job-queue=s' => \@queues,
61
) || pod2usage(1);
62
) || pod2usage(1);
62
63
63
pod2usage(0) if $help;
64
pod2usage(0) if $help;
64
65
66
unless (@queues) {
67
    push @queues, 'default';
68
}
69
65
my $conn;
70
my $conn;
66
try {
71
try {
67
    $conn = Koha::BackgroundJob->connect;
72
    $conn = Koha::BackgroundJob->connect;
Lines 69-99 try { Link Here
69
    warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_;
74
    warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_;
70
};
75
};
71
76
72
my @available_job_types = qw(
73
    batch_biblio_record_modification
74
    batch_authority_record_modification
75
    batch_item_record_modification
76
    batch_biblio_record_deletion
77
    batch_authority_record_deletion
78
    batch_item_record_deletion
79
    batch_hold_cancel
80
    update_elastic_index
81
);
82
83
if ( @job_types ) {
84
    for my $job_type ( @job_types ) {
85
        pod2usage( -verbose => 1, -msg => sprintf "You specify an invalid --job-type value: %s\n", $job_type )
86
            unless grep { $_ eq $job_type } @available_job_types;
87
    }
88
} else {
89
    @job_types = @available_job_types;
90
}
91
92
if ( $conn ) {
77
if ( $conn ) {
93
    # FIXME cf note in Koha::BackgroundJob about $namespace
78
    # FIXME cf note in Koha::BackgroundJob about $namespace
94
    my $namespace = C4::Context->config('memcached_namespace');
79
    my $namespace = C4::Context->config('memcached_namespace');
95
    for my $job_type ( @job_types ) {
80
    for my $queue (@queues) {
96
        $conn->subscribe({ destination => sprintf("/queue/%s-%s", $namespace, $job_type), ack => 'client' });
81
        $conn->subscribe({ destination => sprintf("/queue/%s-%s", $namespace, $queue), ack => 'client' });
97
    }
82
    }
98
}
83
}
99
while (1) {
84
while (1) {
Lines 115-121 while (1) { Link Here
115
        $conn->ack( { frame => $frame } ); # FIXME depending on success?
100
        $conn->ack( { frame => $frame } ); # FIXME depending on success?
116
101
117
    } else {
102
    } else {
118
        my $jobs = Koha::BackgroundJobs->search({ status => 'new' });
103
        my $jobs = Koha::BackgroundJobs->search({ status => 'new', queue => \@queues });
119
        while ( my $job = $jobs->next ) {
104
        while ( my $job = $jobs->next ) {
120
            my $args = decode_json($job->data);
105
            my $args = decode_json($job->data);
121
            process_job( $job, { job_id => $job->id, %$args } );
106
            process_job( $job, { job_id => $job->id, %$args } );
122
- 

Return to bug 27783