view release on metacpan or  search on metacpan

lib/Coro/PrioChannel/Multi.pm  view on Meta::CPAN

{
  $Coro::PrioChannel::Multi::VERSION = '0.005';
}
use strict;
use warnings;

# ABSTRACT: Multiple-listener priority message queues for Coro


use Coro::PrioChannel;
use Scalar::Util qw(weaken);


sub LIM()  { 0 }
sub CHAN() { 1 }

sub new
{
    my $class = shift;
    my $limit = shift;
    my $self  = bless [$limit, []], $class;

lib/Coro/PrioChannel/Multi.pm  view on Meta::CPAN

    $self;
}


sub clean
{
    my $self = shift;
    @{$self->[CHAN]} = grep { defined } @{$self->[CHAN]};

    # when we pull out the refs this way, they're
    # no longer weakened, so re-weaking everything.
    # (easier than using splice to pull undef items out -
    #  if we get too many readers, we'll re-evaluate if this
    #  is slow.)
    weaken($_) for @{$self->[CHAN]};
}


sub number_of_listeners
{
    my $self = shift;
    $self->clean();
    scalar @{$self->[CHAN]};
}

# debugging aid.
sub _status
{
    my $self = shift;
    $self->clean();
    'Channel=size :: ' . join ":", map { $_ . "=" . $self->[CHAN][$_]->size() } 0..$#$self;
}


# create new channel, add it to $self, ensure it's weakened, and return
# the non-weak version.
sub listen
{
    my $self = shift;
    my $channel = Coro::PrioChannel->new($self->[LIM]);
    push @{$self->[CHAN]}, $channel;

    $self->clean();

    $channel;