View | Details | Raw Unified | Return to bug 27344
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
102
    my $borrowernumber = C4::Context->userenv
102
        ? C4::Context->userenv->{number}
103
        ? C4::Context->userenv->{number}
Lines 109-114 sub enqueue { Link Here
109
                {
110
                {
110
                    status         => 'new',
111
                    status         => 'new',
111
                    type           => $job_type,
112
                    type           => $job_type,
113
                    queue          => $job_queue,
112
                    size           => $job_size,
114
                    size           => $job_size,
113
                    data           => $json_args,
115
                    data           => $json_args,
114
                    enqueued_on    => dt_from_string,
116
                    enqueued_on    => dt_from_string,
Lines 127-133 sub enqueue { Link Here
127
                # Also, here we just want the Koha instance's name, but it's not in the config...
129
                # Also, here we just want the Koha instance's name, but it's not in the config...
128
                # Picking a random id (memcached_namespace) from the config
130
                # Picking a random id (memcached_namespace) from the config
129
                my $namespace = C4::Context->config('memcached_namespace');
131
                my $namespace = C4::Context->config('memcached_namespace');
130
                $conn->send_with_receipt( { destination => sprintf("/queue/%s-%s", $namespace, $job_type), body => $json_args } )
132
                $conn->send_with_receipt( { destination => sprintf("/queue/%s-%s", $namespace, $job_queue), body => $json_args } )
131
                  or Koha::Exception->throw('Job has not been enqueued');
133
                  or Koha::Exception->throw('Job has not been enqueued');
132
            } catch {
134
            } catch {
133
                if ( ref($_) eq 'Koha::Exception' ) {
135
                if ( ref($_) eq 'Koha::Exception' ) {
(-)a/Koha/BackgroundJob/UpdateElasticIndex.pm (-1 / +2 lines)
Lines 125-131 sub enqueue { Link Here
125
125
126
    $self->SUPER::enqueue({
126
    $self->SUPER::enqueue({
127
        job_size => scalar @record_ids,
127
        job_size => scalar @record_ids,
128
        job_args => {record_server => $record_server, record_ids => \@record_ids,}
128
        job_args => {record_server => $record_server, record_ids => \@record_ids},
129
        job_queue => 'index',
129
    });
130
    });
130
}
131
}
131
132
(-)a/misc/background_jobs_worker.pl (-34 / +18 lines)
Lines 21-46 background_jobs_worker.pl - Worker script that will process backgroung 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, @job_queues );
58
GetOptions(
59
GetOptions(
59
    'h|help' => \$help,
60
    'h|help' => \$help,
60
    'job-type:s' => \@job_types,
61
    'job-queue=s' => \@job_queues,
61
) || pod2usage(1);
62
) || pod2usage(1);
62
63
63
pod2usage(0) if $help;
64
pod2usage(0) if $help;
64
65
66
unless (@job_queues) {
67
    push @job_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 $job_queue (@job_queues) {
96
        $conn->subscribe({ destination => sprintf("/queue/%s-%s", $namespace, $job_type), ack => 'client' });
81
        $conn->subscribe({ destination => sprintf("/queue/%s-%s", $namespace, $job_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 => \@job_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 27344