view release on metacpan or  search on metacpan

t/17_mock_protocol.t  view on Meta::CPAN

        or diag explain \@errs;
};

subtest 'inbound control line is capped' => sub {
    plan tests => 2;
    my $mock = MockNats->new(on_accept => sub {
        my ($c, $report) = @_;
        MockNats->handshake($c) or return;
        # Never send a newline again: rbuf must not grow without bound.
        my $sent = 0;
        for (1 .. 400) {
            my $n = syswrite($c, 'A' x 65536);
            last unless $n;
            $sent += $n;
            select undef, undef, undef, 0.01;
        }
        print $report "sent=$sent\n";
    })->start;

    my @errs;
    my $nats = EV::Nats->new(
        host => '127.0.0.1', port => $mock->port,
        reconnect => 0,
        on_error  => sub { push @errs, $_[0] },
    );
    run_guarded(8);
    $mock->stop;

    ok((grep { /maximum control line exceeded/ } @errs),
       'peer that never sends a newline is cut off')
        or diag "errors: @errs";
    ok !$nats->is_connected, 'connection torn down';
};

subtest 'tls_required server never sees a plaintext CONNECT' => sub {
    plan tests => 2;
    my $mock = MockNats->new(on_accept => sub {
        my ($c, $report) = @_;
        MockNats->send_info($c,
            '{"server_id":"fake","max_payload":1048576,"tls_required":true}');
        my $buf = MockNats->read_until($c, qr/CONNECT/, 3);
        print $report ($buf ? 'LEAKED' : 'NO-CONNECT') . "\n";
    })->start;

    my @errs;
    my $nats = EV::Nats->new(
        host => '127.0.0.1', port => $mock->port,
        user => 'u', pass => 'secret',   # credentials that must not leak
        reconnect => 0,
        on_error  => sub { push @errs, $_[0] },
    );
    run_guarded(6);
    my $verdict = $mock->report(5);
    $mock->stop;

    is $verdict, 'NO-CONNECT', 'no CONNECT on the wire for a tls_required server';
    ok((grep { /requires TLS/ } @errs), 'error explains the refusal')
        or diag "errors: @errs";
};

subtest 'flush() is not satisfied by a keepalive PONG' => sub {
    plan tests => 4;
    # The keepalive ping and flush() share one pong FIFO. The mock answers
    # the handshake PING, then holds back: PONG #1 only after TWO further
    # PINGs (one keepalive + one from flush), PONG #2 only after SIX. If a
    # keepalive PING pushes no FIFO placeholder, PONG #1 pops flush's
    # callback and it reports success on someone else's PONG.
    my $mock = MockNats->new(on_accept => sub {
        my ($c, $report) = @_;
        MockNats->handshake($c) or return;
        my ($pings, $pongs, $buf) = (0, 0, '');
        my $deadline = time + 12;
        while (time < $deadline && $pongs < 2) {
            my $rin = ''; vec($rin, fileno($c), 1) = 1;
            if (select(my $r = $rin, undef, undef, 0.1)) {
                my $n = sysread($c, my $ch, 4096); last unless $n;
                $buf .= $ch;
                my $new = () = $buf =~ /PING\r\n/g;
                if ($new) { $pings += $new; $buf = substr($buf, -5) }
            }
            if    ($pings >= 2 && $pongs == 0) { syswrite($c, "PONG\r\n"); $pongs = 1 }
            elsif ($pings >= 6 && $pongs == 1) { syswrite($c, "PONG\r\n"); $pongs = 2 }
        }
        print $report "pongs=$pongs\n";
    })->start;

    my ($flush_fired, $flush_err) = (0, 'unset');
    my $nats;
    $nats = EV::Nats->new(
        host => '127.0.0.1', port => $mock->port,
        ping_interval => 1000, max_pings_outstanding => 100,
        on_error   => sub { diag "error: $_[0]" },
        on_connect => sub {
            my $f; $f = EV::timer 1.4, 0, sub {
                undef $f;
                $nats->flush(sub {
                    ($flush_fired, $flush_err) = (1, $_[0]);
                    EV::break;
                });
            };
        },
    );
    # Phase 1: PONG #1 lands at ~+1.4s; the 6th PING (which unlocks
    # PONG #2) is not sent before +5s. 2.5s after arming, flush must
    # still be waiting for its own PONG.
    my $phase1 = EV::timer 3.9, 0, sub { EV::break };
    run_guarded(8);
    ok !$flush_fired, 'one PONG for two PINGs did not complete flush()';

    # Phase 2: the mock answers the backlog; flush must now complete.
    run_guarded(6);
    my $pongs = $mock->report(5);
    $mock->stop;
    ok $flush_fired, 'flush() completed once its own PONG arrived';
    is $flush_err, undef, 'flush() reported success';
    is $pongs, 'pongs=2', 'mock sent exactly the two scripted PONGs';
};

for my $mode ('die', 'ok') {
    subtest "batch { $mode } does not wedge the writer" => sub {
        plan tests => 2;
        my $mock = MockNats->new(on_accept => sub {
            my ($c, $report) = @_;
            MockNats->handshake($c) or return;
            my $buf = MockNats->read_until($c, qr/PUB after\.batch/, 5);
            print $report ($buf ? 'GOT' : 'MISSING') . "\n";