File Coverage

blib/lib/Data/Queue/Batch.pm

Criterion	Covered	Total	%
statement	50	52	96.1
branch	12	12	100.0
condition	6	8	75.0
subroutine	14	16	87.5
pod	11	11	100.0
total	93	99	93.9

package Data::Queue::Batch;
use 5.008001;
use strict;
use warnings;

our $VERSION = "0.03";

sub new {
    my ($class, %args) = @_;
    my $batch_size = delete($args{batch_size}) || 100;
    my $callback = delete($args{callback});
    return bless {
        callback => $callback,
        batch_size => $batch_size,,
        %args,
        available => 0,
        _queue => [],
    }, $class;
}

sub size       { scalar(@{ shift->{_queue} }) }
sub available  { shift->{available} }
sub batch_size { shift->{batch_size} }

sub push { shift->enqueue(@_) }
sub enqueue {
    my ($self, @values) = @_;
    CORE::push(@{ $self->{_queue} }, @values);

    my $unmarked = $self->size - $self->{available};
    my $marking = $unmarked - ($unmarked % $self->{batch_size});
    $self->{available} += $marking;

    if ($self->{callback} && $self->{available}) {
        $self->{callback}->($self->_take($self->{available}));
    }
    return;
}

sub shift { shift->dequeue(@_) }
sub dequeue {
    my ($self) = @_;
    return unless $self->{available};
    my ($dequeued) =  $self->_take(1);
    return $dequeued;
}

sub peek {
    my ($self, $count) = @_;
    $count = $self->{available} if $count > $self->{available};
    my @peeked =  @{$self->{_queue}}[0 .. $count - 1];
    return @peeked;
}

sub flush {
    my ($self) = @_;
    my @taken = $self->_take($self->size);
    if ($self->{callback} && @taken) {
        $self->{callback}->(@taken);
    }
    return @taken;
}

sub clear {
    my ($self) = @_;
    $self->_take($self->size);
    return;
}

sub _take {
    my ($self, $count) = @_;
    my @taken = splice(@{ $self->{_queue} }, 0, $count);
    $self->{available} -= $count;
    $self->{available} = 0 if $self->{available} < 0;
    return @taken;
}

sub DESTROY {
    my ($self) = @_;
    $self->flush if $self->{callback};
}

1;
__END__