view release on metacpan or  search on metacpan

t/50-pubsub/reconnect.t  view on Meta::CPAN

            connect_timeout => 2,
        );
        run { $r->connect };
        $r;
    };
    skip "Redis not available: $@", 6 unless $test_redis;
    $test_redis->disconnect;

    my $TEST_TIMEOUT = 5;

    # Helper: simulate a dropped connection by having the publisher
    # CLIENT KILL the subscriber. Server-side close → kernel delivers
    # EOF on our still-valid fd → reader sees 0 bytes → _reader_fatal
    # fires cleanly. Matches how a real peer-initiated disconnect works,
    # whereas `close $sub_redis->{socket}` leaves a stale poller in
    # Future::IO's select loop with undef fileno, which gets coerced to
    # fd 0 (STDIN) and deadlocks select in a TTY context.
    #
    # CLIENT commands are forbidden once a connection enters pub/sub
    # mode, so the caller MUST capture the subscriber's client ID
    # BEFORE subscribing and issue the KILL from a separate connection.

    subtest 'subscriber reconnects and receives messages after drop' => sub {
        my $pub = Async::Redis->new(
            host => $ENV{REDIS_HOST} // 'localhost',
        );
        run { $pub->connect };

        my $sub_redis = Async::Redis->new(
            host      => $ENV{REDIS_HOST} // 'localhost',
            reconnect => 1,
            reconnect_delay => 0.1,
        );
        run { $sub_redis->connect };

        # Capture subscriber's client ID before pubsub mode locks us out.
        my $sub_client_id = run { $sub_redis->client('ID') };

        my $subscription = run { $sub_redis->subscribe('reconnect:test') };
        ok($subscription, 'subscribed');

        # Verify pre-disconnect message works
        my $pre_future = (async sub {
            await Future::IO->sleep(0.1);
            await $pub->publish('reconnect:test', 'before');
        })->();

        my $msg1 = run { $subscription->next };
        is($msg1->{data}, 'before', 'received message before disconnect');
        run { $pre_future };

        # Server-initiated disconnect via KILL from pub connection.
        eval { run { $pub->client('KILL', 'ID', $sub_client_id) } };

        # Publish after reconnect window
        my $post_future = (async sub {
            await Future::IO->sleep(0.3);
            await $pub->publish('reconnect:test', 'after');
        })->();

        # next() should reconnect transparently and return real message
        my $next_f = $subscription->next;
        my $timeout_f = Future::IO->sleep($TEST_TIMEOUT)->then(sub {
            Future->fail("Timed out waiting for message after reconnect");
        });
        my $msg2 = eval { await_f(Future->wait_any($next_f, $timeout_f)); $next_f->get };

        if ($@) {
            fail("received message after reconnect: $@");
        } else {
            is($msg2->{type}, 'message', 'got real message type after reconnect');
            is($msg2->{data}, 'after', 'correct data after reconnect');
        }

        eval { run { $post_future } };

        is([$subscription->channels], ['reconnect:test'], 'still tracking channel');
        ok(!$subscription->is_closed, 'subscription not closed');

        eval { $pub->disconnect };
        eval { $sub_redis->disconnect };
    };

    subtest 'on_reconnect callback fires on reconnect' => sub {
        my $pub = Async::Redis->new(
            host => $ENV{REDIS_HOST} // 'localhost',
        );
        run { $pub->connect };

        my $sub_redis = Async::Redis->new(
            host      => $ENV{REDIS_HOST} // 'localhost',
            reconnect => 1,
            reconnect_delay => 0.1,
        );
        run { $sub_redis->connect };

        my $sub_client_id = run { $sub_redis->client('ID') };

        my $subscription = run { $sub_redis->subscribe('callback:test') };

        # Register callback
        my @events;
        $subscription->on_reconnect(sub {
            my ($sub) = @_;
            push @events, {
                channels => [$sub->channels],
                patterns => [$sub->patterns],
            };
        });

        # Server-initiated disconnect.
        eval { run { $pub->client('KILL', 'ID', $sub_client_id) } };

        my $publish_future = (async sub {
            await Future::IO->sleep(0.3);
            await $pub->publish('callback:test', 'hello');
        })->();

        # next() reconnects, fires callback, returns real message
        my $next_f = $subscription->next;
        my $timeout_f = Future::IO->sleep($TEST_TIMEOUT)->then(sub {