Attachment #109477 for bug #22417

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




use JSON qw( encode_json decode_json );
use Carp qw( croak );
use Net::Stomp;
use Try::Tiny;

use C4::Context;
use Koha::DateUtils qw( dt_from_string );

            $job_args->{job_id} = $job_id;
            $json_args = encode_json $job_args;

            try {
                my $conn = $self->connect;
                # This namespace is wrong, it must be a vhost instead.
                # But to do so it needs to be created on the server => much more work when a new Koha instance is created.
                # 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_type), body => $json_args } )
                  or Koha::Exceptions::Exception->throw('Job has not been enqueued');
            } catch {
                if ( ref($_) eq 'Koha::Exceptions::Exception' ) {
                    $_->rethrow;
                } else {
                    warn sprintf "The job has not been sent to the message broker: (%s)", $_;
                }
            };
        }
    );


Lines 17-49 Link Here

(-)a/misc/background_jobs_worker.pl (-18 / +34 lines)
17		17
18	use Modern::Perl;	18	use Modern::Perl;
19	use JSON qw( encode_json decode_json );	19	use JSON qw( encode_json decode_json );
		20	use Try::Tiny;
20		21
21	use Koha::BackgroundJobs;	22	use Koha::BackgroundJobs;
22		23
23	my $conn = Koha::BackgroundJob->connect;	24	my $conn;
		25	try {
		26	$conn = Koha::BackgroundJob->connect;
		27	} catch {
		28	warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_;
		29	};
24		30
25	my @job_types = qw( batch_biblio_record_modification batch_authority_record_modification );	31	my @job_types = qw( batch_biblio_record_modification batch_authority_record_modification );
26		32
27	# FIXME cf note in Koha::BackgroundJob about $namespace	33	if ( $conn ) {
28	my $namespace = C4::Context->config('memcached_namespace');	34	# FIXME cf note in Koha::BackgroundJob about $namespace
29	for my $job_type ( @job_types ) {	35	my $namespace = C4::Context->config('memcached_namespace');
30	$conn->subscribe({ destination => sprintf("/queue/%s-%s", $namespace, $job_type), ack => 'client' });	36	for my $job_type ( @job_types ) {
		37	$conn->subscribe({ destination => sprintf("/queue/%s-%s", $namespace, $job_type), ack => 'client' });
		38	}
31	}	39	}
32	while (1) {	40	while (1) {
33	my $frame = $conn->receive_frame;	41	if ( $conn ) {
34	if ( !defined $frame ) {	42	my $frame = $conn->receive_frame;
35	# maybe log connection problems	43	if ( !defined $frame ) {
36	next; # will reconnect automatically	44	# maybe log connection problems
37	}	45	next; # will reconnect automatically
		46	}
38		47
39	my $body = $frame->body;	48	my $body = $frame->body;
40	my $args = decode_json($body);	49	my $args = decode_json($body);
41		50
42	# FIXME This means we need to have create the DB entry before	51	# FIXME This means we need to have create the DB entry before
43	# It could work in a first step, but then we will want to handle job that will be created from the message received	52	# It could work in a first step, but then we will want to handle job that will be created from the message received
44	my $job = Koha::BackgroundJobs->find($args->{job_id});	53	my $job = Koha::BackgroundJobs->find($args->{job_id});
45	my $success = $job->process( $args );	54	my $success = $job->process( $args );
46		55
47	$conn->ack( { frame => $frame } ); # FIXME depending on $success?	56	$conn->ack( { frame => $frame } ); # FIXME depending on $success?
		57	} else {
		58	my $jobs = Koha::BackgroundJobs->search({ status => 'new' });
		59	while ( my $job = $jobs->next ) {
		60	my $args = decode_json($job->data);
		61	$job->process( { job_id => $job->id, %$args } );
		62	}
		63	sleep 10;
		64	}
48	}	65	}
49	$conn->disconnect;	66	$conn->disconnect;
50	-

Return to bug 22417

Lines 19-24 use Modern::Perl; Link Here

(-)a/Koha/BackgroundJob.pm (-8 / +17 lines)
19	use JSON qw( encode_json decode_json );	19	use JSON qw( encode_json decode_json );
20	use Carp qw( croak );	20	use Carp qw( croak );
21	use Net::Stomp;	21	use Net::Stomp;
		22	use Try::Tiny;
22		23
23	use C4::Context;	24	use C4::Context;
24	use Koha::DateUtils qw( dt_from_string );	25	use Koha::DateUtils qw( dt_from_string );
Lines 104-117 sub enqueue { Link Here
104	$job_args->{job_id} = $job_id;	105	$job_args->{job_id} = $job_id;
105	$json_args = encode_json $job_args;	106	$json_args = encode_json $job_args;
106		107
107	my $conn = $self->connect;	108	try {
108	# This namespace is wrong, it must be a vhost instead.	109	my $conn = $self->connect;
109	# But to do so it needs to be created on the server => much more work when a new Koha instance is created.	110	# This namespace is wrong, it must be a vhost instead.
110	# Also, here we just want the Koha instance's name, but it's not in the config...	111	# But to do so it needs to be created on the server => much more work when a new Koha instance is created.
111	# Picking a random id (memcached_namespace) from the config	112	# Also, here we just want the Koha instance's name, but it's not in the config...
112	my $namespace = C4::Context->config('memcached_namespace');	113	# Picking a random id (memcached_namespace) from the config
113	$conn->send_with_receipt( { destination => sprintf("/queue/%s-%s", $namespace, $job_type), body => $json_args } )	114	my $namespace = C4::Context->config('memcached_namespace');
114	or Koha::Exceptions::Exception->throw('Job has not been enqueued');	115	$conn->send_with_receipt( { destination => sprintf("/queue/%s-%s", $namespace, $job_type), body => $json_args } )
		116	or Koha::Exceptions::Exception->throw('Job has not been enqueued');
		117	} catch {
		118	if ( ref($_) eq 'Koha::Exceptions::Exception' ) {
		119	$_->rethrow;
		120	} else {
		121	warn sprintf "The job has not been sent to the message broker: (%s)", $_;
		122	}
		123	};
115	}	124	}
116	);	125	);
117		126