view release on metacpan or  search on metacpan

t/22_lifecycle.t  view on Meta::CPAN

use strict;
use warnings;
use Test::More;
use IO::Socket::INET;
use Errno qw(EAGAIN EWOULDBLOCK);
use Scalar::Util qw(weaken);
use B;
use EV;
use EV::Kafka;

# Regression tests for connection object lifecycle and memory ownership:
#   M1  response mortals are reclaimed per event, not when EV::run returns
#   M2  DESTROY fires every pending callback (not just the first)
#   M5  a custom EV::Loop is refcounted by the conn
#   M6  explicit DESTROY inerts the blessed ref; stale use croaks
#   M7  conn_emit_error warns (not croaks) when no on_error is installed
#   M8  disconnect() from on_disconnect/on_error neither recurses nor
#       fires on_disconnect twice
#   M9  pending-request error callbacks fire before on_disconnect
#
# Uses the in-process mock-broker pattern from t/20_mock_conn.t — no live
# broker needed.

plan tests => 21;

sub i16 { pack 'n', $_[0] }
sub i32 { pack 'N', $_[0] }

my $apis_body =
      i16(0)         # no error
    . i32(2)         # 2 entries
    . i16(0) . i16(0) . i16(7)     # API_PRODUCE, v0..v7
    . i16(18) . i16(0) . i16(0);   # API_API_VERSIONS, v0..v0

# Start a mock broker on a free port. $on_request->($api_key, $corr_id,
# $send, $ctx) runs for each full request frame; $send->($corr, $body)
# frames and writes a response. $ctx keeps watchers/sockets/timers alive
# and offers $ctx->{close_client}. Returns ($port, $ctx).
sub mock_broker {
    my (%opt) = @_;
    my $server = IO::Socket::INET->new(
        LocalAddr => '127.0.0.1',
        LocalPort => 0,
        Listen    => 1,
        Proto     => 'tcp',
        ReuseAddr => 1,
    ) or BAIL_OUT "cannot bind localhost listener: $!";
    $server->blocking(0);

    my $ctx = { server => $server, timers => [] };
    my $on_request = $opt{on_request} || sub {
        my ($api, $corr, $send) = @_;
        $send->($corr, $apis_body) if $api == 18;
    };

    $ctx->{close_client} = sub {
        undef $ctx->{read_w};
        close $ctx->{client_fh} if $ctx->{client_fh};
        undef $ctx->{client_fh};
    };

    $ctx->{accept_w} = EV::io fileno($server), EV::READ, sub {
        my $fh = $server->accept or return;
        $fh->blocking(0);
        $ctx->{client_fh} = $fh;
        my $send = sub {
            my ($corr, $body) = @_;
            return unless $ctx->{client_fh};
            my $payload = i32($corr) . $body;
            syswrite $ctx->{client_fh}, i32(length $payload) . $payload;
        };
        my $incoming = '';
        $ctx->{read_w} = EV::io fileno($fh), EV::READ, sub {
            my $buf;
            my $n = sysread $fh, $buf, 4096;
            if (!defined $n) {
                return if $!{EAGAIN} || $!{EWOULDBLOCK};
                undef $ctx->{read_w};
                return;
            }
            if ($n == 0) { undef $ctx->{read_w}; return; }
            $incoming .= $buf;
            while (length($incoming) >= 4) {
                my $size = unpack 'N', substr($incoming, 0, 4);
                last if length($incoming) < 4 + $size;
                my $api  = unpack 'n', substr($incoming, 4, 2);
                my $corr = unpack 'N', substr($incoming, 8, 4);
                substr($incoming, 0, 4 + $size) = '';
                $on_request->($api, $corr, $send, $ctx);
            }
        };
    };

    return ($server->sockport, $ctx);
}

sub timeout_w { EV::timer 5, 0, sub { diag "test timed out"; EV::break } }

# --- M1: a parsed response must not stay alive past its event -------------
# Broker answers the first Metadata request immediately and the second
# 0.2s later, so the two callbacks run in SEPARATE watcher events.  The
# first callback keeps only a weak ref to its result; if response mortals
# are pinned until EV::run returns (the bug), $saved is still defined when
# the second callback runs.
{
    my $meta_body = i32(0) . i32(1) . i32(0);  # v1: 0 brokers, controller 1, 0 topics
    my $meta_seen = 0;
    my ($port, $broker) = mock_broker(on_request => sub {
        my ($api, $corr, $send, $ctx) = @_;
        if ($api == 18) { $send->($corr, $apis_body); return; }
        if ($api == 3) {
            $meta_seen++;
            if ($meta_seen == 1) {
                $send->($corr, $meta_body);
            } else {
                push @{ $ctx->{timers} },
                    EV::timer(0.2, 0, sub { $send->($corr, $meta_body) });
            }
        }
    });

    my $conn = EV::Kafka::Conn::_new('EV::Kafka::Conn', undef);
    my $saved;
    $conn->on_error(sub { diag "M1 error: $_[0]"; EV::break });
    $conn->on_connect(sub {
        $conn->metadata(undef, sub { $saved = $_[0]; weaken($saved); });
        $conn->metadata(undef, sub {
            ok !defined($saved),
                'M1: response mortal is reclaimed when its event ends';
            EV::break;
        });
    });
    $conn->connect('127.0.0.1', $port, 5.0);
    my $t = timeout_w();
    EV::run;
}

# --- M2: DESTROY must fire every pending callback -------------------------
{
    my ($port, $broker) = mock_broker();    # handshake only, swallows the rest
    my $conn = EV::Kafka::Conn::_new('EV::Kafka::Conn', undef);
    my $ready = 0;
    $conn->on_error(sub { diag "M2 error: $_[0]"; EV::break });
    $conn->on_connect(sub { $ready = 1; EV::break });
    $conn->connect('127.0.0.1', $port, 5.0);
    my $t = timeout_w();
    EV::run;
    ok $ready, 'M2: handshake complete';

    my (@fired, @refs);
    for (1..3) {
        my $cb = sub { push @fired, $_[1] };
        push @refs, $cb;
        weaken($refs[-1]);
        $conn->metadata(undef, $cb);
    }
    is $conn->pending, 3, 'M2: three requests in flight';

    $conn->DESTROY;