Attachment #132613 for bug #27344

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




    my $job_type = $self->job_type;
    my $job_size = $params->{job_size};
    my $job_args = $params->{job_args};
    my $job_queue = $params->{job_queue} // 'default';

    my $borrowernumber = C4::Context->userenv
        ? C4::Context->userenv->{number}

                {
                    status         => 'new',
                    type           => $job_type,
                    queue          => $job_queue,
                    size           => $job_size,
                    data           => $json_args,
                    enqueued_on    => dt_from_string,

                # Also, here we just want the Koha instance's name, but it's not in the config...
                # Picking a random id (memcached_namespace) from the config
                my $namespace = C4::Context->config('memcached_namespace');
                $conn->send_with_receipt( { destination => sprintf("/queue/%s-%s", $namespace, $job_queue), body => $json_args } )
                  or Koha::Exception->throw('Job has not been enqueued');
            } catch {
                if ( ref($_) eq 'Koha::Exception' ) {

Lines 125-131 sub enqueue { Link Here

(-)a/Koha/BackgroundJob/UpdateElasticIndex.pm (-1 / +2 lines)
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





=head1 SYNOPSIS

./background_jobs_worker.pl [--job-queue QUEUE]

=head1 DESCRIPTION

This script will connect to the Stomp server (RabbitMQ) and subscribe to the queues passed in parameter (or the 'default' queue), 
or if a Stomp server is not active it will poll the database every 10s for new jobs in the passed queue.

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.

=head1 OPTIONS

=over

=item B<--job-queue>

Repeatable. Give the job queues this worker will process.

The different values available are:

    default
    index
    update_elastic_index

=back



use Koha::BackgroundJobs;

my ( $help, @job_queues );
GetOptions(
    'h|help' => \$help,
    'job-queue=s' => \@job_queues,
) || pod2usage(1);

pod2usage(0) if $help;

unless (@job_queues) {
    push @job_queues, 'default';
}

my $conn;
try {
    $conn = Koha::BackgroundJob->connect;

    warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_;
};

my @available_job_types = qw(
    batch_biblio_record_modification
    batch_authority_record_modification
    batch_item_record_modification
    batch_biblio_record_deletion
    batch_authority_record_deletion
    batch_item_record_deletion
    batch_hold_cancel
    update_elastic_index
);

if ( @job_types ) {
    for my $job_type ( @job_types ) {
        pod2usage( -verbose => 1, -msg => sprintf "You specify an invalid --job-type value: %s\n", $job_type )
            unless grep { $_ eq $job_type } @available_job_types;
    }
} else {
    @job_types = @available_job_types;
}

if ( $conn ) {
    # FIXME cf note in Koha::BackgroundJob about $namespace
    my $namespace = C4::Context->config('memcached_namespace');
    for my $job_queue (@job_queues) {
        $conn->subscribe({ destination => sprintf("/queue/%s-%s", $namespace, $job_queue), ack => 'client' });
    }
}
while (1) {

        $conn->ack( { frame => $frame } ); # FIXME depending on success?

    } else {
        my $jobs = Koha::BackgroundJobs->search({ status => 'new', queue => \@job_queues });
        while ( my $job = $jobs->next ) {
            my $args = decode_json($job->data);
            process_job( $job, { job_id => $job->id, %$args } );
- 

Return to bug 27344

Lines 21-46 background_jobs_worker.pl - Worker script that will process backgroung jobs Link Here

(-)a/misc/background_jobs_worker.pl (-34 / +18 lines)
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	-

Lines 97-102 sub enqueue { Link Here

(-)a/Koha/BackgroundJob.pm (-1 / +3 lines)
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' ) {