File Coverage

blib/lib/Async/Event/Interval.pm

Criterion	Covered	Total	%
statement	136	169	80.4
branch	41	62	66.1
condition	13	15	86.6
subroutine	36	37	97.3
pod	15	15	100.0
total	241	298	80.8

package Async::Event::Interval;

use warnings;
use strict;

our $VERSION = '1.10';

use Carp qw(croak);
use IPC::Shareable;
use Parallel::ForkManager;

$SIG{CHLD} = 'IGNORE';
$SIG{__WARN__} = sub {
    my $warn = shift;
    warn $warn if $warn !~ /^child process/;
};

my $id = 0;

my %events;
my $shared_memory_protect_lock = _rand_shm_lock();

my $shared_memory_segment_created;

while (! $shared_memory_segment_created) {
    $shared_memory_segment_created = eval {
        tie %events, 'IPC::Shareable', {
            key         => _rand_shm_key(),
            create      => 1,
            exclusive   => 1,
            protected   => _shm_lock(),
            mode        => 0600,
            destroy     => 1
        };
        1;
    };
}

*restart = \&start;

sub new {
    my $self = bless {}, shift;

    $self->id($id);
    $id++;

    $events{$self->id} = {};

    $self->_pm;
    $self->_setup(@_);
    $self->_started(0);

    return $self;
}
sub error {
    my ($self) = @_;
    $self->status;
    return $self->pid && $self->pid == -99 ? 1 : 0;
}
sub errors {
    my ($self) = @_;
    return $self->_errors || 0;
}
sub error_message {
    my ($self) = @_;
    return $self->_error_message;
}
sub events {
    return \%events;
}
sub id {
    my ($self, $id) = @_;
    $self->{id} = $id if defined $id;
    return $_[0]->{id};
}
sub info {
    my ($self) = @_;
    return $self->events()->{$self->id};
}
sub interval {
    my ($self, $interval) = @_;

    if (defined $interval) {
        if ($interval !~ /^\d+$/ && $interval !~ /^(\d+)?\.\d+$/) {
            croak "\$interval must be an integer or float";
        }
        $events{$self->id}{interval} = $interval;
    }

    return $events{$self->id}->{interval};
}
sub pid {
    my ($self) = @_;
    return $self->_pid;
}
sub runs {
    my ($self) = @_;
    return $self->_runs || 0;
}
sub shared_scalar {
    my ($self) = @_;

    my $shm_key;
    my $unique_shm_key_found = 0;

    for (0..9) {
        $shm_key = _rand_shm_key();
        if (! exists $events{$self->id}->{shared_scalars}{$shm_key}) {
            $unique_shm_key_found = 1;
            last;
        }
    }

    if (! $unique_shm_key_found) {
        croak("Could not generate a unique shared memory segment.");
    }

    tie my $scalar, 'IPC::Shareable', $shm_key, {create => 1, destroy => 1};

    $events{$self->id}->{shared_scalars}{$shm_key} = \$scalar;

    return \$scalar;
}
sub start {
    my ($self) = @_;
    if ($self->_started){
        warn "Event already running...\n";
        return;
    }
    $self->_started(1);
    $self->_event;
}
sub status {
    my ($self) = @_;

    if ($self->_started){
        if (! $self->pid){
            croak "Event is started, but no PID can be found. This is a " .
                "fatal error. Exiting...\n";
        }
        if ($self->pid > 0){
            if (kill 0, $self->pid){
                return $self->pid;
            }
            else {
                # proc must have crashed
                $self->_started(0);
                $self->_pid(-99);
                $self->error;
            }
        }
    }
    return 0;
}
sub stop {
    my $self = shift;

    if ($self->pid){
        kill 9, $self->pid;

        $self->_started(0);

        # time to ensure the proc was killed

        sleep 1;

        if (kill 0, $self->pid){
            croak "Event stop was called, but the process hasn't been killed. " .
                  "This is a fatal event. Exiting...\n";
        }
    }
}
sub waiting {
    my ($self) = @_;
    return 1 if $self->error || ! $self->status;
    return 0;
}

sub _args {
    my ($self, $args) = @_;

    if (defined $args) {
        $self->{args} = $args;
    }

    return $self->{args};
}
sub _cb {
    my ($self, $cb) = @_;

    if (defined $cb) {
        croak "Callback must be a code reference." if ref $cb ne 'CODE';
        $self->{cb} = $cb;
    }

    return $self->{cb};
}
sub _errors {
    my ($self, $increment) = @_;
    $events{$self->id}->{errors}++ if defined $increment;
    return $events{$self->id}->{errors};
}
sub _error_message {
    my ($self, $msg) = @_;
    $events{$self->id}->{error_message} = $msg if defined $msg;
    return $events{$self->id}->{error_message};
}
sub _event {
    my ($self) = @_;

    for (0..1){
        my $pid = $self->_pm->start;
        if ($pid){
            # this is the parent process
            $self->_pid($pid);
            last;
        }

        # set the child's proc id

        $self->_pid($$);

        # if no interval, run only once

        if ($self->interval) {
            while (1) {
                select(undef, undef, undef, $self->interval);

                my $callback_success = eval {
                    $self->_cb->(@{ $self->_args });
                    1;
                };

                if (! $callback_success) {
                    $self->_errors(1);
                    $self->_error_message($@);
                    $self->_runs(1);
                    $self->status;
                    croak $@;
                }

                $self->_runs(1);
                $self->status;
            }
        }
        else {

            my $callback_success = eval {
                $self->_cb->(@{$self->_args});
                1;
            };

            if (! $callback_success) {
                $self->_errors(1);
                $self->_error_message($@);
                $self->_runs(1);
                $self->status;
                croak $@;
            }

            $self->_runs(1);
            $self->status;
        }

        $self->_pm->finish;
    }
}
sub _pm {
    my ($self) = @_;

    if (! exists $self->{pm}) {
        $self->{pm} = Parallel::ForkManager->new(1);
    }

    return $self->{pm};
}
sub _pid {
    my ($self, $pid) = @_;
    $self->{pid} = $pid if defined $pid;
    $events{$self->id}->{pid} = $self->{pid} if $self->{pid};
    return $self->{pid} || undef;
}
sub _rand_shm_key {
    my $key_str;

    for (0..11) {
        srand();
        $key_str .= ('A'..'Z')[rand(26)];
    }

   return $key_str;
}
sub _rand_shm_lock {
    # Used for the 'protected' option in the %events hash creation

    srand();
    return int(rand(1_000_000));
}
sub _runs {
    my ($self, $increment) = @_;
    $events{$self->id}->{runs}++ if defined $increment;
    return $events{$self->id}->{runs};
}
sub _setup {
    my ($self, $interval, $cb, @args) = @_;
    $self->interval($interval);
    $self->_cb($cb);
    $self->_args(\@args);
}
sub _shm_lock {
    return $shared_memory_protect_lock;
}
sub _started {
    my ($self, $started) = @_;
    $self->{started} = $started if defined $started;
    return $self->{started};
}
sub DESTROY {
    $_[0]->stop if $_[0]->pid;

    # On events with interval of zero, ForkManager runs finish(), which
    # calls our destroy method. We only want to blow away the %events
    # hash if we truly go out of scope

    return if (caller())[0] eq 'Parallel::ForkManager::Child';

    delete $events{$_[0]->id};
}
sub _end {
    if (keys %events) {
        warn "The following events remain: " . join(', ', keys %events);
    }

    IPC::Shareable::clean_up_protected(_shm_lock());
}
END {
    _end();
}
sub _vim{}

1;

__END__