view release on metacpan or  search on metacpan

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

# ABSTRACT: Priority message queues for Coro


use Coro qw(:prio);
use Coro::Semaphore ();

use List::Util qw(first sum);
use AnyEvent ();

sub SGET()      { 0 }
sub SPUT()      { 1 }
sub REPRIO()    { 2 }
sub NEXTCHECK() { 3 }
sub DATA()      { 4 }
sub MAX()       { PRIO_MAX - PRIO_MIN + DATA + 1 }


sub new {
   # we cheat, just like Coro::Channel.
   bless [
      (Coro::Semaphore::_alloc 0), # counts data
      (Coro::Semaphore::_alloc +($_[1] || 2_000_000_000) - 1), # counts remaining space
      $_[2], # reprioritization check time
      (defined $_[2] ? (AnyEvent->now + $_[2]) : undef), # last reprioritization check
      [], # initially empty

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

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;

    $self;
}