File Coverage

blib/lib/AnyEvent/FIFO.pm

Criterion	Covered	Total	%
statement	53	53	100.0
branch	11	14	78.5
condition	5	10	50.0
subroutine	11	11	100.0
pod	6	6	100.0
total	86	94	91.4

package AnyEvent::FIFO;
use strict;
use AnyEvent;
use AnyEvent::Util ();

our $VERSION = '0.00003';

sub new {
    my $class = shift;
    my $self = {
        max_active => 1,
        @_,
        active => {},
        events => {},
    };
    $self->{cv} ||= AE::cv;
    return bless($self, $class);
}

sub push {
    my ($self, $slot, $cb, @args) = @_;
    # the first argument must be the name of the slot or a callback
    if (ref $slot) {
        unshift @args, $cb;
        $cb = $slot;
        $slot = "__default__";
    }

    push @{$self->{events}->{$slot}}, [$cb, @args];
    $self->{cv}->begin();

    AE::postpone sub {
        $self->drain();
    };
}

sub active {
    my ($self, $slot) = @_;
    $slot = "__default__" unless(defined($slot));
    return $self->{active}->{$slot} || 0;
}

sub waiting {
    my ($self, $slot) = @_;
    $slot = "__default__" unless(defined($slot));
    return $self->{events}->{$slot} ? (0 + @{$self->{events}->{$slot}}) : 0;
}

sub cv {
    my $self = shift;
    $self->{cv} = $_[0] if(@_);
    return $self->{cv};
}

sub drain {
    my $self = shift;

    my @slots = keys %{$self->{events}};
    my $dispatched = 1;
    while ($dispatched) {
        $dispatched = 0;
        foreach my $slot (@slots) {
            my $events = $self->{events}->{$slot};
            if ( @$events && ($self->{active}->{$slot} ||= 0) < $self->{max_active} ) {
                $dispatched++;
                my $stuff = shift @$events;
                my ($cb, @args) = @$stuff;
                $self->{active}->{$slot}++;
                $cb->( AnyEvent::Util::guard {
                    $self->{active}->{$slot}--;
                    if ($self->{active}->{$slot} <= 0) {
                        delete $self->{active}->{$slot};
                    }
		    $self->{cv}->end();
                    AE::postpone sub {
                        $self->drain();
                    };
                }, @args );
            }
        }
    }
}

1;

__END__