view release on metacpan or  search on metacpan

lib/Async/Redis.pm  view on Meta::CPAN

# Key extraction for prefixing
use Async::Redis::KeyExtractor;

# Transaction support
use Async::Redis::Transaction;

# Script support
use Async::Redis::Script;

# Iterator support
use Async::Redis::Iterator;

# Pipeline support
use Async::Redis::Pipeline;
use Async::Redis::AutoPipeline;

# PubSub support
use Async::Redis::Subscription;

# Telemetry support
use Async::Redis::Telemetry;

# Try XS version first, fall back to pure Perl
BEGIN {
    eval { require Protocol::Redis::XS; 1 }
        or require Protocol::Redis;
}

sub _parser_class {
    return $INC{'Protocol/Redis/XS.pm'} ? 'Protocol::Redis::XS' : 'Protocol::Redis';
}

sub _calculate_backoff {
    my ($self, $attempt) = @_;

    # Exponential: delay * 2^(attempt-1)
    my $delay = $self->{reconnect_delay} * (2 ** ($attempt - 1));

    # Cap at max
    $delay = $self->{reconnect_delay_max} if $delay > $self->{reconnect_delay_max};

    # Apply jitter: delay * (1 +/- jitter)
    if ($self->{reconnect_jitter} > 0) {
        my $jitter_range = $delay * $self->{reconnect_jitter};
        my $jitter = (rand(2) - 1) * $jitter_range;
        $delay += $jitter;
    }

    return $delay;
}

# Free function, not a method. Call as _await_with_deadline($f, $deadline).
# Race a read future against a deadline. Returns a Future resolving to
# ($read_future, $timed_out_bool). The caller inspects $timed_out and
# $read_future->is_failed explicitly; we never throw from here.
#
# On timeout win: $read_future is left pending. _reader_fatal is the sole
# owner of its cancellation (it must happen before _close_socket so
# Future::IO unregisters while fileno is still valid).
# On read win: the internal timeout timer is cancelled here for hygiene.
sub _await_with_deadline {
    my ($read_f, $deadline) = @_;

    if (!defined $deadline) {
        return $read_f->followed_by(sub { Future->done($read_f, 0) });
    }

    my $remaining = $deadline - Time::HiRes::time();
    if ($remaining <= 0) {
        return Future->done($read_f, 1);
    }

    my $timeout_f = Future::IO->sleep($remaining)
        ->then(sub { Future->fail('__deadline__') });

    # Use without_cancel so that if timeout wins, wait_any's cancel of the
    # losing future does not propagate to $read_f (caller owns its lifecycle).
    return Future->wait_any($read_f->without_cancel, $timeout_f)
        ->followed_by(sub {
            my ($f) = @_;
            my $timed_out = $f->is_failed
                && (($f->failure)[0] // '') eq '__deadline__' ? 1 : 0;

            if (!$timed_out && !$timeout_f->is_ready) {
                $timeout_f->cancel;
            }

            return Future->done($read_f, $timed_out);
        });
}

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

    # Parse URI if provided
    if ($args{uri}) {
        require Async::Redis::URI;
        my $uri = Async::Redis::URI->parse($args{uri});
        if ($uri) {
            my %uri_args = $uri->to_hash;
            # URI values are defaults, explicit args override
            %args = (%uri_args, %args);
            delete $args{uri};  # don't store the string
        }
    }

    my $self = bless {
        path     => $args{path},
        host     => $args{path} ? undef : ($args{host} // 'localhost'),
        port     => $args{path} ? undef : ($args{port} // 6379),
        socket   => undef,
        parser   => undef,
        connected          => 0,
        _socket_live       => 0,
        _fatal_in_progress => 0,
        _reader_running    => 0,   # dedup guard; the selector owns the reader Future itself
        _write_lock        => undef,     # will be a Future used as a lock, populated lazily
        _reconnect_future  => undef,
        _tasks             => Future::Selector->new,