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

(-)a/Koha/BackgroundJob.pm (-8 / +17 lines)
Lines 19-24 use Modern::Perl; Link Here
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
(-)a/misc/background_jobs_worker.pl (-18 / +34 lines)
Lines 17-49 Link Here
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