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 107-112 sub enqueue { Link Here
107
                {
108
                {
108
                    status         => 'new',
109
                    status         => 'new',
109
                    type           => $job_type,
110
                    type           => $job_type,
111
                    queue          => $job_queue,
110
                    size           => $job_size,
112
                    size           => $job_size,
111
                    data           => $json_args,
113
                    data           => $json_args,
112
                    enqueued_on    => dt_from_string,
114
                    enqueued_on    => dt_from_string,
Lines 125-131 sub enqueue { Link Here
125
                # Also, here we just want the Koha instance's name, but it's not in the config...
127
                # Also, here we just want the Koha instance's name, but it's not in the config...
126
                # Picking a random id (memcached_namespace) from the config
128
                # Picking a random id (memcached_namespace) from the config
127
                my $namespace = C4::Context->config('memcached_namespace');
129
                my $namespace = C4::Context->config('memcached_namespace');
128
                $conn->send_with_receipt( { destination => sprintf("/queue/%s-%s", $namespace, $job_type), body => $json_args } )
130
                $conn->send_with_receipt( { destination => sprintf("/queue/%s-%s", $namespace, $job_queue), body => $json_args } )
129
                  or Koha::Exception->throw('Job has not been enqueued');
131
                  or Koha::Exception->throw('Job has not been enqueued');
130
            } catch {
132
            } catch {
131
                if ( ref($_) eq 'Koha::Exception' ) {
133
                if ( ref($_) eq 'Koha::Exception' ) {
(-)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