File Coverage

blib/lib/IPC/Simple/Group.pm

Criterion	Covered	Total	%
statement	45	48	93.7
branch	7	12	58.3
condition			n/a
subroutine	14	15	93.3
pod	6	9	66.6
total	72	84	85.7

package IPC::Simple::Group;
# ABSTRACT: work with several processes as a group
$IPC::Simple::Group::VERSION = '0.08';

use strict;
use warnings;

use Carp;
use IPC::Simple::Channel qw();

sub new {
  my $class = shift;

  my $self = bless{
    members  => {},
    messages => IPC::Simple::Channel->new,
  }, $class;

  $self->add(@_);

  return $self;
}

sub add {
  my $self = shift;

  for (@_) {
    croak 'processes must be named to be grouped'
      unless $_->name;

    croak 'processes with a recv_cb may not be grouped'
      if $_->{cb};

    croak 'processes with a term_cb may not be grouped'
      if $_->{term_cb};
  }

  for (@_) {
    $self->{members}{ $_->{name} } = $_;
    $_->{recv_cb} = sub{ $self->{messages}->put( $_[0] ) };
    $_->{term_cb} = sub{ $self->drop( $_[0] ) };

    # If the process has already been launched, move existing messages into the
    # group queue.
    unless ($_->is_ready) {
      $self->{messages}->put( $_->{messages}->clear );
    }
  }
}

sub drop {
  my $self = shift;

  delete $self->{members}{ $_->{name} }
    for @_;

  unless (%{ $self->{members} }) {
    $self->{messages}->shutdown;
  }
}

sub members {
  my $self = shift;
  return values %{ $self->{members} };
}

sub launch {
  my $self = shift;

  for ($self->members) {
    $_->launch if $_->is_ready;
  }
}

sub terminate {
  my $self = shift;
  $_->terminate(@_) for $self->members;
}

sub signal {
  my ($self, $signal) = @_;
  $self->signal($signal) for $self->members;
}

sub join {
  my $self = shift;
  $_->join for $self->members;
}

sub recv {
  my $self = shift;
  $self->{messages}->recv;
}

1;

__END__