view release on metacpan or  search on metacpan

lib/AnyEvent/Beanstalk/Worker.pm  view on Meta::CPAN

package AnyEvent::Beanstalk::Worker;

use 5.016001;
use strict;
use warnings;
use feature 'current_sub';
use AnyEvent;
use AnyEvent::Log;
use AnyEvent::Beanstalk;

our $VERSION = '0.05';

sub new {
    my $class = shift;
    my $self  = {};
    bless $self => $class;

    my %args = @_;

    $self->{_cb}     = {};
    $self->{_event}  = {};
    $self->{_jobs}   = {};
    $self->{_events} = [];    ## event queue
    $self->{_handled_jobs} = 0;  ## simple job counter

    $self->{_running}        = 0;
    $self->{_stop_tries}     = 0;
    $self->{_max_stop_tries} = $args{max_stop_tries} // 3;
    $self->{_max_jobs}       = $args{max_jobs} || 0;
    $self->{_concurrency}    = $args{concurrency} || 1;
    $self->{_log_level}      = $args{log_level} // 4;

    $self->{_reserve_timeout}        = $args{reserve_timeout} || 1;
    $self->{_reserve_base}           = $self->{_reserve_timeout};
    $self->{_reserve_timeout_factor} = 1.1;
    $self->{_reserve_timeout_max}    = 4;
    $self->{_release_delay}          = $args{release_delay} || 3;

    $self->{_initial_state}          = $args{initial_state};

    $self->{_log_ctx} = AnyEvent::Log::ctx;
    $self->{_log_ctx}->title(__PACKAGE__);
    $self->{_log_ctx}->level($self->{_log_level});

    $self->{_log}          = {};
    $self->{_log}->{trace} = $self->{_log_ctx}->logger("trace");
    $self->{_log}->{debug} = $self->{_log_ctx}->logger("debug");
    $self->{_log}->{info}  = $self->{_log_ctx}->logger("info");
    $self->{_log}->{note}  = $self->{_log_ctx}->logger("note");

    $self->{_signal} = {};
    $self->{_signal}->{TERM} = AnyEvent->signal(
        signal => "TERM",
        cb =>
          sub { $self->{_log_ctx}->log( warn => "TERM received" ); $self->stop }
    );
    $self->{_signal}->{INT} = AnyEvent->signal(
        signal => "INT",
        cb =>
          sub { $self->{_log_ctx}->log( warn => "INT received" ); $self->stop }
    );
    $self->{_signal}->{USR2} = AnyEvent->signal(
        signal => "USR2",
        cb     => sub {
            $self->{_log_level} =
              ( $self->{_log_level} >= 9 ? 2 : $self->{_log_level} + 1 );
            $self->{_log_ctx}->level($self->{_log_level});
        }
    );

    $args{beanstalk_host} ||= 'localhost';
    $args{beanstalk_port} ||= 11300;

    unless ($args{beanstalk_watch}) {
        die "beanstalk_watch argument required\n";
    }