Async-Redis

 view release on metacpan or  search on metacpan

Changes  view on Meta::CPAN

        - Pipeline results contain error objects at failed slot positions
    - Bug Fix: Pool max connection race condition
        - Concurrent acquire() calls could exceed max connections
        - Added _creating counter to track in-flight connection creations
    - Dependency Change: Future::IO bumped to 0.23 (was 0.19)
        - Future::IO 0.22+ broke load_impl('IOAsync') by adding a poll
          method check that IO::Async's impl doesn't satisfy
        - Removed IO::Async as a test dependency entirely
        - Tests now use Future::IO's built-in poll-based default impl
        - Test suite is fully event-loop agnostic
    - Examples: pagi-chat stats timer uses Future::IO instead of IO::Async

0.001005  2026-03-15
    - Feature: Unix domain socket connections
        - Connect via path parameter or redis+unix:// URI scheme
        - Constructor stores path and skips host/port for unix sockets
        - Added docker-compose redis-unix service for testing
    - Feature: PubSub auto-resubscribe on reconnect
        - Subscriptions automatically re-established after connection drop
        - on_reconnect callback on Subscription object for notification
        - _read_pubsub_frame checks connected state to fail fast

examples/bulk-insert/app.pl  view on Meta::CPAN

        my @futures;
        for my $i ($first .. $last) {
            push @futures, $redis->set("$prefix$i", $payload, ex => $opts->{ttl});
        }

        $state->{issued} += scalar @futures;
        await Future->needs_all(@futures);
        $state->{confirmed} += scalar @futures;
        $state->{batches}++;

        # Give timer-based tasks a scheduling point between large batches.
        await Future::IO->sleep(0);
    }

    return;
}

async sub heartbeat {
    my ($redis, $state, $opts, $start) = @_;

    while (!$state->{done}) {

examples/pagi-chat/lib/ChatApp/State.pm  view on Meta::CPAN

    # Separate connection for PubSub
    my $pubsub_redis = Async::Redis->new(host => $host, port => $port);
    await $pubsub_redis->connect;
    $pubsub = await $pubsub_redis->subscribe(BROADCAST_CHANNEL);

    # Initialize background task selector and start the runner
    $background_selector = Future::Selector->new;
    _start_selector_runner();

    # Start periodic stats timer (every 10 seconds)
    _start_stats_timer();

    # Initialize default rooms
    await add_room('general', 'system');
    await add_room('random', 'system');
    await add_room('help', 'system');

    print STDERR "Worker $$: Redis state initialized\n";
    return $redis;
}

sub get_redis { $redis }
sub get_pubsub { $pubsub }

# Background selector runner
my $selector_runner_future;

# Periodic stats timer
my $stats_timer;

# Start the selector with the PubSub listener as the main long-running task
sub _start_selector_runner {
    # Add the broadcast listener as a long-running task
    # Use gen => to provide a generator that creates the future
    # (f => expects a completed future, gen => expects a coderef)
    $background_selector->add(
        data => 'pubsub-listener',
        gen  => sub { _broadcast_listener() },
    );

    # Run the selector in the background
    $selector_runner_future = $background_selector->run->on_fail(sub {
        my ($err) = @_;
        warn "[selector] Runner failed: $err";
    })->retain;
}

# Start periodic stats timer using Future::IO (event-loop agnostic)
sub _start_stats_timer {
    $stats_timer = (async sub {
        while (1) {
            await Future::IO->sleep(10);
            next unless %local_sessions;  # Skip if no clients
            eval {
                my $stats = await get_stats();
                my $msg = $JSON->encode({
                    global  => 1,
                    payload => {
                        type         => 'stats',
                        users_online => $stats->{users_online},

examples/pagi-chat/public/js/app.js  view on Meta::CPAN

            console.log(`Reconnecting in ${Math.round(delay)}ms (attempt ${state.reconnectAttempts})...`);
            setConnectionStatus('reconnecting');
            setTimeout(connectWebSocket, delay);
        };

        state.ws.onerror = (error) => {
            console.error('WebSocket error:', error);
        };

        state.ws.onmessage = (event) => {
            // Any message resets the heartbeat timer
            state.lastPongTime = Date.now();

            try {
                const data = JSON.parse(event.data);
                handleWebSocketMessage(data);
            } catch (e) {
                console.error('Failed to parse message:', e);
            }
        };
    }

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

}

# 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);

t/01-unit/await-with-deadline.t  view on Meta::CPAN

};

subtest 'timeout wins' => sub {
    my $read_f = Future->new;
    my $wait   = $helper->($read_f, Time::HiRes::time() + 0.05);
    my ($returned_f, $timed_out) = $wait->get;
    is $timed_out, 1, 'timed out';
    ok !$read_f->is_ready, 'read future still pending';
};

subtest 'timeout timer cancelled when read wins' => sub {
    # This is a hygiene check: after read_f completes, the internal
    # timeout Future must be cancelled so no stray timer fires later.
    # We verify indirectly: run the event loop after completion and
    # assert no exceptions / no extra work.
    my $read_f = Future->new;
    my $wait   = $helper->($read_f, Time::HiRes::time() + 0.05);
    $read_f->done('fast');
    $wait->get;
    # Pump the loop past where the timer would have fired.
    Future::IO->sleep(0.1)->get;
    ok 1, 'no stray timer fired';
};

done_testing;