File Coverage

blib/lib/Async/Redis/Pipeline.pm

Criterion	Covered	Total	%
statement	11	70	15.7
branch	0	26	0.0
condition	0	5	0.0
subroutine	4	12	33.3
pod	5	6	83.3
total	20	119	16.8

package Async::Redis::Pipeline;

use strict;
use warnings;
use 5.018;

use Future::AsyncAwait;

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

    return bless {
        redis     => $args{redis},
        commands  => [],
        executed  => 0,
        max_depth => $args{max_depth} // 10000,
    }, $class;
}

# Queue a command - returns self for chaining
sub _queue {
    my ($self, $cmd, @args) = @_;

    die "Pipeline already executed" if $self->{executed};

    if (@{$self->{commands}} >= $self->{max_depth}) {
        die "Pipeline depth limit ($self->{max_depth}) exceeded";
    }

    push @{$self->{commands}}, [$cmd, @args];
    return $self;
}

# Generate AUTOLOAD to capture any command call
our $AUTOLOAD;

sub AUTOLOAD {
    my $self = shift;
    my $cmd = $AUTOLOAD;
    $cmd =~ s/.*:://;
    return if $cmd eq 'DESTROY';

    return $self->_queue(uc($cmd), @_);
}

# Allow explicit command() calls
sub command {
    my ($self, $cmd, @args) = @_;
    return $self->_queue($cmd, @args);
}

# Explicit add() for backwards compatibility
sub add {
    my ($self, @cmd) = @_;
    return $self->_queue(@cmd);
}

# Queue a registered script by name
# Usage: $pipe->run_script('name', @keys_then_args)
sub run_script {
    my ($self, $name, @args) = @_;

    die "Pipeline already executed" if $self->{executed};
    die "Script name required" unless defined $name;

    if (@{$self->{commands}} >= $self->{max_depth}) {
        die "Pipeline depth limit ($self->{max_depth}) exceeded";
    }

    # Queue with special marker - resolved at execute time
    push @{$self->{commands}}, ['__SCRIPT__', $name, @args];
    return $self;
}

async sub execute {
    my ($self) = @_;

    # Mark as executed (single-use)
    if ($self->{executed}) {
        return [];
    }
    $self->{executed} = 1;

    my @commands = @{$self->{commands}};
    return [] unless @commands;

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

    # Resolve __SCRIPT__ entries to EVALSHA commands
    my %scripts_used;
    for my $cmd (@commands) {
        if ($cmd->[0] eq '__SCRIPT__') {
            my (undef, $name, @args) = @$cmd;

            my $script = $redis->get_script($name)
                or die "Unknown script '$name' in pipeline - use define_command() first";

            $scripts_used{$name} = $script;

            my $num_keys = $script->num_keys;
            if ($num_keys eq 'dynamic') {
                $num_keys = shift @args;
                die "Key count required as first argument for dynamic script '$name'"
                    unless defined $num_keys;
            }

            # Convert to EVALSHA command
            @$cmd = ('EVALSHA', $script->sha, $num_keys, @args);
        }
    }

    # Preload any scripts used (ensures EVALSHA will work)
    for my $script (values %scripts_used) {
        await $redis->script_load($script->script);
    }

    # Apply key prefixing if configured
    if (defined $redis->{prefix} && $redis->{prefix} ne '') {
        require Async::Redis::KeyExtractor;
        for my $cmd (@commands) {
            my ($name, @args) = @$cmd;
            @args = Async::Redis::KeyExtractor::apply_prefix(
                $redis->{prefix}, $name, @args
            );
            @$cmd = ($name, @args);
        }
    }

    # Execute pipeline via Redis connection
    return await $redis->_execute_pipeline(\@commands);
}

sub count { scalar @{shift->{commands}} }

1;

__END__