PAGI-Server
view release on metacpan or search on metacpan
had. Pinned by t/http2/40-pending-io-at-disconnect.t.
Specification conformance (PAGI 0.002006: core spec 0.5 / Www 0.4)
- Scopes report pagi.version '0.5' and, for http/websocket/sse,
spec_version '0.4' -- the feature-detection gate for the new
settlement contract, which this server already implemented. The
lifespan scope keeps spec_version '0.3' (its sub-spec did not bump).
- New end-to-end conformance tests pin the settlement contract over
real sockets on every protocol: a send parked on backpressure at
abrupt disconnect resolves (never fails, never hangs), the resumed
coroutine observes the completed state transition, on_disconnect is
never invoked inside the application's send call frame, a pending
receive resolves with the protocol's disconnect event, and h2
RST_STREAM settles per-stream I/O without cancelling the application
Future (t/61-pending-io-at-disconnect.t,
t/http2/40-pending-io-at-disconnect.t).
Maintenance
- t/http2/36 builds its 20KB payload once, outside its send loop: a
repeat-op inside a foreach that also awaits yields undef on later
iterations under Future::AsyncAwait on ITHREADS perls (upstream bug,
timeout, server shutdown, ...). arm_drain previously piggybacked
directly on the same Future queue a blocking $send await uses, so
teardown resolved both indiscriminately; a new, separate _drain_fires
list now holds the on_drain callback, fired only on a genuine drain and
dropped unfired on teardown, matching how HTTP/2 already kept
stream_drain_waiters and transport_drain_fires apart. A producer parked
on the blocking backpressure path still resumes on teardown either way.
- _handle_disconnect now marks the HTTP connection-state object (and, on
HTTP/2, every open stream's connection-state) as disconnected before
cancelling pending drain waiters, not after. Cancelling a drain waiter
can synchronously resume an awaiting app coroutine (a $send blocked on
backpressure), and that resumed code's first act may be to check
is_connected()/disconnect_reason() -- previously it could observe a
stale "still connected" snapshot for the instant between its own
resumption and the state being marked.
- Fail the Future returned by $send when a file response's handle cannot be
read, instead of treating a read error as EOF. Applications may now safely
close a file resource only after the corresponding send Future resolves.
- The HTTP/1 SSE decline response now sends Connection: close explicitly, so
pooled clients do not attempt to reuse a socket the server has already
closed.
lib/PAGI/Server/Compliance.pod view on Meta::CPAN
surface and semantics -- C<buffered_amount>, the high/low watermarks,
C<on_high_water>/C<on_drain> -- whether the connection is HTTP/1.1 or
HTTP/2.
C<on_drain> fires only for a genuine drain -- the buffer actually falling
back below the low mark -- on both transports. Tearing a connection down
while the buffer is still above the high mark (client disconnect, timeout,
server shutdown, ...) does not fire C<on_drain>: the connection is going
away, not draining. A producer parked on the blocking backpressure path
(a C<$send> awaiting the buffer to drain) still resumes on teardown, so no
coroutine is left hanging; only the app-facing hysteresis callback is
withheld.
=head2 WebSocket over HTTP/2 (RFC 8441)
PAGI::Server accepts a WebSocket upgrade over HTTP/2 through RFC 8441
Extended CONNECT (C<:protocol =E<gt> 'websocket'>). Each accepted stream carries
its own C<websocket.receive>/C<websocket.disconnect> event stream,
multiplexed alongside every other stream on the connection.
=head3 Framing Enforcement
lib/PAGI/Server/Connection.pm view on Meta::CPAN
$reason //= 'connection closed';
my @waiters = splice @{$self->{_drain_waiters}};
for my $f (@waiters) {
# Resolve (not fail) - app should check connection state after await
$f->done unless $f->is_ready;
}
# Drop (don't fire) the app's on_drain fires: the connection is going
# away, not draining -- matches h2's teardown handling of
# transport_drain_fires. The blocking waiters above still resume (so no
# coroutine leak), but on_drain is a hysteresis signal for a buffer that
# actually fell back below the low mark, which never happened here.
$self->{_drain_fires} = [];
$self->{_drain_check_active} = 0;
}
# HTTP/2 per-stream backpressure: the h2 analogue of _wait_for_drain. Resolves
# when this stream's send queue falls below the low watermark. Each multiplexed
# stream is bounded independently, so a quiet TCP buffer can't let one stream's
# queue grow without limit.
sub _h2_wait_for_stream_drain {
lib/PAGI/Server/Connection.pm view on Meta::CPAN
# so the guard on $stream->{connection_state} skips them naturally.
if ($self->{is_h2} && $self->{h2_streams} && !$is_completion) {
for my $stream (values %{$self->{h2_streams}}) {
$stream->{connection_state}->_mark_disconnected($reason)
if $stream->{connection_state};
}
}
# Cancel any pending drain waiters (backpressure) AFTER connection state
# is marked above: resolving a parked waiter can synchronously resume an
# awaiting app coroutine (Future::AsyncAwait resumes inline off ->done),
# and that resumed app's first act may be to read is_connected() /
# disconnect_reason() -- those must already reflect this disconnect, not
# a stale "still connected" snapshot from before it was detected.
$self->_cancel_drain_waiters($reason);
# Record the abnormal reason so the WebSocket disconnect event reports it
# (instead of the old empty string). SSE tracks its own reason at the
# detection sites via sse_disconnect_reason.
if ($self->{websocket_mode} && !$is_completion) {
$self->{ws_disconnect_reason} = $reason;
lib/PAGI/Server/Connection.pm view on Meta::CPAN
}
# Combined disconnect and close - use this from callbacks where $weak_self may
# become undefined after _handle_disconnect completes its Future callbacks.
# This method holds a strong reference to $self throughout the operation.
sub _handle_disconnect_and_close {
my ($self, $reason) = @_;
# Mark the transport closed before notifying: _handle_disconnect below
# completes any pending receive(), which can synchronously resume the
# app coroutine (it may run straight through a subsequent send()), so
# the send-side closed-check needs "closed" to already be true at that
# point (spec order: closed-check precedes validation). Resource
# cleanup itself still happens in _close, gated by its own idempotency
# flag so it isn't skipped by this early flip.
$self->{closed} = 1;
$self->_handle_disconnect($reason);
$self->_close;
}
t/23-connection-cleanup.t view on Meta::CPAN
};
# =============================================================================
# Test: h1 teardown with the outbound buffer above the high mark must NOT
# fire on_drain (the connection is going away, not draining -- h2 already
# gets this right via its separate transport_drain_fires list; h1's arm_drain
# used to piggyback directly on the same _drain_waiters Futures that
# blocking producers await, so tearing down resolved both indiscriminately
# and fired on_drain with the buffer nowhere near the low mark). A producer
# genuinely parked on a blocking backpressure await must still resume (no
# coroutine leak) -- only the app-facing on_drain callback must be dropped.
#
# Exercises the real Connection-level machinery (_h1_transport_state's
# arm_drain, _wait_for_drain, _cancel_drain_waiters) directly against a
# fake stream that reports a controlled, non-draining buffer size -- a
# unit-level handler invocation, not a real socket, so the buffer-stays-full
# precondition is exact rather than timing-dependent.
# =============================================================================
package Local::FakeWriter {
sub new { my ($c, $d) = @_; return bless { data => $d }, $c }
t/23-connection-cleanup.t view on Meta::CPAN
# A genuinely parked producer, awaiting the buffer to drain before its
# next write -- must still resume when the connection tears down.
my $parked = $conn->_wait_for_drain;
ok(!$parked->is_ready, 'parked wait is pending before teardown (buffer still above low mark)');
# Teardown: the buffer is still full (nothing has actually drained).
$conn->_cancel_drain_waiters('connection closing');
is($drain_fired, 0, 'on_drain did NOT fire on teardown (connection going away, not draining)');
ok($parked->is_ready, 'the parked blocking wait resumed anyway (no coroutine leak)');
};
subtest 'h1 on_drain still fires normally when the buffer genuinely drains (not torn down)' => sub {
my $fake_loop = IO::Async::Loop->new;
my $fake_server = Local::FakeServer->new($fake_loop);
my $fake_stream = Local::FakeStream->new;
my $conn = PAGI::Server::Connection->new(
stream => $fake_stream, server => $fake_server, app => sub { },
write_high_watermark => 10, write_low_watermark => 2,
t/37-connection-state.t view on Meta::CPAN
# A producer parked on a blocking backpressure await -- pushed directly
# onto _drain_waiters (the same queue _wait_for_drain uses), without
# needing a real stream/buffer to get there.
my $parked = Future->new;
push @{$conn->{_drain_waiters}}, $parked;
my ($observed_connected, $observed_reason);
$parked->on_ready(sub {
# Fires synchronously from within _handle_disconnect below, exactly
# as an awaiting coroutine resumes -- this is the resumed app's very
# first chance to look at its own connection state.
$observed_connected = $conn_state->is_connected;
$observed_reason = $conn_state->disconnect_reason;
});
$conn->_handle_disconnect('client_closed');
is($observed_connected, 0,
'resumed waiter observes is_connected already false (no stale-true window)');
is($observed_reason, 'client_closed',
t/52-mandatory-validation.t view on Meta::CPAN
my $ws_server = PAGI::Server->new(app => $ws_app, host => '127.0.0.1', port => 0, quiet => 1, validate_events => 0);
$loop->add($ws_server);
$ws_server->listen->get;
my $ws_port = $ws_server->port;
# Raw handshake boilerplate lifted from t/04-websocket.t: the app under test
# reads/writes application-level events, so a raw socket lets us drive the
# handshake without a WebSocket client library getting in the way. Pumps the
# loop until the server closes the connection, so that whatever the app
# coroutine recorded in its package vars is settled by the time we assert.
use IO::Socket::INET;
my $ws_handshake_and_drain = sub {
my ($port) = @_;
my $sock = IO::Socket::INET->new(
PeerAddr => '127.0.0.1',
PeerPort => $port,
Proto => 'tcp',
Timeout => 5,
);
t/59-ws-disconnect-exactly-once.t view on Meta::CPAN
# Normal, well-behaved app: stop calling receive() the moment the
# first websocket.disconnect arrives -- exactly one call, ever.
my $ev = await $receive->();
push @seen, $ev;
# Stay alive well past the TCP close that follows the peer's Close
# frame (~50ms later, per the reproduction) without calling receive()
# again. This isolates the on_closed path as the only thing that can
# still act after the Close frame was handled -- the app's own
# session-complete teardown (which fires the moment this coroutine
# returns) must not be what triggers a second delivery.
await $loop->delay_future(after => 0.5);
};
my $server = PAGI::Server->new(
app => $test_app,
host => '127.0.0.1',
port => 0,
quiet => 1,
);
t/61-pending-io-at-disconnect.t view on Meta::CPAN
#!/usr/bin/env perl
# =============================================================================
# Test: Settlement of pending I/O at disconnect (spec 0.5 / Www 0.4)
#
# Pins the now-normative contract end-to-end over real sockets:
# 1. A send Future parked on backpressure when the client disconnects
# settles by RESOLVING successfully -- never fails, never hangs.
# 2. The resumed coroutine observes the connection-state transition already
# complete: is_connected() false, disconnect_reason() set (Www.pod
# State Transition Order invariant).
# 3. on_disconnect callbacks are never invoked synchronously within the
# application's call into $send (callback invocation context).
# 4. A receive Future pending at disconnect resolves with the protocol's
# disconnect event (http.disconnect / websocket.disconnect /
# sse.disconnect).
#
# HTTP/2 coverage (RST_STREAM settlement) lives in
# t/http2/40-pending-io-at-disconnect.t.
t/61-pending-io-at-disconnect.t view on Meta::CPAN
my $f = $send->({ type => 'http.response.body', body => $chunk, more => 1 });
$in_send_frame = 0;
my $ok = eval { await $f; 1 };
unless ($ok) {
$obs{send_failed} = "$@";
last;
}
$obs{completed}++;
if (!$conn->is_connected && !$obs{resumed}) {
# The invariant under test: by the time the awaiting
# coroutine resumes, the transition is already complete.
$obs{resumed} = {
connected => $conn->is_connected ? 1 : 0,
reason => $conn->disconnect_reason,
};
last;
}
}
$obs{app_completed} = 1;
return;
};
t/61-pending-io-at-disconnect.t view on Meta::CPAN
ok($parked, 'a send is parked on backpressure')
or diag("started=$obs{started} completed=$obs{completed}");
close($sock);
ok(pump_until(sub { $obs{app_completed} }, 10),
'application completed after disconnect (parked send did not hang)');
ok(!$obs{send_failed}, 'the parked send resolved successfully, not failed')
or diag("send failed with: $obs{send_failed}");
ok($obs{resumed}, 'application resumed from the parked send and observed disconnect');
is($obs{resumed}{connected}, 0, 'resumed coroutine sees is_connected false');
like($obs{resumed}{reason}, qr/^(client_closed|write_error|read_error)$/,
'resumed coroutine sees a standard disconnect reason');
is($obs{cb_count}, 1, 'on_disconnect fired exactly once');
is($obs{cb_in_send_frame}, 0,
'on_disconnect was not invoked inside the application send call frame');
is($obs{cb_reason}, $obs{resumed}{reason}, 'callback and accessor agree on the reason');
shutdown_server($server);
};
# =============================================================================
# 2. HTTP: pending receive resolves with http.disconnect
t/http2/15-sse-keepalive.t view on Meta::CPAN
# for a long while yet, so this cannot race B's teardown.
my $before_b = comment_count($b_data, 'pingB');
my $settle_deadline = Time::HiRes::time() + 0.7;
while (Time::HiRes::time() < $settle_deadline) {
exchange_frames($client, $client_sock, 1);
}
ok(comment_count($b_data, 'pingB') > $before_b,
"B's keepalive kept ticking after A's stream closed")
or diag("b_data: $b_data");
# Drain: let B's own app coroutine reach its 2.5s close and return
# naturally before tearing down, so no suspended async sub is abandoned
# mid-await. Ceiling 6s is a wide margin over B's own budget.
$deadline = Time::HiRes::time() + 6;
while (Time::HiRes::time() < $deadline && !$closed{$b_id}) {
exchange_frames($client, $client_sock, 1);
}
ok($closed{$b_id}, "B's own close completed before teardown");
$stream_io->close_now;
$loop->remove($server);
t/http2/15-sse-keepalive.t view on Meta::CPAN
# ============================================================
# Per-stream SSE idle timeout independence (design section 11.3)
# ============================================================
# Before this task, sse_idle_timer lived on $self (the connection): only
# the FIRST stream's sse.start actually armed it (a `return if
# $self->{sse_idle_timer}` guard on the second), and on expiry it called
# the connection-wide _handle_disconnect_and_close, tearing down every
# multiplexed stream -- not just the idle one. This proves one idle stream
# closes on its own while an active sibling (which keeps resetting its own
# idle timer via sse.send) is unaffected.
my $active_done = 0; # flips true once the '/active' app coroutine returns
subtest 'per-stream SSE idle timeout: an idle stream closes without killing an active sibling' => sub {
$active_done = 0;
my $app = async sub {
my ($scope, $receive, $send) = @_;
return unless $scope->{type} eq 'sse';
await $receive->();
await $send->({ type => 'sse.start', status => 200 });
t/http2/15-sse-keepalive.t view on Meta::CPAN
await $receive->();
}
elsif ($scope->{path} eq '/active') {
# First send immediately (no delay), so this stream already has
# data well before the idle stream's independently-timed 0.3s
# timeout can fire -- the two streams are dispatched moments
# apart, so a delayed first send here would race that closure
# instead of reliably preceding it. Then resets its own idle
# timer on every further send (8 * 0.1s = 0.8s), comfortably
# outliving the 0.3s idle timeout, but still short enough that
# the coroutine returns (rather than being abandoned mid-await)
# before this subtest's own teardown -- see $active_done below.
await $send->({ type => 'sse.send', data => 'tick0' });
for my $i (1 .. 8) {
await $loop->delay_future(after => 0.1);
await $send->({ type => 'sse.send', data => "tick$i" });
}
$active_done = 1;
}
};
t/http2/15-sse-keepalive.t view on Meta::CPAN
# enough to land mid-loop (active's total budget is 0.8s from its own
# dispatch, which starts only slightly after idle's).
my $before = length($active_data);
my $settle_deadline = Time::HiRes::time() + 0.3;
while (Time::HiRes::time() < $settle_deadline) {
exchange_frames($client, $client_sock, 1);
}
ok(length($active_data) > $before,
'active stream continued receiving data after the sibling idle-timeout');
# Drain: let the active app coroutine's bounded loop finish and return
# naturally before tearing down, so no suspended async sub is abandoned
# mid-await. Ceiling 5s is a wide margin over its ~0.8s own budget.
my $drain_deadline = Time::HiRes::time() + 5;
while (Time::HiRes::time() < $drain_deadline && !$active_done) {
exchange_frames($client, $client_sock, 1);
}
ok($active_done, "active stream's app coroutine returned before teardown");
$stream_io->close_now;
$loop->remove($server);
};
done_testing;
t/http2/18-transport-leak.t view on Meta::CPAN
);
$client_sock->syswrite($client->mem_send);
# Pump until the stream completes and closes (request fully handled).
pump($client, $client_sock, sub { $stream_closed });
ok($stream_closed, 'stream completed and closed');
ok($saw_handle, 'transport handle was attached to the h2 scope');
# Drive deferred teardown (loop->later) AND adopted-future cleanup so the
# scope/coroutine that transiently hold the handle are released. Break early
# once the probe is collected.
for (1 .. 200) {
last unless defined $probe;
$loop->loop_once(0.01);
}
is($probe, undef,
'transport handle (and its $ss cycle) collected after teardown; no leak');
$stream_io->close_now;
t/http2/20-sse-transport.t view on Meta::CPAN
ok($hit_high, 'on_high_water fired when the per-stream queue exceeded the high mark');
ok($hit_drain, 'on_drain fired once nghttp2 drained the queue below the low mark');
$stream_io->close_now;
$loop->remove($server);
};
subtest 'SSE-over-h2 transport handle (and its $ss cycle) is collected at teardown' => sub {
# The app weak-probes its OWN transport handle (race-free: the app always
# runs), sends a few events, then returns. With the app coroutine complete
# the scope is released, so the handle is held only by $ss->{transport_state}.
# A client RST_STREAM drives _h2_on_close, which must delete that ref and
# break the cycle -- otherwise the stream state leaks for the life of the
# process (one per SSE request).
my ($saw_handle, $probe);
my $app = async sub {
my ($scope, $receive, $send) = @_;
await $receive->();
t/http2/26-mandatory-validation.t view on Meta::CPAN
'send after http response complete raises on h2, not silently swallowed' );
(undef, $body) = get_h2('/ok', app => $http_app);
is( $body, 'NO-ERROR', 'a conforming app is unaffected' );
# ============================================================
# SSE: mis-sequencing after a terminal state raises, not swallowed
# ============================================================
# Once a stream is 'closed' (sse.close) or 'decline_complete', its
# h2_streams entry is reclaimed asynchronously by _h2_on_close. These probe
# sends happen on the very next tick of the same app coroutine -- before
# that reclaim can plausibly have run -- but the sequence check must not
# rely on that timing: it consults the closure-local $seq, not the h2_streams
# entry, precisely so a send after the entry is actually gone still raises.
sub sse_probe {
my (%args) = @_;
my ($conn, $stream_io, $client_sock, $server) =
create_h2c_connection(app => $args{app});
my $client = create_client();
h2c_handshake($client, $client_sock);
t/http2/38-ws-queue-overflow.t view on Meta::CPAN
# the configured cap, the next inbound message closes the connection with
# 1008 ("Message queue overflow") and delivers a queue_overflow
# websocket.disconnect instead of queueing without bound. h2's
# _h2_process_ws_frames had no such check: a client flooding messages
# faster than the app drains receive() could grow a stream's receive_queue
# without limit -- an unbounded per-connection memory DoS on a
# transport that otherwise mirrors HTTP/1.1's WebSocket support exactly.
#
# This uses an app that never calls receive() at all (accepts and returns
# immediately) as its "never drains" case -- the stream stays open at the
# h2 level (nothing about returning from a WebSocket app coroutine tears
# the stream down; only an explicit close/END_STREAM does), so every
# flooded message is available to inspect via white-box access to the
# stream's own receive_queue.
use PAGI::Server::Connection;
use PAGI::Server;
use PAGI::Server::Protocol::HTTP1;
use PAGI::Server::Protocol::HTTP2;
use Protocol::WebSocket::Frame;
t/http2/38-ws-queue-overflow.t view on Meta::CPAN
sub close_codes {
my ($raw) = @_;
return map { unpack('n', substr($_->{bytes}, 0, 2)) }
grep { $_->{opcode} == 8 && length($_->{bytes}) >= 2 }
extract_ws_frames($raw);
}
# App that accepts and immediately returns -- it never calls receive() at
# all, so nothing ever drains the stream's receive_queue. Nothing about a
# WebSocket app coroutine returning tears the h2 stream down by itself
# (only an explicit close/END_STREAM does), so the stream stays open and
# every flooded message lands in receive_queue for white-box inspection.
sub make_never_draining_app {
return async sub {
my ($scope, $receive, $send) = @_;
return unless $scope->{type} eq 'websocket';
await $send->({ type => 'websocket.accept' });
return;
};
}
t/http2/40-pending-io-at-disconnect.t view on Meta::CPAN
# ============================================================
# Test: RST_STREAM settles pending I/O (spec 0.5 / Www 0.4)
# ============================================================
# The spec maps RST_STREAM to the standard abnormal-disconnect transition
# for that stream's scope. Pinned here, end-to-end over a real nghttp2
# client session:
# 1. A send Future parked on the stream (flow-control window exhausted, no
# client WINDOW_UPDATEs) settles by RESOLVING when the client resets the
# stream -- never fails, never hangs.
# 2. The resumed coroutine observes is_connected() false with the reason
# set, and on_disconnect fired outside the application send call frame.
# 3. The application Future is NOT cancelled: post-disconnect cleanup that
# must be resumed by the event loop still runs to completion.
# 4. A receive Future pending at RST_STREAM resolves with http.disconnect.
#
# h1/WebSocket/SSE settlement coverage lives in
# t/61-pending-io-at-disconnect.t.
use PAGI::Server::Connection;
use PAGI::Server;
t/http2/40-pending-io-at-disconnect.t view on Meta::CPAN
if (!$conn->is_connected && !$obs{resumed}) {
$obs{resumed} = {
connected => $conn->is_connected ? 1 : 0,
reason => $conn->disconnect_reason,
};
last;
}
}
# Post-disconnect cleanup that requires the event loop to resume
# this coroutine again: only reachable if the application Future
# was settled-with, not cancelled out from under, the app.
await $loop->delay_future(after => 0.05);
$obs{cleanup_ran} = 1;
$obs{app_completed} = 1;
return;
};
my ($conn, $stream, $client_sock, $server) = create_h2_connection(app => $app);
my $client = create_client();
complete_h2_handshake($client, $client_sock);
t/http2/40-pending-io-at-disconnect.t view on Meta::CPAN
# Client resets the stream (CANCEL).
$client->submit_rst_stream($stream_id, 8);
$client_sock->syswrite($client->mem_send);
ok(pump_until(sub { $obs{app_completed} }, 10),
'application completed after RST_STREAM (parked send did not hang)');
ok(!$obs{send_failed}, 'the parked send resolved successfully, not failed')
or diag("send failed with: $obs{send_failed}");
ok($obs{resumed}, 'application resumed from the parked send and observed disconnect');
is($obs{resumed}{connected}, 0, 'resumed coroutine sees is_connected false');
is($obs{resumed}{reason}, 'client_closed', 'RST_STREAM reports client_closed');
is($obs{cb_count}, 1, 'on_disconnect fired exactly once');
is($obs{cb_in_send_frame}, 0,
'on_disconnect was not invoked inside the application send call frame');
is($obs{cleanup_ran}, 1,
'post-disconnect cleanup ran (application Future was not cancelled)');
$server->shutdown->get;
eval { $loop->remove($server) };
};
t/sse-close.t view on Meta::CPAN
my $n = sysread($sock, $chunk, 4096);
if (defined $n && $n > 0) { $buf .= $chunk }
elsif (defined $n && $n == 0) { $eof = 1; last } # server closed the connection
last if $buf =~ $stop;
$loop->loop_once(0.05);
}
return ($buf, $eof);
}
# Open an SSE GET and read the stream up to its chunked terminator, then drain
# the loop so the app coroutine finishes. Returns (socket, wire, saw_eof); the
# socket is left open so callers can assert connection reuse on it.
sub sse_get {
my ($port) = @_;
my $sock = IO::Socket::INET->new(
PeerAddr => '127.0.0.1', PeerPort => $port, Proto => 'tcp', Timeout => 5,
) or return (undef, '', 0);
print $sock "GET / HTTP/1.1\r\nHost: 127.0.0.1:$port\r\nAccept: text/event-stream\r\n\r\n";
$sock->blocking(0);
my ($wire, $eof) = read_until($sock, qr/\r\n0\r\n\r\n/);
$loop->loop_once(0.05) for 1 .. 20; # let the app coroutine run to completion
return ($sock, $wire, $eof);
}
# Issue an ordinary request on an already-used socket.
sub plain_request_on {
my ($sock, $port) = @_;
print $sock "GET /after HTTP/1.1\r\nHost: 127.0.0.1:$port\r\n\r\n";
my ($wire) = read_until($sock, qr/REUSED/);
return $wire;
}
t/sse-decline.t view on Meta::CPAN
# that would look exactly like an abnormal end that never happened.
my ($settled, $event);
my $app = async sub {
my ($scope, $receive, $send) = @_;
await $send->({ type => 'sse.http.response.start', status => 404, headers => [] });
await $send->({ type => 'sse.http.response.body', body => 'nope', more => 0 });
# Decline is now complete. Call receive() once more, WITHOUT
# awaiting it directly (an unresolved Future would hang this
# coroutine forever under the fix) -- just observe, after a bounded
# wait on the same loop, whether it ever resolved and with what.
my $future = $receive->();
await $loop->delay_future(after => 0.3);
$settled = $future->is_ready;
$event = $future->is_ready ? $future->get : undef;
};
my $server = create_server($app);
my ($wire, $eof) = sse_get($server->port);
( run in 2.368 seconds using v1.01-cache-2.11-cpan-364913b4093 )