File Coverage

blib/lib/Async/Redis/AutoPipeline.pm

Criterion	Covered	Total	%
statement	17	65	26.1
branch	0	12	0.0
condition	0	5	0.0
subroutine	6	14	42.8
pod	0	2	0.0
total	23	98	23.4

package Async::Redis::AutoPipeline;

use strict;
use warnings;
use 5.018;

use Future;
use Future::IO;
use Scalar::Util qw(blessed);

sub new {
    my ($class, %args) = @_;

    return bless {
        redis         => $args{redis},
        max_depth     => $args{max_depth} // 1000,
        _queue        => [],
        _flush_pending => 0,
        _flushing     => 0,
    }, $class;
}

sub command {
    my ($self, @args) = @_;

    my $future = Future->new;
    push @{$self->{_queue}}, { cmd => \@args, future => $future };

    # Schedule flush exactly once per batch
    unless ($self->{_flush_pending}) {
        $self->{_flush_pending} = 1;
        $self->_schedule_flush;
    }

    return $future;
}

sub _schedule_flush {
    my ($self) = @_;

    # Use event loop's "next tick" mechanism
    # sleep(0) yields to event loop then immediately returns
    Future::IO->sleep(0)->on_done(sub {
        $self->_do_flush;
    });
}

sub _do_flush {
    my ($self) = @_;

    # Reentrancy guard
    return if $self->{_flushing};
    $self->{_flushing} = 1;

    # Reset pending flag before flush (allows new commands to queue)
    $self->{_flush_pending} = 0;

    # Take current queue atomically
    my @batch = splice @{$self->{_queue}};

    if (@batch) {
        # Respect depth limit
        my $max = $self->{max_depth};
        if (@batch > $max) {
            # Put excess back, schedule another flush
            unshift @{$self->{_queue}}, splice(@batch, $max);
            $self->{_flush_pending} = 1;
            $self->_schedule_flush;
        }

        $self->_send_batch(\@batch);
    }

    $self->{_flushing} = 0;
}

sub _send_batch {
    my ($self, $batch) = @_;

    my $redis = $self->{redis};

    # Build pipeline commands
    my @commands = map { $_->{cmd} } @$batch;
    my @futures  = map { $_->{future} } @$batch;

    # Execute pipeline and distribute results
    # Keep reference to the pipeline future to prevent it from being lost
    my $pipeline_f = $redis->_execute_pipeline(\@commands);

    $pipeline_f->on_done(sub {
        my ($results) = @_;

        for my $i (0 .. $#$results) {
            my $result = $results->[$i];
            my $future = $futures[$i];

            # Check if result is an error object (from pipeline inline capture)
            if (blessed($result) && $result->isa('Async::Redis::Error')) {
                $future->fail($result);
            }
            else {
                $future->done($result);
            }
        }
    });

    $pipeline_f->on_fail(sub {
        my ($error) = @_;

        # Transport failure - fail all futures
        for my $future (@futures) {
            $future->fail($error) unless $future->is_ready;
        }
    });

    # Retain the future to keep the async operation alive
    $pipeline_f->retain;
}

1;

__END__