File Coverage

blib/lib/Promises/Channel.pm

Criterion	Covered	Total	%
statement	70	72	97.2
branch	2	2	100.0
condition	8	12	66.6
subroutine	16	17	94.1
pod	8	11	72.7
total	104	114	91.2

package Promises::Channel;
# ABSTRACT: a coordination channel implemented with Promises
$Promises::Channel::VERSION = '0.01';

use strict;
use warnings;

use Moo;
use Promises qw(deferred);

extends 'Exporter';

our @EXPORT_OK = qw(
  channel
  chan
);



has limit =>
  is        => 'ro',
  predicate => 'has_limit';


has is_shutdown =>
  is      => 'ro',
  default => 0;

has backlog =>
  is      => 'ro',
  default => sub { [] };

has inbox =>
  is      => 'ro',
  default => sub { [] };

has outbox =>
  is      => 'ro',
  default => sub { [] };



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


sub is_full {
  my $self = shift;
  return $self->has_limit
      && $self->size == $self->limit;
}


sub is_empty {
  my $self = shift;
  return $self->size == 0
      && !@{ $self->backlog };
}


sub put {
  my ($self, $item) = @_;
  my $soon = deferred;

  my $promise = $soon->promise->then(sub {
    my ($self, $item) = @_;
    $self->drain;
    return $self;
  });

  push @{ $self->backlog }, [$item, $soon];
  $self->pump;
  $soon->promise;
}


sub get {
  my $self = shift;
  my $soon = deferred;
  push @{ $self->outbox }, $soon;

  my $promise = $soon->promise->then(sub {
    my ($self, $item) = @_;
    $self->pump;
    return ($self, $item);
  });

  $self->drain;

  return $promise;
}


sub shutdown {
  my $self = shift;
  $self->{is_shutdown} = 1;
  $self->drain;
  $self->pump;
}

sub pump {
  my $self = shift;

  while (@{ $self->backlog } && !$self->is_full) {
    my ($item, $soon) = @{ shift @{ $self->backlog } };
    push @{ $self->inbox }, $item;
    $soon->resolve($self->size);
  }
}

sub drain {
  my $self = shift;

  while (@{ $self->inbox } && @{ $self->outbox }) {
    my $soon = shift @{ $self->outbox };
    my $msg  = shift @{ $self->inbox };
    $soon->resolve($self, $msg);
  }

  if ($self->is_shutdown) {
    while (@{ $self->outbox }) {
      my $soon = shift @{ $self->outbox };
      $soon->resolve($self, undef);
    }
  }

  return;
}

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



sub channel { Promises::Channel->new(@_) }
sub chan    { Promises::Channel->new(@_) }

1;

__END__