File Coverage

blib/lib/AnyEvent/RabbitMQ/LocalQueue.pm

Criterion	Covered	Total	%
statement	28	35	80.0
branch	2	2	100.0
condition			n/a
subroutine	6	7	85.7
pod	0	3	0.0
total	36	47	76.6

package AnyEvent::RabbitMQ::LocalQueue;

use strict;
use warnings;

sub new {
    my $class = shift;
    return bless {
        _message_queue    => [],
        _drain_code_queue => [],
    }, $class;
}

sub push {
    my $self = shift;

    CORE::push @{$self->{_message_queue}}, @_;
    return $self->_drain_queue();
}

sub get {
    my $self = shift;

    CORE::push @{$self->{_drain_code_queue}}, @_;
    return $self->_drain_queue();
}

sub _drain_queue {
    my $self = shift;

    my $message_count = scalar @{$self->{_message_queue}};
    my $drain_code_count = scalar @{$self->{_drain_code_queue}};

    my $count = $message_count < $drain_code_count
              ? $message_count : $drain_code_count;

    for (1 .. $count) {
        &{shift @{$self->{_drain_code_queue}}}(
            shift @{$self->{_message_queue}}
        );
    }

    return $self;
}

sub _flush {
    my ($self, $frame) = @_;

    $self->_drain_queue;

    while (my $cb = shift @{$self->{_drain_code_queue}}) {
        local $@; # Flush frames immediately, throwing away errors for on-close
        eval { $cb->($frame) };
    }
}

1;