File Coverage

lib/Beekeeper/Service/Sinkhole/Worker.pm

Criterion	Covered	Total	%
statement	18	59	30.5
branch	0	16	0.0
condition			n/a
subroutine	6	15	40.0
pod	4	7	57.1
total	28	97	28.8

package Beekeeper::Service::Sinkhole::Worker;

use strict;
use warnings;

our $VERSION = '0.10';

use Beekeeper::Worker ':log';
use base 'Beekeeper::Worker';

use Beekeeper::JSONRPC::Error;
use JSON::XS;


sub authorize_request {
    my ($self, $req) = @_;

    if ($req->{method} eq '_bkpr.sinkhole.unserviced_queues') {

        return unless $self->__has_authorization_token('BKPR_SYSTEM');
    }

    # All requests will be rejected later by reject_request
    return BKPR_REQUEST_AUTHORIZED;
}

sub on_startup {
    my $self = shift;

    $self->{Draining} = {};

    $self->accept_notifications(
        '_bkpr.sinkhole.unserviced_queues' => 'on_unserviced_queues',
    );

    my $local_bus = $self->{_BUS}->{bus_role};

    # Watch the Supervisor data traffic in order to stop rejecting
    # requests as soon as a worker handling these becomes online

    my $topic = "msg/$local_bus/_sync/workers/set";

    $self->{_BUS}->subscribe(
        topic      => $topic,
        on_publish => sub {
            my ($payload_ref, $properties) = @_;
            $self->on_worker_status( decode_json($$payload_ref)->[1] );
        },
        on_suback => sub {
            my ($success) = @_;
            log_error "Could not subscribe to topic '$topic'" unless $success;
        }
    );

    log_info "Ready";
}

sub on_shutdown {
    my $self = shift;

    log_info "Stopped";
}

sub log_handler {
    my $self = shift;

    # Use pool's logfile
    $self->SUPER::log_handler( foreground => 1 );
}

sub on_unserviced_queues {
    my ($self, $params) = @_;

    my $queues = $params->{queues};

    foreach my $queue (@$queues) {

        # Nothing to do if already draining $queue
        next if $self->{Draining}->{$queue};

        # As no one is processing requests, respond these with errors
        $self->{Draining}->{$queue} = 1;

        my $local_bus = $self->{_BUS}->{bus_role};
        log_error "Draining unserviced req/$local_bus/$queue";

        $self->accept_remote_calls( "$queue.*" => 'reject_request' );
    }
}

sub on_worker_status {
    my ($self, $status) = @_;

    return unless ($status->{queue});

    return if ($status->{class} eq 'Beekeeper::Service::Sinkhole::Worker');

    foreach my $queue (@{$status->{queue}}) {

        # Nothing to do if not draining queue
        next unless $self->{Draining}->{$queue};

        # A worker servicing a previously unserviced queue has just become
        # online, so do not respond with errors anymore
        delete $self->{Draining}->{$queue};

        my $local_bus = $self->{_BUS}->{bus_role};
        log_warn "Stopped draining req/$local_bus/$queue";

        $self->stop_accepting_calls( "$queue.*" );
    }
}

sub reject_request {
    my ($self, $params, $req) = @_;

    # Just return a JSONRPC error response

    if ($req->mqtt_properties->{'auth'}) {
        # When client provided some kind of authentication tell him the truth
        # about the service being down. Otherwise the one trying to fix the
        # issue may be deceived into looking for auth/permissions problems
        return Beekeeper::JSONRPC::Error->method_not_available;
    }
    else {
        return Beekeeper::JSONRPC::Error->request_not_authorized;
    }
}

1;

__END__