Beekeeper
view release on metacpan or search on metacpan
lib/Beekeeper/Worker.pm view on Meta::CPAN
# Honor --debug command line option and 'debug' config option from pool.config.json
$LogLevel = LOG_DEBUG if $self->{_WORKER}->{debug} || $self->{_WORKER}->{config}->{debug};
my $log_handler = $self->log_handler;
$self->{_LOGGER} = $log_handler;
$Logger = sub {
# ($level, @messages) = @_
$log_handler->log(@_);
};
$SIG{__WARN__} = sub { $Logger->( LOG_WARN, @_ ) };
}
sub log_handler {
my $self = shift;
Beekeeper::Logger->new(
worker_class => ref $self,
foreground => $self->{_WORKER}->{foreground},
log_file => $self->{_WORKER}->{config}->{log_file},
host => $self->{_WORKER}->{hostname},
pool => $self->{_WORKER}->{pool_id},
_BUS => $self->{_BUS},
@_
);
}
sub __init_client {
my $self = shift;
my $bus_id = $self->{_WORKER}->{bus_id};
my $config = $self->{_WORKER}->{bus_config}->{$bus_id};
my $client = Beekeeper::Client->new(
%$config,
timeout => 0, # retry forever
on_error => sub {
my $errmsg = $_[0] || ""; $errmsg =~ s/\s+/ /sg;
log_fatal "Connection to $bus_id failed: $errmsg";
$self->stop_working;
},
);
$self->{_CLIENT} = $client->{_CLIENT};
$self->{_BUS} = $client->{_BUS};
$Beekeeper::Client::singleton = $self;
}
sub __init_worker {
my $self = shift;
$self->on_startup;
$self->__report_status;
AnyEvent->now_update;
$self->{_WORKER}->{report_status_timer} = AnyEvent->timer(
after => rand( $REPORT_STATUS_PERIOD ),
interval => $REPORT_STATUS_PERIOD,
cb => sub { $self->__report_status },
);
}
sub on_startup {
# Placeholder, intended to be overrided
my $class = ref $_[0];
log_fatal "Worker class $class doesn't define on_startup() method";
}
sub on_shutdown {
# Placeholder, can be overrided
}
sub authorize_request {
# Placeholder, must to be overrided
my $class = ref $_[0];
log_fatal "Worker class $class doesn't define authorize_request() method";
return undef; # do not authorize
}
sub accept_notifications {
my ($self, %args) = @_;
my $worker = $self->{_WORKER};
my $callbacks = $worker->{callbacks};
my ($file, $line) = (caller)[1,2];
my $at = "at $file line $line\n";
foreach my $fq_meth (keys %args) {
$fq_meth =~ m/^ ( [\w-]+ (?: \.[\w-]+ )* )
\. ( [\w-]+ | \* ) $/x or die "Invalid notification method '$fq_meth' $at";
my ($service, $method) = ($1, $2);
my $callback = $self->__get_cb_coderef($fq_meth, $args{$fq_meth});
die "Already accepting notifications '$fq_meth' $at" if exists $callbacks->{"msg.$fq_meth"};
$callbacks->{"msg.$fq_meth"} = $callback;
my $local_bus = $self->{_BUS}->{bus_role};
my $topic = "msg/$local_bus/$service/$method";
$topic =~ tr|.*|/#|;
$self->{_BUS}->subscribe(
topic => $topic,
on_publish => sub {
# ($payload_ref, $properties) = @_;
# Enqueue notification
push @{$worker->{task_queue_high}}, [ @_ ];
unless ($worker->{queued_tasks}) {
lib/Beekeeper/Worker.pm view on Meta::CPAN
$response = Beekeeper::JSONRPC::Error->server_error;
$response->{id} = $request->{id};
$json = $JSON->encode( $response );
$self->{_WORKER}->{error_count}++;
}
if ($request->{_deflate_response} && length($json) > $request->{_deflate_response}) {
my $compressed_json;
$DEFLATE->deflate(\$json, $compressed_json);
$DEFLATE->flush(\$compressed_json);
$DEFLATE->deflateReset;
$json = $compressed_json;
}
$self->{_BUS}->publish(
topic => $request->{_mqtt_properties}->{'response_topic'},
addr => $request->{_mqtt_properties}->{'addr'},
payload => \$json,
);
if (defined $timing_tasks) {
$BUSY_TIME += Time::HiRes::time - $BUSY_SINCE;
undef $BUSY_SINCE;
}
}
sub stop_accepting_notifications {
my ($self, @methods) = @_;
my ($file, $line) = (caller)[1,2];
my $at = "at $file line $line\n";
die "No method specified $at" unless @methods;
foreach my $fq_meth (@methods) {
$fq_meth =~ m/^ ( [\w-]+ (?: \.[\w-]+ )* )
\. ( [\w-]+ | \* ) $/x or die "Invalid method '$fq_meth' $at";
my ($service, $method) = ($1, $2);
my $worker = $self->{_WORKER};
unless (defined $worker->{callbacks}->{"msg.$fq_meth"}) {
log_warn "Not previously accepting notifications '$fq_meth' $at";
next;
}
my $local_bus = $self->{_BUS}->{bus_role};
my $topic = "msg/$local_bus/$service/$method";
$topic =~ tr|.*|/#|;
# Cannot remove callbacks right now, as new notifications could be in flight or be
# already queued. We must wait for unsubscription completion, and then until the
# notification queue is empty to ensure that all received ones were processed. And
# even then wait a bit more, as some brokers may send messages *after* unsubscription.
my $postpone = sub {
my $unsub_tmr; $unsub_tmr = AnyEvent->timer(
after => $UNSUBSCRIBE_LINGER, cb => sub {
delete $worker->{callbacks}->{"msg.$fq_meth"};
undef $unsub_tmr;
}
);
};
$self->{_BUS}->unsubscribe(
topic => $topic,
on_unsuback => sub {
my ($success, $prop) = @_;
log_error "Could not unsubscribe from topic '$topic' $at" unless $success;
my $postponed = $worker->{postponed} ||= [];
push @$postponed, $postpone;
AnyEvent::postpone { $self->__drain_task_queue };
}
);
}
}
sub stop_accepting_calls {
my ($self, @methods) = @_;
my ($file, $line) = (caller)[1,2];
my $at = "at $file line $line\n";
die "No method specified $at" unless @methods;
foreach my $fq_meth (@methods) {
$fq_meth =~ m/^ ( [\w-]+ (?: \.[\w-]+ )* )
\. ( [\w-]+ | \* ) $/x or die "Invalid remote call method '$fq_meth' $at";
my ($service, $method) = ($1, $2);
unless ($method eq '*') {
# Known limitation. As all calls for an entire service class are received
# through a single MQTT subscription (in order to load balance them), it is
# not possible to reject a single method. A workaround is to use a different
# class for each method that need to be individually rejected.
die "Cannot stop accepting individual methods, only '$service.*' is allowed $at";
}
my $worker = $self->{_WORKER};
my $callbacks = $worker->{callbacks};
my @cb_keys = grep { $_ =~ m/^req.\Q$service\E\b/ } keys %$callbacks;
unless (@cb_keys) {
log_warn "Not previously accepting remote calls '$fq_meth' $at";
next;
}
my $local_bus = $self->{_BUS}->{bus_role};
my $topic = "\$share/BKPR/req/$local_bus/$service";
$topic =~ tr|.*|/#|;
# Cannot remove callbacks right now, as new requests could be in flight or be already
# queued. We must wait for unsubscription completion, and then until the task queue
# is empty to ensure that all received requests were processed. And even then wait a
# bit more, as some brokers may send requests *after* unsubscription.
my $postpone = sub {
$worker->{stop_cv}->begin;
my $unsub_tmr; $unsub_tmr = AnyEvent->timer(
after => $UNSUBSCRIBE_LINGER, cb => sub {
delete $worker->{callbacks}->{$_} foreach @cb_keys;
delete $worker->{subscriptions}->{$service};
undef $unsub_tmr;
return unless $worker->{shutting_down};
if ($worker->{in_progress} > 0) {
# The task queue is empty now, but an asynchronous method handler is
# still busy processing some requests received previously. Wait for
# these requests to be completed before telling _work_forever to stop
my $wait_time = 60;
$worker->{stop_cv}->begin;
my $busy_tmr; $busy_tmr = AnyEvent->timer( after => 1, interval => 1, cb => sub {
unless ($worker->{in_progress} > 0 && --$wait_time > 0) {
undef $busy_tmr;
$worker->{stop_cv}->end;
}
});
}
# Tell _work_forever to stop
$worker->{stop_cv}->end;
}
);
};
$self->{_BUS}->unsubscribe(
topic => $topic,
on_unsuback => sub {
my ($success, $prop) = @_;
log_error "Could not unsubscribe from topic '$topic' $at" unless $success;
my $postponed = $worker->{postponed} ||= [];
push @$postponed, $postpone;
AnyEvent::postpone { $self->__drain_task_queue };
}
);
}
}
sub __work_forever {
my $self = shift;
# Called by WorkerPool->spawn_worker
eval {
my $worker = $self->{_WORKER};
$worker->{stop_cv} = AnyEvent->condvar;
# Blocks here until stop_working is called
$worker->{stop_cv}->recv;
$self->on_shutdown;
$self->__report_exit;
};
if ($@) {
log_fatal "Worker died: $@";
CORE::exit(255);
}
if ($self->{_BUS}->{is_connected}) {
$self->{_BUS}->disconnect;
}
}
lib/Beekeeper/Worker.pm view on Meta::CPAN
Makes a worker start accepting the specified RPC requests from the message bus.
C<$method> is a string with the format C<{service_class}.{method}>. A default
or fallback handler can be specified using a wildcard like C<{service_class}.*>.
C<$callback> is the method handler (a method name or a coderef) that will be
called when a request is received. When executed, the handler will receive two
parameters C<$params> (which contains the notification data itself) and C<$req>
which is a L<Beekeeper::JSONRPC::Request> object.
The value or reference returned by the handler will be sent back to the caller
as response (unless the response is deferred with C<$req-E<gt>async_response>).
The handler is executed within an eval block. If it dies the error will be logged
and the caller will receive a generic error response, but the worker will continue
running.
Example:
package MyWorker;
use Beekeeper::Worker ':log';
use base 'Beekeeper::Worker';
sub on_startup {
my ($self) = @_;
$self->accept_remote_calls(
'foo.inc' => 'increment', # call $self->increment for requests to 'foo.inc'
'foo.baz' => $coderef, # call $coderef->() for requests to 'foo.baz'
'foo.*' => 'fallback', # call $self->fallback for any other 'foo.*'
);
}
sub increment {
my ($self, $params, $req) = @_;
# $self is a MyWorker object
# $params is a ref to the parameters of the request
# $req is a Beekeeper::JSONRPC::Request object
log_warn "Got a call to foo.inc";
return $params->{number} + 1;
}
Remote calls can be processed concurrently by means of calling C<$req-E<gt>async_response>
to tell Beekeeper that the response for the request will be deferred until it is
available, freeing the worker to accept more requests. Once the response is ready,
it must be sent back to the caller with C<$req-E<gt>send_response>.
This handler process requests concurrently:
sub increment {
my ($self, $params, $req) = @_;
my $number = $params->{number};
$req->async_response;
my $t; $t = AnyEvent->timer( after => 1, cb => sub {
undef $t;
$req->send_response( $number + 1 );
});
}
Note that callback closures will not be executed in Beekeeper scope but in the event loop
one, so uncatched exceptions in these closures will cause the worker to die and be respawn.
Asynchronous method handlers use system resources more efficiently, but are significantly
harder to write and debug.
=head3 stop_accepting_notifications ( $method, ... )
Makes a worker stop accepting the specified notifications from the message bus.
C<$method> must be one of the strings used previously in C<accept_notifications>.
=head3 stop_accepting_calls ( $method, ... )
Makes a worker stop accepting the specified RPC requests from the message bus.
C<$method> must be one of the strings used previously in C<accept_remote_calls>.
=head3 stop_working
Makes a worker stop accepting new RPC requests, process all requests already
received, execute C<on_shutdown> method, and then exit.
This is the default signal handler for C<TERM> signal.
Please note that it is not possible to stop worker pools calling this method, as
WorkerPool will immediately respawn another worker after the current one exits.
=head1 SEE ALSO
L<Beekeeper::Client>, L<Beekeeper::Config>, L<Beekeeper::Logger>, L<Beekeeper::WorkerPool>.
=head1 AUTHOR
José Micó, C<jose.mico@gmail.com>
=head1 COPYRIGHT AND LICENSE
Copyright 2015-2023 José Micó.
This is free software; you can redistribute it and/or modify it under the same
terms as the Perl 5 programming language itself.
This software is distributed in the hope that it will be useful, but it is
provided “as is†and without any express or implied warranties. For details,
see the full text of the license in the file LICENSE.
=cut
( run in 0.856 second using v1.01-cache-2.11-cpan-6aa56a78535 )