File Coverage

blib/lib/Mojo/Promise/Limitter.pm

Criterion	Covered	Total	%
statement	37	37	100.0
branch	5	6	83.3
condition	2	2	100.0
subroutine	11	11	100.0
pod	2	2	100.0
total	57	58	98.2

package Mojo::Promise::Limitter;
use Mojo::Base 'Mojo::EventEmitter';
use Mojo::Promise;

our $VERSION = '0.001';

has outstanding => 0;
has concurrency => 0;
has jobs => sub { [] };

sub new {
    my ($class, $concurrency) = @_;
    $class->SUPER::new(concurrency => $concurrency);
}

sub limit {
    my ($self, $sub, $name) = @_;
    $name //= 'anon';
    if ($self->outstanding < $self->concurrency) {
        return $self->_run($sub, $name);
    } else {
        return $self->_queue($sub, $name);
    }
}

sub _run {
    my ($self, $sub, $name) = @_;
    $self->{outstanding}++;
    $self->emit(run => $name);
    $sub->()->then(
        sub { $self->_remove($name); @_ },
        sub { $self->_remove($name); Mojo::Promise->reject(@_) },
    );
}

sub _queue {
    my ($self, $sub, $name) = @_;
    $self->emit(queue => $name);
    Mojo::Promise->new(sub {
        my ($resolve, $reject) = @_;
        push @{$self->jobs}, { sub => $sub, name => $name, resolve => $resolve, reject => $reject };
    });
}

sub _dequeue {
    my $self = shift;
    if (my $job = shift @{$self->jobs}) {
        $self->emit(dequeue => $job->{name});
        $self->_run($job->{sub}, $job->{name})->then($job->{resolve}, $job->{reject});
    }
}

sub _remove {
    my ($self, $name) = @_;
    $self->{outstanding}--;
    $self->emit(remove => $name);
    if ($self->outstanding < $self->concurrency) {
        $self->_dequeue;
    }
}

1;
__END__