File Coverage

blib/lib/IPC/Simple/Channel.pm

Criterion	Covered	Total	%
statement	44	58	75.8
branch	2	2	100.0
condition	3	3	100.0
subroutine	10	14	71.4
pod	0	10	0.0
total	59	87	67.8

package IPC::Simple::Channel;
$IPC::Simple::Channel::VERSION = '0.09';
use strict;
use warnings;

use AnyEvent qw();

sub new {
  my ($class) = @_;

  bless{
    waiters     => [],
    buffer      => [],
    is_shutdown => 0,
  }, $class;
}

sub DESTROY {
  my $self = shift;
  $self->shutdown;
}

sub shutdown {
  my $self = shift;

  $self->{is_shutdown} = 1;

  # flush any remaining messages that have pending receivers
  $self->flush;

  # send undef to any remaining receivers
  $_->send for @{ $self->{waiters} };
}

sub size {
  my $self = shift;
  return scalar @{ $self->{buffer} };
}

sub put {
  my $self = shift;
  push @{ $self->{buffer} }, @_;
  $self->flush;
  return $self->{size};
}

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

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

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

sub async {
  my $self = shift;
  my $cv = AnyEvent->condvar;

  if ($self->{is_shutdown}) {
    my $msg = shift @{ $self->{buffer} };
    $cv->send($msg);
    return $cv;
  }
  else {
    push @{ $self->{waiters} }, $cv;
    $self->flush;
    return $cv;
  }
}

sub flush {
  my $self = shift;
  while (@{ $self->{waiters} } && @{ $self->{buffer} }) {
    my $cv = shift @{ $self->{waiters} };
    $cv->send( shift @{ $self->{buffer} } );
  }
}

sub clear {
  my $self = shift;
  $self->flush;
  my @msgs = @{ $self->{buffer} };
  $self->{buffer} = [];
  $self->{size} = 0;
  return @msgs;
}

1;

__END__