File Coverage

blib/lib/Daemon/Device.pm

Criterion	Covered	Total	%
statement	88	193	45.6
branch	21	76	27.6
condition	8	25	32.0
subroutine	15	30	50.0
pod	11	12	91.6
total	143	336	42.5

package Daemon::Device;
# ABSTRACT: Forking daemon device construct

use 5.012;
use strict;
use warnings;

use Daemon::Control;
use Carp qw( croak carp );
use POSIX ":sys_wait_h";
use IO::Pipe;

our $VERSION = '1.08'; # VERSION

sub new {
    my $class = shift;
    croak 'new() called with uneven number of parameters' if ( @_ % 2 );

    my $self = bless( {@_}, $class );

    $self->{ '_' . $_ } = delete $self->{$_} for ( qw(
        daemon
        spawn
        replace_children
        parent_hup_to_child
        parent
        child
        on_startup
        on_shutdown
        on_spawn
        on_parent_hup
        on_child_hup
        on_parent_death
        on_child_death
        on_replace_child
        on_message
        data
    ) );

    if ( not $self->{_daemon}{user} ) {
        my $user = getlogin || getpwuid($<) || 'root';
        $self->{_daemon}{user} ||= $user;
    }
    $self->{_daemon}{group} ||= ( getgrgid( (getpwnam( $self->{_daemon}{user} ) )[3] ) )[0];

    croak 'new() called without "daemon" parameter as a hashref' unless ( ref( $self->{_daemon} ) eq 'HASH' );
    for ( qw( program program_args ) ) {
        croak qq{new() called with "daemon" hashref containing "$_" key} if ( $self->{_daemon}{$_} );
    }
    for ( qw(
        parent child
        on_startup on_shutdown on_spawn on_parent_hup on_child_hup
        on_parent_death on_child_death on_replace_child
    ) ) {
        croak qq{new() called with "$_" parameter not a coderef}
            if ( exists $self->{$_} and ref( $self->{$_} ) ne 'CODE' );
    }

    $self->{_daemon}{program}      = \&_parent;
    $self->{_daemon}{program_args} = [$self];

    $self->{_spawn}               ||= 1;
    $self->{_replace_children}    //= 1;
    $self->{_parent_hup_to_child} //= 1;
    $self->{_data}                //= {};

    $self->{_children} = [];
    $self->{_daemon}   = Daemon::Control->new( %{ $self->{_daemon} } );

    return $self;
}

sub run {
    my ($self) = @_;
    return $self->{_daemon}->run;
}

sub daemon {
    my ($self) = @_;
    return $self->{_daemon};
}

sub _parent {
    my ( $daemon, $self ) = @_;

    $self->{_ppid} = $$;

    $SIG{'HUP'} = sub {
        $self->{_on_parent_hup}->($self) if ( $self->{_on_parent_hup} );
        if ( $self->{_parent_hup_to_child} ) {
            kill( 'HUP', $_->{pid} ) for ( @{ $self->{_children} } );
        }
    };

    my $terminate = sub {
        $self->{_on_parent_death}->($self) if ( $self->{_on_parent_death} );
        kill( 'TERM', $_->{pid} ) for ( @{ $self->{_children} } );
        $self->{_on_shutdown}->($self) if ( $self->{_on_shutdown} );
        exit;
    };
    $SIG{$_} = $terminate for ( qw( TERM INT ABRT QUIT ) );

    $SIG{'CHLD'} = sub {
        if ( $self->{_replace_children} ) {
            $self->{_on_replace_child}->($self) if ( $self->{_on_replace_child} );
            for ( @{ $self->{_children} } ) {
                $_ = _spawn($self) if ( waitpid( $_->{pid}, WNOHANG ) );
            }
        }
    };

    $SIG{'URG'} = sub {
        if ( $self->{_on_message} ) {
            my @messages = map { split(/\r?\n/) } map { $_->{io_up}->getlines } @{ $self->{_children} };
            $self->{_on_message}->( $self, @messages );
        }
    };

    $self->{_on_startup}->($self) if ( $self->{_on_startup} );

    for ( 1 .. $self->{_spawn} ) {
        push( @{ $self->{_children} }, _spawn($self) );
    }

    if ( $self->{_parent} ) {
        $self->{_parent}->($self);
    }
    else {
        wait;
    }

    return;
}

sub _spawn {
    my ($self) = @_;

    $self->{_on_spawn}->($self) if ( $self->{_on_spawn} );

    my ( $io_up, $io_dn );
    if ( $self->{_on_message} ) {
        $io_up = IO::Pipe->new;
        $io_dn = IO::Pipe->new;
    }

    if ( my $pid = fork ) {
        my $child_data = { pid => $pid };

        if ( $self->{_on_message} ) {
            $io_up->reader;
            $io_dn->writer;
            for ( $io_up, $io_dn ) {
                $_->autoflush;
                $_->blocking(0);
            }
            $child_data->{io_up} = $io_up;
            $child_data->{io_dn} = $io_dn;
        }

        return $child_data;
    }
    else {
        if ( $self->{_on_message} ) {
            $io_up->writer;
            $io_dn->reader;
            for ( $io_up, $io_dn ) {
                $_->autoflush;
                $_->blocking(0);
            }
            $self->{_io_up} = $io_up;
            $self->{_io_dn} = $io_dn;
        }

        $self->{_cpid} = $$;
        _child($self);
        exit;
    }

    return;
}

sub _child {
    my ($self) = @_;

    $SIG{'HUP'} = sub {
        $self->{_on_child_hup}->($self) if ( $self->{_on_child_hup} );
    };

    my $terminate = sub {
        $self->{_on_child_death}->($self) if ( $self->{_on_child_death} );
        exit;
    };
    $SIG{$_} = $terminate for ( qw( TERM INT ABRT QUIT ) );

    $SIG{'URG'} = sub {
        if ( $self->{_on_message} ) {
            my @messages = map { split(/\r?\n/) } $self->{_io_dn}->getlines;
            $self->{_on_message}->( $self, @messages );
        }
    };

    if ( $self->{_child} ) {
        $self->{_child}->($self);
    }
    else {
        while (1) {
            exit unless ( $self->parent_alive );
            sleep 1;
        }
    }

    return;
}

sub ppid {
    return shift->{_ppid};
}

sub cpid {
    return shift->{_cpid};
}

sub children {
    return [ map { $_->{pid} } @{ shift->{_children} } ];
}

sub adjust_spawn {
    my ( $self, $new_spawn_count ) = @_;
    $self->{_spawn} = $new_spawn_count;

    if ( @{ $self->{_children} } > 0 and @{ $self->{_children} } < $self->{_spawn} ) {
        push( @{ $self->{_children} }, _spawn($self) ) while ( @{ $self->{_children} } < $self->{_spawn} );
    }
    elsif ( @{ $self->{_children} } > 0 and @{ $self->{_children} } > $self->{_spawn} ) {
        my $set_replace_children = $self->{_replace_children};
        $self->{_replace_children} = 0;

        my @killed_pids;
        while ( @{ $self->{_children} } > $self->{_spawn} ) {
            my $child = shift @{ $self->{_children} };
            kill( 'TERM', $child->{pid} );
            push( @killed_pids, $child->{pid} );
        }

        waitpid( $_, 0 ) for (@killed_pids);
        $self->{_replace_children} = $set_replace_children;
    }

    return;
}

sub replace_children {
    my $self = shift;
    $self->{_replace_children} = $_[0] if (@_);
    return $self->{_replace_children};
}

sub parent_hup_to_child {
    my $self = shift;
    $self->{_parent_hup_to_child} = $_[0] if (@_);
    return $self->{_parent_hup_to_child};
}

sub parent_alive {
    my ($self) = @_;
    return kill( 0, $self->{_ppid} );
}

sub data {
    my $self = shift;
    return $self->{'_data'} unless (@_);

    if ( @_ == 1 ) {
        return $self->{'_data'}{ $_[0] } if ( not ref $_[0] );
        $self->{'_data'}{$_} = $_[0]->{$_} for ( keys %{ $_[0] } );
        return $self;
    }

    if ( @_ % 2 != 0 ) {
        carp( 'data() called with uneven number of parameters' );
    }
    else {
        my %params = @_;
        $self->{'_data'}{$_} = $params{$_} for ( keys %params );
    }

    return $self;
}

sub message {
    my ( $self, $pid, $message ) = @_;

    unless ( $self->{_on_message} ) {
        carp('message() called without an on_message handler set');
        return;
    }

    my $io;
    eval {
        $io = ( $self->{_ppid} == $pid )
            ? $self->{_io_up}
            : ( grep { $_->{pid} == $pid } @{ $self->{_children} } )[0]->{io_dn}
    };
    if ($@) {
        carp("Failed to find IO path for message to $pid");
        return;
    }

    $io->say($message) or carp("Failed to send message to $pid");
    kill( 'URG', $pid ) or carp("Failed to signal process $pid to new message");
}

our $AUTOLOAD;
sub AUTOLOAD {
    my $self = shift;
    my $key = ( split( '::', $AUTOLOAD ) )[-1];
    return $self->data($key);
}

1;

__END__