PAGI-Server
view release on metacpan or search on metacpan
frame (app messages, pong, the close-frame echo, keepalive ping)
directly to nghttp2's submit_data, which registers only a single
pending-send slot per stream; submitting a second frame before nghttp2
had fully drained the first destroyed that slot's still-queued
remainder instead of queuing behind it. Any two h2 WebSocket frames
that landed close enough together for the first to still be in flight
-- a real, reachable condition, not just a theoretical one -- could
silently lose or truncate data with no error raised anywhere.
- feat(h2): as part of the fix above, the h2 WebSocket send path now
frames every send through the same per-stream send queue and
pull-based data-provider model the h2 http-streaming and SSE send paths
already use (FIFO, so wire ordering is unaffected), and now provides
pagi.transport (buffered_amount, the high/low watermarks, and
on_high_water/on_drain), matching HTTP/1.1 WebSocket and every other h2
streaming scope type: from the application's perspective the transport
is invisible, the same pagi.transport surface and semantics apply
whether a WebSocket app runs over HTTP/1.1 or HTTP/2. Close-frame +
END_STREAM wire behavior is unchanged: the close frame still carries
END_STREAM on its own final DATA frame, and a send() that wakes from
backpressure after the close frame is already queued now correctly
no-ops instead of risking a frame after Close (RFC 6455 5.5.1).
- HTTP/2 requests whose HEADERS block exceeds max_header_list_size now
receive a real 431 (Request Header Fields Too Large) response instead
of a bare RST_STREAM, matching HTTP/1.1 and RFC 9113 section 10.5.1.
HPACK decoding of the oversized block still runs to completion (fields
stream used to reinitialize the request accumulator, so the trailer
block wiped the already-committed request state and triggered a second,
spurious application dispatch with an empty pseudo-header set; the
original request's body also never reached completion, since its
END_STREAM arrived on the trailer HEADERS rather than a DATA frame.
HEADERS blocks are now classified by nghttp2's own headers_category: a
later block on an established stream accumulates separately, is
validated and discarded (PAGI defines no request-trailer receive event
yet), and its END_STREAM still completes the original request's body.
- HTTP/1.1 http.response.body{file} sends now pre-check the file with -f
and -r before streaming, matching HTTP/2's existing behavior: a missing
file fails the send Future with "File not found: $file" and an unreadable
file fails with "Cannot read file: $file", instead of failing later
inside the file-send helper with a less specific stat/open error. The
response's own headers, if already flushed, are unaffected; a
conforming app may still recover by sending a normal body afterward.
- HTTP/1.1 websocket.accept's extra application-supplied headers now go
through the same Transfer-Encoding/Connection strip as every other h1
response-header path. Previously an app-supplied ['Connection', 'close']
(or 'Transfer-Encoding') reached the 101 Switching Protocols response
verbatim, alongside the server's own literal "Connection: Upgrade" line
- fix(server): populate the WebSocket disconnect reason/code on the
disconnect event; renamed the queue_overflow close path.
- feat(server): fire on_complete on clean HTTP request completion, distinct
from the abnormal-disconnect Future; the lifespan scope state is a
documented HashRef.
- fix(server): drop non-spec scope keys — pagi.features from all scopes, and
pagi.connection from HTTP/2 WebSocket/SSE scopes.
Backpressure / flow control
- feat: pagi.transport flow-control handle on HTTP, WebSocket and SSE scopes
(HTTP/1.1, plus HTTP/2 streaming responses and SSE-over-HTTP/2), exposing
buffered_amount and edge-triggered on_high_water / on_drain watermark
callbacks.
- fix(h2): bound streaming backpressure on the per-stream send queue rather
than the shared TCP buffer, and break a transport_state reference cycle at
stream teardown.
- feat(cli): expose --write-high-watermark and --write-low-watermark on
pagi-server, threading the write backpressure watermarks through to the
server constructor (previously settable only via the constructor API).
Lifespan
- feat(lifespan): lifespan_mode (auto|on|off) and a matching --lifespan CLI
flag.
- feat(lifespan): bound startup with lifespan_startup_timeout (default 30s).
Makefile.PL
README
README.md
SECURITY.md
UPGRADING.md
bin/pagi-server
cpanfile
dist.ini
examples/01-hello-http/README.md
examples/01-hello-http/app.pl
examples/02-streaming-response/README.md
examples/02-streaming-response/app.pl
examples/03-request-body/README.md
examples/03-request-body/app.pl
examples/04-websocket-echo/README.md
examples/04-websocket-echo/app.pl
examples/05-sse-broadcaster/README.md
examples/05-sse-broadcaster/app.pl
examples/06-lifespan-state/README.md
examples/06-lifespan-state/app.pl
examples/07-extension-fullflush/README.md
examples/07-extension-fullflush/app.pl
lib/PAGI/Server/Connection.pm
lib/PAGI/Server/ConnectionState.pm
lib/PAGI/Server/EventValidator.pm
lib/PAGI/Server/Protocol/HTTP1.pm
lib/PAGI/Server/Protocol/HTTP2.pm
lib/PAGI/Server/Runner.pm
lib/PAGI/Server/TransportState.pm
nytprof.out
t/00-load.t
t/01-hello-http.t
t/02-streaming.t
t/03-request-body.t
t/04-websocket.t
t/05-sse.t
t/06-lifespan.t
t/07-extensions.t
t/08-tls.t
t/10-http-compliance.t
t/11-multiworker.t
t/12-fork-loop-isolation.t
t/13-multiworker-signal.t
t/http2/02-server-config.t
t/http2/03-detection.t
t/http2/04-read-handler.t
t/http2/05-request-lifecycle.t
t/http2/06-integration.t
t/http2/07-websocket.t
t/http2/08-websocket-edge.t
t/http2/09-cli.t
t/http2/09-multiworker.t
t/http2/10-h2c.t
t/http2/11-streaming.t
t/http2/12-error-handling.t
t/http2/13-sse-detection.t
t/http2/14-sse-events.t
t/http2/15-sse-keepalive.t
t/http2/16-sse-cleanup.t
t/http2/17-h2-ws-sse-no-connection-state.t
t/http2/18-transport-leak.t
t/http2/19-transport-callbacks.t
t/http2/20-sse-transport.t
t/http2/21-http-date-header.t
# HTTP/2 over TLS (experimental)
pagi-server --http2 --ssl-cert cert.pem --ssl-key key.pem ./app.pl
```
Run `perldoc pagi-server` for the full list of options (workers, timeouts,
limits, watermarks, TLS, listeners, and more).
## Examples
The [`examples/`](examples/) directory contains progressively more advanced,
runnable applications — minimal HTTP, streaming with disconnect handling,
request-body draining, a WebSocket echo server, an SSE broadcaster,
lifespan/shared-state, extension-aware streaming, TLS introspection, a job
runner, UTF-8 handling, and a backpressure test harness. Each has its own
`README.md`. Start with [`examples/01-hello-http`](examples/01-hello-http/).
## Documentation
- [`PAGI::Server`](lib/PAGI/Server.pm) — the server class, constructor options,
and operational notes (`perldoc PAGI::Server`).
- [`pagi-server`](bin/pagi-server) — the command-line launcher
(`perldoc pagi-server`).
- [`PAGI::Server::Runner`](lib/PAGI/Server/Runner.pm) — application loading and
examples/02-streaming-response/README.md view on Meta::CPAN
# 02 – Streaming Response with Disconnect Handling
Shows how to:
- Drain the incoming `http.request` body (if any) before replying.
- Send multiple `http.response.body` chunks with `more => 1`.
- Emit `http.response.trailers` when `trailers => 1` was advertised.
- Watch for `{ type => 'http.disconnect' }` while streaming and stop if the client drops.
## Quick Start
**1. Start the server:**
```bash
pagi-server --app examples/02-streaming-response/app.pl --port 5000
```
**2. Demo with curl:**
```bash
# Watch chunks stream in (one per second)
curl -N http://localhost:5000/
# => Chunk 1
# => Chunk 2
# => Chunk 3
# Test disconnect handling - press Ctrl+C during streaming
curl -N http://localhost:5000/
# (press Ctrl+C to see server handle disconnect)
```
## Spec References
Covered by the PAGI specification in the upstream PAGI distribution
(`PAGI::Spec` POD and protocol documents, https://github.com/jjn1056/pagi):
- HTTP events, trailers, disconnect
examples/07-extension-fullflush/README.md view on Meta::CPAN
# 07 – Extension-Aware Streaming with FullFlush
Demonstrates how to:
- Check for extension support via `scope->{extensions}{fullflush}`
- Use `http.fullflush` event during streaming to force immediate TCP buffer flush
- Only send extension events when the server advertises support
The fullflush extension is useful for real-time streaming scenarios where you want each chunk delivered to the client immediately rather than waiting for TCP buffer fill or Nagle's algorithm.
## Quick Start
**1. Start the server:**
```bash
pagi-server --app examples/07-extension-fullflush/app.pl --port 5000
```
**2. Demo with curl:**
```bash
# Watch real-time streaming with immediate flush
curl -N http://localhost:5000/
# => Line 1 (flushed immediately)
# => Line 2 (flushed immediately)
# => Line 3 (flushed immediately)
# Each chunk appears instantly rather than being buffered
```
**Note:** The difference from regular streaming is most noticeable with small chunks that would normally be buffered by TCP.
## Spec References
Covered by the PAGI specification in the upstream PAGI distribution
(`PAGI::Spec` POD and protocol documents, https://github.com/jjn1056/pagi):
- Extensions section
- Fullflush extension
examples/07-extension-fullflush/app.pl view on Meta::CPAN
use strict;
use warnings;
use Future::AsyncAwait;
# Demonstrates fullflush extension during streaming response.
# The fullflush event forces immediate TCP buffer flush, useful for
# Server-Sent Events or real-time streaming where latency matters.
async sub app {
my ($scope, $receive, $send) = @_;
die "Unsupported scope type: $scope->{type}" if $scope->{type} ne 'http';
# Drain request body if present
while (1) {
my $event = await $receive->();
last if $event->{type} ne 'http.request';
examples/11-job-runner/README.md view on Meta::CPAN
```bash
perl -Ilib -Iexamples/11-job-runner/lib bin/pagi-server \
--app examples/11-job-runner/app.pl --port 5001
```
Then open http://localhost:5001 in your browser.
## Features
- **Real-time job queue** - Create countdown jobs and watch them execute
- **Live progress streaming** - SSE updates show second-by-second progress
- **WebSocket dashboard** - Queue-wide updates pushed to all connected clients
- **Concurrent execution** - Worker processes up to 3 jobs simultaneously
## Architecture
```
┌─────────────────────────────────────────────────────────────â”
│ Browser (app.js) │
│ ┌──────────────┠┌──────────────┠┌──────────────────┠│
│ │ WebSocket │ │ SSE │ │ HTTP (REST) │ │
examples/11-job-runner/lib/JobRunner/SSE.pm view on Meta::CPAN
data => $JSON->encode($data),
});
}
1;
__END__
# NAME
JobRunner::SSE - Server-Sent Events for job progress streaming
# DESCRIPTION
Provides real-time progress streaming for individual jobs via SSE.
## Endpoint
GET /api/jobs/:id/progress
## Events
- **status** - Initial job status on connection
- **progress** - Progress update { percent, message }
- **complete** - Job completed successfully { status, result, duration }
examples/README.md view on Meta::CPAN
middleware if the PAGI-Tools distribution happens to be installed, and
silently skips it otherwise.)
Note: Some advanced examples (job-runner, chat) use `IO::Async` directly for
timer and subprocess features. These are PAGI::Server-specific patterns.
Examples assume you understand the core PAGI specification (see the `PAGI::Spec` POD from the `PAGI` distribution on CPAN, https://github.com/jjn1056/pagi) plus the relevant protocol documents.
## Example List
1. `01-hello-http` - minimal HTTP response
2. `02-streaming-response` - chunked body, trailers, disconnect handling
3. `03-request-body` - reads multi-event request bodies
4. `04-websocket-echo` - handshake and echo loop
5. `05-sse-broadcaster` - server-sent events
6. `06-lifespan-state` - lifespan protocol with shared state
7. `07-extension-fullflush` - middleware using the `fullflush` extension
8. `08-tls-introspection` - prints TLS metadata when present
9. `11-job-runner` - background job processing example
10. `12-utf8` - UTF-8 handling demonstration
11. `13-custom-logging` - sending the server's diagnostics somewhere else (needs `Log::Dispatch`)
lib/PAGI/Server.pm view on Meta::CPAN
=head2 effective_max_connections
my $max = $server->effective_max_connections;
Returns the effective maximum connections limit. If C<max_connections>
was set explicitly, returns that value. Otherwise returns the default
of 1000.
=head1 FILE RESPONSE STREAMING
PAGI::Server supports efficient file streaming via the C<file> and C<fh>
keys in C<http.response.body> events:
# Stream entire file
await $send->({
type => 'http.response.body',
file => '/path/to/file.mp4',
more => 0,
});
# Stream partial file (for Range requests)
lib/PAGI/Server/AsyncFile.pm view on Meta::CPAN
use PAGI::Server::AsyncFile;
use IO::Async::Loop;
# Create or obtain an IO::Async::Loop
my $loop = IO::Async::Loop->new;
# Read entire file
my $content = await PAGI::Server::AsyncFile->read_file($loop, '/path/to/file');
# Read file in chunks (streaming)
await PAGI::Server::AsyncFile->read_file_chunked($loop, '/path/to/file', async sub {
my ($chunk) = @_;
# Process each chunk
}, chunk_size => 65536);
# Write file
await PAGI::Server::AsyncFile->write_file($loop, '/path/to/file', $content);
# Append to file
await PAGI::Server::AsyncFile->append_file($loop, '/path/to/file', $log_line);
=head1 DESCRIPTION
This module provides non-blocking file I/O operations using L<IO::Async::Function>
worker processes. It is used internally by L<PAGI::Server> for efficient file
streaming.
B<Note:> This is a PAGI::Server internal module. PAGI applications are
loop-agnostic and should use synchronous file I/O (which is simple and fast
for typical file sizes) or bring their own async file library if needed.
It uses L<IO::Async::Function> to offload blocking file operations to worker
processes, preventing the main event loop from being blocked during disk I/O.
Regular file I/O in POSIX is always blocking at the kernel level - even
C<select()>/C<poll()>/C<epoll()> report regular files as always "ready".
This module works around this limitation by running file operations in
lib/PAGI/Server/AsyncFile.pm view on Meta::CPAN
# Process chunk
}, chunk_size => 65536);
# For Range requests (partial file):
await PAGI::Server::AsyncFile->read_file_chunked($loop, $path, $callback,
offset => 1000, # Start at byte 1000
length => 5000, # Read 5000 bytes total
);
Read a file in chunks, calling a callback for each chunk. This is suitable
for streaming large files without loading the entire file into memory.
Parameters:
=over 4
=item * C<$loop> - IO::Async::Loop instance
=item * C<$path> - Path to the file to read
=item * C<$callback> - Async callback called with each chunk. Receives the chunk data.
lib/PAGI/Server/Compliance.pod view on Meta::CPAN
=head1 PAGI SPECIFICATION SUPPORT
Beyond protocol conformance, PAGI::Server implements the optional capabilities
defined in L<PAGI::Spec::Www>, across both HTTP/1.1 and HTTP/2:
=over 4
=item * B<Transport flow control> (C<pagi.transport>) -- C<buffered_amount>,
high/low watermarks, and C<on_high_water>/C<on_drain> backpressure callbacks, on
both HTTP/1.1 and HTTP/2, for every streaming scope type PAGI offers it on:
http, sse, and websocket (see L</"Transport Flow Control (pagi.transport)">).
=item * B<Connection state> (C<pagi.connection>) for HTTP scopes --
C<is_connected>, C<disconnect_reason>, C<on_disconnect> (abnormal only),
C<on_complete> (success only), and C<disconnect_future>. Tracked
independently per stream on HTTP/2 (see L</"Connection State
(pagi.connection)">).
=item * B<WebSocket Denial Response> (the C<websocket.http.response> extension)
-- reject a handshake with a custom HTTP response instead of a bare C<403>, on
lib/PAGI/Server/Compliance.pod view on Meta::CPAN
=item * Server SETTINGS: configurable max_concurrent_streams, initial_window_size,
max_frame_size, max_header_list_size
=item * A request HEADERS block exceeding max_header_list_size gets a real
431 response (RFC 9113 section 10.5.1) instead of a bare RST_STREAM,
matching HTTP/1.1
=item * HEAD request body suppression, matching HTTP/1.1: DATA frames are
withheld and file/fh bodies are never opened
=item * File and filehandle body streaming through the per-stream send queue,
under the same per-stream backpressure watermark as chunked bodies
=item * C<http.fullflush> on HTTP and SSE streams
=back
Filehandle (C<fh>) response bodies are read B<synchronously>, in fixed-size
chunks inside the connection's own send loop, on both HTTP/1.1 and HTTP/2 --
the handle is application-owned and cannot be handed to the async worker
pool across a fork boundary. This differs from C<file> (path) bodies opened
lib/PAGI/Server/Compliance.pod view on Meta::CPAN
=head2 Transport Flow Control (pagi.transport)
PAGI exposes outbound flow-control introspection to applications through the
C<pagi.transport> scope key (see L<PAGI::Spec::Www/"Transport Flow Control">):
C<buffered_amount>, the high/low watermarks, and the
C<on_high_water>/C<on_drain> backpressure callbacks. Over HTTP/2 the handle
measures the B<per-stream> send backlog, so each multiplexed stream is bounded
independently.
PAGI::Server provides C<pagi.transport> for every streaming scope type on both
transports: C<http> (streaming responses), C<sse>, and C<websocket>, on both
HTTP/1.1 and HTTP/2. The HTTP/2 WebSocket send path frames application
messages, protocol replies (pong, the close-frame echo), and its own keepalive
ping through the same per-stream send queue and pull-based data-provider model
that HTTP/2 streaming and SSE use. From the application's perspective the
transport is invisible: a WebSocket app sees the same C<pagi.transport>
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
lib/PAGI/Server/Compliance.pod view on Meta::CPAN
=head2 Response Headers
On C<sse.start> the server supplies C<Content-Type>, C<Cache-Control>, and
C<Date> only when the application did not already set them; an
application-supplied value for any of these is sent as-is, not duplicated
alongside a server default. C<Connection: keep-alive> on HTTP/1.1 is the
exception: it is a framing header the protocol requires the server to
control, so it is always advertised regardless of what the application
sent. HTTP/2 never emits an HTTP/1-only C<Connection> header or a
C<Transfer-Encoding> (chunked-framing) header -- an HTTP/2 SSE stream is
framed as DATA frames via C<submit_response_streaming>, which carries no
such headers.
The C<Date>-only-when-absent rule above holds server-wide, not only for
C<sse.start>: on every response path, the server supplies a plain C<Date>
header only when a C<Date> is not already present; where the application
supplies its own header list, an application-supplied C<Date> is always
honored as-is instead of being duplicated. This spans both transports and
every response shape the server can emit:
=over 4
lib/PAGI/Server/Connection.pm view on Meta::CPAN
use URI::Escape qw(uri_unescape);
use IO::Async::Timer::Countdown;
use IO::Async::Timer::Periodic;
use Time::HiRes qw(gettimeofday tv_interval);
use PAGI::Server::AsyncFile;
use PAGI::Server::ConnectionState;
use PAGI::Server::TransportState;
use PAGI::Server::EventValidator;
use constant FILE_CHUNK_SIZE => 65536; # 64KB chunks for file streaming
# Per-second cache for CLF timestamp in access log (same pattern as HTTP1::format_date)
my $_cached_log_timestamp;
my $_cached_log_time = 0;
# =============================================================================
# Header Validation (CRLF Injection Prevention)
# =============================================================================
# RFC 7230 Section 3.2.6: Field values MUST NOT contain CR or LF
lib/PAGI/Server/Connection.pm view on Meta::CPAN
my ($self, $stream_id, $stream_state) = @_;
weaken(my $weak_self = $self);
my $status;
my @response_headers;
# Streaming state for deferred data provider pattern.
# The send queue lives on per-stream state ($ss->{send_queue} /
# $ss->{send_queue_bytes}) so the h2 transport handle can measure it;
# $eof_pending / $streaming_started stay closure-local.
my $eof_pending = 0;
my $streaming_started = 0;
# Mirrors the stream state's own starting point (see the seq_state =>
# 'initial' initializer in _h2_on_request); the two must stay in step.
my $seq = 'initial';
my $is_head = (($stream_state->{pseudo}{':method'} // '') eq 'HEAD');
# Set once, from http.response.start's own 'trailers' flag (section 6
# below), and never changed again. $data_callback's no_end computation
# MUST key off this rather than $ss->{seq_state}: the trailers arm
# advances that mirror to 'complete' as soon as the app's send() call
# is made, which can be BEFORE the data provider has actually drained
# (deferred submit, see below) -- keying off seq_state there raced the
lib/PAGI/Server/Connection.pm view on Meta::CPAN
};
my $err = $@;
my $f = $ss2 && delete $ss2->{trailer_wait};
return unless $f;
$weak_self->{server}->loop->later(sub {
return if $f->is_ready;
if ($ok) { $f->done(1) } else { $f->fail($err) }
});
};
# Data callback for nghttp2's streaming response.
# Returns ($data, $eof) when data is available, or undef to defer.
my $data_callback = sub {
my ($cb_stream_id, $max_len) = @_;
my $ss = $weak_self && $weak_self->{h2_streams}{$stream_id};
return undef unless $ss;
my $q = $ss->{send_queue} ||= [];
if (@$q) {
my $chunk = shift @$q;
lib/PAGI/Server/Connection.pm view on Meta::CPAN
# Shared preamble for the file/fh body arms below (extraction, not a
# redesign -- see the arms themselves for why each piece sits where it
# does). Split in two because the two pieces cross the arms' own
# eval boundary: the file arm must report an illegal-sequence error
# (from advance_http) ahead of a misleading "File not found" from its
# own -f/-r checks, but must NOT have submitted response headers to the
# client before those checks pass -- so the sequence advance runs
# BEFORE the arm's eval (a comment-preserving-only move: advance_http
# is a pure function, and a caught vs. uncaught throw here already
# behaved identically before this extraction, since $seq is simply
# never reassigned on a throw either way), while the streaming-start
# call must stay INSIDE the eval, at each arm's own correct position
# relative to its own pre-checks, so a failure there still rolls back
# via the shared tail below instead of leaking a submitted-but-broken
# response.
# Outside-eval half: snapshot+advance+mirror the sequence state ahead
# of either arm's own validity checks. Returns the pre-advance $seq so
# the tail helper below can roll back to it on failure.
my $advance_seq_for_body = sub {
my ($ss, $event) = @_;
my $seq_before = $seq;
$seq = PAGI::Server::EventValidator::advance_http($seq, $event);
$ss->{seq_state} = $seq if $ss;
return $seq_before;
};
# Inside-eval half: submit the streaming response exactly once, the
# first time either arm actually has a chunk ready to send. Each arm
# calls this at its own correct point (after its own pre-checks, so a
# failed check never causes headers to reach the client for a response
# that goes on to fail).
my $ensure_h2_streaming_started = sub {
my ($ss) = @_;
return if $streaming_started;
$streaming_started = 1;
$ss->{send_queue} //= []; $ss->{send_queue_bytes} //= 0;
$weak_self->{h2_session}->submit_response_streaming(
$stream_id,
status => $status, headers => \@response_headers,
data_callback => $data_callback,
);
$weak_self->_h2_write_pending;
};
# Shared failure-rollback-or-finish tail for the file/fh body arms:
# given the arm's own eval result, either rolls $seq back to its
# pre-event value and re-raises (recoverable per contract, unless the
lib/PAGI/Server/Connection.pm view on Meta::CPAN
# invocation, and await that -- resolving this send() any
# earlier would let submit_trailer race ahead of not-yet-
# extracted DATA (see the closure-top comment). This PARKS
# the send() until the body actually finishes draining --
# e.g. until the peer grants enough flow-control window --
# which can be a real, unbounded wait from the app's point
# of view (closure-top comment has the full contract note).
$pending_trailer_headers = $trailer_headers;
my $f = $weak_self->{server}->loop->new_future;
$ss->{trailer_wait} = $f;
$weak_self->{h2_session}->resume_stream($stream_id) if $streaming_started;
$weak_self->_h2_write_pending;
# eval wraps the await (not just a plain assignment) so a
# failed $f (native rejection, via $deliver_trailer_eof's
# own $f->fail) is caught into $@ and folds into the same
# $ok/rollback handling the direct branch above uses.
$ok = eval { await $f; 1 };
# $f resolves DONE (not fail) on a whole-connection teardown
# too (_close's own sweep releases trailer_wait -- see
# _h2_resolve_stream_trailer_wait) -- the h2_closed carve-
# out's "successful no-op" contract, same discipline as the
lib/PAGI/Server/Connection.pm view on Meta::CPAN
# A gone stream here is the h2_closed carve-out's ordinary
# "successful no-op" case: the trailers already landed (ok=1)
# before the stream went away, so returning quietly is correct,
# not a swallowed error.
$ss = $weak_self ? $weak_self->{h2_streams}{$stream_id} : undef;
return if !$ss || $ss->{h2_closed};
# Flush any still-queued DATA before the trailing HEADERS --
# nghttp2 orders trailers after queued data on its own, but the
# stream must be resumed if the data callback had deferred.
$weak_self->{h2_session}->resume_stream($stream_id) if $streaming_started;
$weak_self->_h2_write_pending;
return;
}
# 4. file body: streamed through the send queue via $emit_chunk, one
# chunk at a time, under this stream's own backpressure. The type
# guard (not just "defined $event->{file}") keeps a nonconforming
# http.response.start carrying a stray 'file' key from being
# misrouted into this arm.
if ($type eq 'http.response.body' && defined $event->{file}) {
lib/PAGI/Server/Connection.pm view on Meta::CPAN
# sequence state reports the SEQUENCE error, not a misleading
# "File not found" -- h1 parity. A failed file send must NOT
# mark the response complete, so on failure $seq rolls back to
# $seq_before exactly as the h1 twin in _create_send does.
my $seq_before = $advance_seq_for_body->($ss, $event);
my $ok = eval {
die "File not found: $file\n" unless -f $file;
die "Cannot read file: $file\n" unless -r $file;
$ensure_h2_streaming_started->($ss);
# Mirror h1's effective-length computation (_send_file_response)
# so the sync-vs-async choice below sees the same number h1
# would, including the Www.pod rule that an offset past EOF
# clamps to zero bytes rather than failing.
my $file_size = -s $file;
die "Cannot stat file $file: $!\n" unless defined $file_size;
my $effective_length = $length // ($file_size - $offset);
$effective_length = 0 if $effective_length < 0;
lib/PAGI/Server/Connection.pm view on Meta::CPAN
# application owns $fh (opened before the send, closed or not by the
# app afterward) -- the server never closes it. Mirrors the file arm
# above; the read loop is adapted from h1's _send_fh_response. The
# type guard mirrors the file arm's M1 fix above.
if ($type eq 'http.response.body' && defined $event->{fh}) {
my $fh = $event->{fh};
my $seq_before = $advance_seq_for_body->($ss, $event);
my $ok = eval {
$ensure_h2_streaming_started->($ss);
if (my $off = $event->{offset}) {
seek($fh, $off, 0) or die "Cannot seek: $!\n";
}
my $remaining = $event->{length};
while (1) {
my $to_read = FILE_CHUNK_SIZE;
if (defined $remaining) {
$to_read = $remaining if $remaining < $to_read;
last if $to_read <= 0;
}
lib/PAGI/Server/Connection.pm view on Meta::CPAN
# Server-supplied Date header (HTTP/1.1 parity) — add if the app didn't.
unless (grep { lc($_->[0]) eq 'date' } @response_headers) {
push @response_headers, ['date', $weak_self->{protocol}->format_date];
}
}
elsif ($type eq 'http.response.body') {
my $body = $event->{body} // '';
my $more = $event->{more} // 0;
if ($more) {
if (!$streaming_started) {
# First streaming chunk — submit with data callback
$streaming_started = 1;
$ss->{send_queue} //= [];
$ss->{send_queue_bytes} //= 0;
if (length $body) {
push @{$ss->{send_queue}}, $body;
$ss->{send_queue_bytes} += length $body;
}
# Synchronous: we're in the app's send path (not nghttp2's
# extract), so on_high_water can fire here to tell the app to
# pause its source.
$ss->{transport_state}->_check_watermarks if $ss->{transport_state};
$weak_self->{h2_session}->submit_response_streaming(
$stream_id,
status => $status,
headers => \@response_headers,
data_callback => $data_callback,
);
$weak_self->_h2_write_pending;
} else {
# Subsequent chunk — backpressure check then push and resume.
# Bound on THIS stream's send queue (per-stream), not the
# shared TCP buffer which is meaningless across multiplexed
lib/PAGI/Server/Connection.pm view on Meta::CPAN
if (length $body) {
push @{$ss->{send_queue}}, $body;
$ss->{send_queue_bytes} += length $body;
}
# Synchronous — app send path, not nghttp2 extract.
$ss->{transport_state}->_check_watermarks if $ss->{transport_state};
$weak_self->{h2_session}->resume_stream($stream_id);
$weak_self->_h2_write_pending;
}
} else {
if ($streaming_started) {
# Final chunk on an already-streaming response. Bound on THIS
# stream's send queue (per-stream), not the shared TCP buffer.
if (($ss->{send_queue_bytes} // 0) >= $weak_self->{write_high_watermark}) {
await $weak_self->_h2_wait_for_stream_drain($stream_id);
return unless $weak_self;
return if $weak_self->{closed};
return unless $weak_self->{h2_streams}{$stream_id};
}
$eof_pending = 1;
if (length $body) {
push @{$ss->{send_queue}}, $body;
lib/PAGI/Server/Connection.pm view on Meta::CPAN
}
# Synchronous — app send path, not nghttp2 extract.
$ss->{transport_state}->_check_watermarks if $ss->{transport_state};
$weak_self->{h2_session}->resume_stream($stream_id);
$weak_self->_h2_write_pending;
} elsif ($seq eq 'awaiting_trailers') {
# Trailers were declared and this is the (single-shot)
# terminal body event: a plain submit_response would set
# END_STREAM immediately, ending the stream before the
# trailers arrive -- design §8.3 forbids that. Route
# through the streaming path instead, mirroring the
# file/fh arms' $streaming_started idiom; the data
# callback above already reserves END_STREAM for the
# trailing HEADERS block once it observes
# 'awaiting_trailers'.
$streaming_started = 1;
$ss->{send_queue} //= [];
$ss->{send_queue_bytes} //= 0;
$eof_pending = 1;
if (length $body) {
push @{$ss->{send_queue}}, $body;
$ss->{send_queue_bytes} += length $body;
}
$ss->{transport_state}->_check_watermarks if $ss->{transport_state};
$weak_self->{h2_session}->submit_response_streaming(
$stream_id,
status => $status,
headers => \@response_headers,
data_callback => $data_callback,
);
$weak_self->_h2_write_pending;
} else {
# Non-streaming: single response (unchanged one-shot path)
$weak_self->{h2_session}->submit_response($stream_id,
status => $status,
headers => \@response_headers,
body => $body,
);
$weak_self->_h2_write_pending;
}
}
}
elsif ($type eq 'http.fullflush') {
# Hand any pending frames to the session's write path (design §8.4).
$weak_self->{h2_session}->resume_stream($stream_id) if $streaming_started;
$weak_self->_h2_write_pending;
}
};
}
# =============================================================================
# HTTP/2 WebSocket over HTTP/2 (RFC 8441)
# =============================================================================
sub _h2_create_websocket_scope {
lib/PAGI/Server/Connection.pm view on Meta::CPAN
# max_frame_size: omitted when unenforced (max_ws_frame_size 0/undef
# means unlimited, per Protocol::WebSocket::Frame's max_payload_size
# semantics -- a server that does not enforce a cap must not
# advertise one). max_receive_queue has no unlimited mode (a hard,
# always-enforced cap), so it is always present.
($self->{max_ws_frame_size}
? (max_frame_size => $self->{max_ws_frame_size})
: ()
),
max_receive_queue => $self->{max_receive_queue},
# Per-stream outbound flow-control handle. Like the h2 sse/streaming
# scopes, it measures THIS stream's send queue (h2 multiplexes many
# streams over one connection, so the shared TCP buffer is
# meaningless per stream). Gives WebSocket-over-h2 the same
# pagi.transport surface HTTP/1.1 WebSocket already provides.
'pagi.transport' => ($stream_state->{transport_state} = $self->_h2_transport_state($stream_state)),
};
}
sub _h2_create_websocket_receive {
my ($self, $stream_id, $stream_state) = @_;
lib/PAGI/Server/Connection.pm view on Meta::CPAN
return $future;
};
}
sub _h2_create_websocket_send {
my ($self, $stream_id, $stream_state) = @_;
weaken(my $weak_self = $self);
my $seq = 'connecting';
# Data callback for nghttp2's streaming response (the same pull-based
# data-provider model _h2_create_send/_h2_create_sse_send already use).
# Pulls raw WS frame bytes from the per-stream queue -- app messages,
# protocol replies (pong, close-echo), and this stream's own keepalive
# ping are all pushed there in call order, so FIFO ordering on the wire
# is preserved exactly as it was under direct submit_data calls.
#
# $ss->{ws_eof_pending} lives on the stream state (not a closure-local,
# unlike the http streaming callback's $eof_pending) because it is set
# from other subs entirely -- _h2_ws_close and the close-frame arm of
# _h2_process_ws_frames -- not just from this closure. When set, it
# merges END_STREAM onto the LAST queued chunk (the close frame itself)
# rather than emitting a separate empty terminal frame, matching the
# previous submit_data($id, $close_frame_bytes, 1) behavior exactly.
my $data_callback = sub {
my ($cb_stream_id, $max_len) = @_;
my $ss = $weak_self && $weak_self->{h2_streams}{$stream_id};
return undef unless $ss;
lib/PAGI/Server/Connection.pm view on Meta::CPAN
# transfer-encoding, etc. before submission.
@headers = @{ $weak_self->_h2_strip_connection_headers(\@headers) };
$ss->{ws_accepted} = 1;
$ss->{response_started} = 1;
$ss->{ws_frame} = Protocol::WebSocket::Frame->new(
max_payload_size => $weak_self->{max_ws_frame_size},
);
# Submit 200 response with a pull-based data provider (same
# model as h2 streaming/SSE): frames are pushed onto
# $ss->{send_queue} and pulled by $data_callback as nghttp2's
# per-stream flow-control window allows.
$ss->{send_queue} //= [];
$ss->{send_queue_bytes} //= 0;
$weak_self->{h2_session}->submit_response_streaming($stream_id,
status => 200,
headers => \@headers,
data_callback => $data_callback,
);
$weak_self->_h2_write_pending;
# Process any data that arrived before accept
if (length($ss->{body}) > 0) {
my $buffered = $ss->{body};
$ss->{body} = '';
lib/PAGI/Server/Connection.pm view on Meta::CPAN
query_string => $query_string,
root_path => '',
headers => $headers,
(defined $self->{client_host}
? (client => [$self->{client_host}, $self->{client_port}])
: ()
),
server => [$self->{server_host}, $self->{server_port}],
state => keys %{$self->{state}} ? { %{$self->{state}} } : {},
extensions => $self->_get_extensions_for_scope,
# Per-stream outbound flow-control handle. Like the h2 streaming scope,
# it measures THIS stream's send queue (h2 multiplexes many streams over
# one connection, so the shared TCP buffer is meaningless per stream).
'pagi.transport' => ($stream_state->{transport_state} = $self->_h2_transport_state($stream_state)),
};
}
sub _h2_create_sse_receive {
my ($self, $stream_id, $stream_state) = @_;
weaken(my $weak_self = $self);
lib/PAGI/Server/Connection.pm view on Meta::CPAN
my $ss = $weak_self->{h2_streams}{$stream_id};
return $sse_disconnect->() unless $ss;
# Check queue first
if (@{$ss->{receive_queue}}) {
return shift @{$ss->{receive_queue}};
}
# First call returns sse.request once the full body has arrived.
# A POST body still streaming across DATA frames must not be
# truncated behind a truthful more=>0 -- wait for body_complete
# (set on END_STREAM or stream close, see _h2_on_body/_h2_on_close)
# before delivering it as a single terminal event. This is the
# smaller change relative to reworking sse.request into a
# truthful multi-chunk stream: the wire shape here was already
# one-shot, so completing that contract fixes the dispatch-timing
# bug (design section 11.1) without touching the event shape.
if (!$ss->{sse_request_sent}) {
while (!$ss->{body_complete}) {
if (@{$ss->{receive_queue}}) {
lib/PAGI/Server/Connection.pm view on Meta::CPAN
}
sub _h2_create_sse_send {
my ($self, $stream_id, $stream_state) = @_;
weaken(my $weak_self = $self);
my $seq = 'initial';
# Streaming state for the data-provider pull pattern. The send queue lives on
# per-stream state ($ss->{send_queue} / $ss->{send_queue_bytes}) so the
# pagi.transport handle can measure THIS stream's backlog. $streaming_started
# stays closure-local.
my $streaming_started = 0;
# Data callback for nghttp2's streaming response. Pulls from the per-stream
# queue; SSE responses stay open, so this never signals EOF (returns eof=0),
# or undef to defer when the queue is empty.
my $data_callback = sub {
my ($cb_stream_id, $max_len) = @_;
my $ss = $weak_self && $weak_self->{h2_streams}{$stream_id};
return undef unless $ss;
my $q = $ss->{send_queue} ||= [];
if (@$q) {
lib/PAGI/Server/Connection.pm view on Meta::CPAN
# supply them (design doc section 11.4).
unless (grep { lc($_->[0]) eq 'cache-control' } @final_headers) {
push @final_headers, ['cache-control', 'no-cache'];
}
# Server-supplied Date header (HTTP/1.1 parity) — the h1 SSE path adds
# this too; add it unless the app supplied one.
unless (grep { lc($_->[0]) eq 'date' } @final_headers) {
push @final_headers, ['date', $weak_self->{protocol}->format_date];
}
$streaming_started = 1;
$ss->{send_queue} //= [];
$ss->{send_queue_bytes} //= 0;
$weak_self->{h2_session}->submit_response_streaming(
$stream_id,
status => $status,
headers => \@final_headers,
data_callback => $data_callback,
);
$weak_self->_h2_write_pending;
# Protocol-specific keepalive writer (HTTP/2 DATA frames), scoped to
# THIS stream (design section 11.3): a second multiplexed SSE stream
# must not steal or replace this one's writer. Keepalive bytes are
lib/PAGI/Server/Connection.pm view on Meta::CPAN
}
$weak_self->{h2_session}->submit_response($stream_id,
status => $ss->{sse_decline_status},
headers => $ss->{sse_decline_headers},
body => $ss->{sse_decline_body},
);
$weak_self->_h2_write_pending;
}
elsif ($type eq 'http.fullflush') {
# Hand any pending frames to the session's write path (design §8.4).
$weak_self->{h2_session}->resume_stream($stream_id) if $streaming_started;
$weak_self->_h2_write_pending;
}
return;
};
}
sub _h2_process_ws_frames {
my ($self, $stream_id, $stream, $data) = @_;
lib/PAGI/Server/Connection.pm view on Meta::CPAN
=head2 How It Works
A request is detected as SSE when its combined C<Accept> header values
contain the exact media range C<text/event-stream>, case-insensitively,
with an effective quality value greater than zero (see L<PAGI::Spec::Www/
"SSE Connection Detection">); a C<q=0> refusal or a wildcard range such as
C<*/*> never signals SSE. Detection works identically regardless of HTTP
version. Over HTTP/1.1, SSE data is sent using chunked Transfer-Encoding.
Over HTTP/2, SSE data is sent as DATA frames via the
C<submit_response_streaming>/C<data_callback> mechanism. This difference is
transparent to the application.
The C<http_version> field in the scope hash will be C<'2'> for HTTP/2
connections, allowing applications to distinguish if needed.
=head2 SSE Idle Timeout over HTTP/2
The C<sse_idle_timeout> setting is enforced B<per stream> on HTTP/2: each
SSE stream owns its own idle timer, armed when that stream's
C<sse.start> is sent and reset by that stream's own send activity
lib/PAGI/Server/ConnectionState.pm view on Meta::CPAN
sub response_started { return $_[0]->{_response_started} ? 1 : 0 }
# Server-internal: called from the send path when http.response.start is emitted.
sub _mark_response_started { $_[0]->{_response_started} = 1; return }
=head2 response_complete
my $done = $conn->response_complete; # undef = unsupported, else 0 or 1
Returns true once this request's response body has been fully sent, false
while a response is still streaming or has not started, and C<undef> if the
server does not track completion. This server does not currently track
response-body completion, so this accessor always returns C<undef>. SHOULD-level
per L<PAGI::Spec::Www/"Connection State">; test C<defined> before relying on it.
=cut
sub response_complete { return undef }
=head2 disconnect_reason
lib/PAGI/Server/EventValidator.pm view on Meta::CPAN
my $type = $event->{type} // '';
if ($state eq 'closed') {
return 'closed' if $type eq 'sse.close';
croak "cannot send '$type' after sse.close";
}
croak "cannot send '$type': decline response already complete"
if $state eq 'decline_complete';
if ($state eq 'initial') {
return 'streaming' if $type eq 'sse.start';
return 'declining' if $type eq 'sse.http.response.start';
croak "cannot send '$type' before sse.start";
}
if ($state eq 'streaming') {
return 'streaming' if $type eq 'sse.send' || $type eq 'sse.comment' || $type eq 'sse.keepalive' || $type eq 'http.fullflush';
return 'closed' if $type eq 'sse.close';
croak "cannot decline with sse.http.response.start after sse.start"
if $type eq 'sse.http.response.start';
croak "cannot send duplicate sse.start"
if $type eq 'sse.start';
croak "cannot send '$type' after sse.start";
}
if ($state eq 'declining') {
if ($type eq 'sse.http.response.body') {
lib/PAGI/Server/EventValidator.pm view on Meta::CPAN
(C<"cannot send '<type>' after websocket.http.response.start">); any event
once C<closed> (C<"cannot send '<type>' after websocket.close">, including a
second C<websocket.close> - unlike SSE, WebSocket close is not idempotent);
any event once C<denial_complete>
(C<"cannot send '<type>': denial response already complete">).
=head2 advance_sse($state, $event)
Pure send-sequence transition function for the SSE family. Does not
validate event shape; call C<validate_sse_send> separately first. States:
C<initial>, C<streaming>, C<declining>, C<decline_complete>, C<closed>.
Starting state is C<initial>.
From C<initial>: C<sse.start> advances to C<streaming>;
C<sse.http.response.start> (a decline) advances to C<declining>. From
C<streaming>: C<sse.send>, C<sse.comment>, C<sse.keepalive>, and
C<http.fullflush> keep C<streaming>; C<sse.close> advances to C<closed>.
From C<declining>:
C<sse.http.response.body> with a true C<more> field keeps C<declining>; a
terminal body chunk advances to C<decline_complete>. From C<closed>,
C<sse.close> is idempotent and stays C<closed>.
Croaks: any event in C<initial> other than C<sse.start>/decline start
(C<"cannot send '<type>' before sse.start">); a decline start after
C<sse.start> (C<"cannot decline with sse.http.response.start after sse.start">);
a duplicate C<sse.start> (C<"cannot send duplicate sse.start">); any other
event once C<streaming> (C<"cannot send '<type>' after sse.start">); any
non-body event once declining has started
(C<"cannot send '<type>' after sse.http.response.start">); any event other
than C<sse.close> once C<closed> (C<"cannot send '<type>' after sse.close">);
any event once C<decline_complete>
(C<"cannot send '<type>': decline response already complete">).
=head2 advance_lifespan($state, $event)
Pure send-sequence transition function for the lifespan family. Does not
validate event shape; call C<validate_lifespan_send> separately first.
lib/PAGI/Server/Protocol/HTTP2.pm view on Meta::CPAN
=head2 submit_response
$session->submit_response($stream_id,
status => 200,
headers => [['content-type', 'text/html']],
body => $body,
);
Submit a response on a stream. C<body> can be a string (sent as single
response) or a coderef for streaming.
=cut
sub submit_response {
my ($self, $stream_id, %args) = @_;
return $self->{nghttp2}->submit_response($stream_id, %args);
}
=head2 submit_response_streaming
$session->submit_response_streaming($stream_id,
status => 200,
headers => [['content-type', 'text/event-stream']],
data_callback => sub {
my ($stream_id, $max_len) = @_;
return ($chunk, $is_eof);
},
);
Submit a streaming response with a data provider callback.
=cut
sub submit_response_streaming {
my ($self, $stream_id, %args) = @_;
return $self->{nghttp2}->submit_response($stream_id,
status => $args{status},
headers => $args{headers},
data_callback => $args{data_callback},
callback_data => $args{callback_data},
);
}
=head2 resume_stream
lib/PAGI/Server/Protocol/HTTP2.pm view on Meta::CPAN
Key differences that affect PAGI integration:
=over 4
=item * Multiplexing - Multiple concurrent requests on one TCP connection
=item * Binary Framing - nghttp2 handles all framing; PAGI feeds/extracts bytes
=item * Header Compression - HPACK is built into nghttp2
=item * Flow Control - Per-stream and connection-level, via streaming callbacks
=back
=head1 SEE ALSO
L<Net::HTTP2::nghttp2>, L<PAGI::Server::Protocol::HTTP1>
=cut
t/02-streaming.t view on Meta::CPAN
use Test2::V0;
use IO::Async::Loop;
use Net::Async::HTTP;
use Future::AsyncAwait;
use PAGI::Server;
plan skip_all => "Server integration tests not supported on Windows" if $^O eq 'MSWin32';
# Step 2: Streaming Responses and Disconnect Handling
# Tests for examples/02-streaming-response/app.pl
my $loop = IO::Async::Loop->new;
# Test 1: Streaming response with multiple chunks
subtest 'Streaming response uses chunked Transfer-Encoding' => sub {
my $simple_streaming_app = async sub {
my ($scope, $receive, $send) = @_;
die "Unsupported: $scope->{type}" if $scope->{type} ne 'http';
while (1) {
my $event = await $receive->();
last if $event->{type} ne 'http.request';
last unless $event->{more};
}
await $send->({
t/02-streaming.t view on Meta::CPAN
status => 200,
headers => [['content-type', 'text/plain']],
});
await $send->({ type => 'http.response.body', body => "Chunk 1\n", more => 1 });
await $send->({ type => 'http.response.body', body => "Chunk 2\n", more => 1 });
await $send->({ type => 'http.response.body', body => "Chunk 3\n", more => 0 });
};
my $server = PAGI::Server->new(
app => $simple_streaming_app,
host => '127.0.0.1',
port => 0,
quiet => 1,
);
$loop->add($server);
$server->listen->get;
my $port = $server->port;
t/02-streaming.t view on Meta::CPAN
my $body = $response->decoded_content;
like($body, qr/Chunk 1.*Chunk 2.*Chunk 3/s, 'Response body contains all chunks in order');
$server->shutdown->get;
$loop->remove($server);
};
# Test 2: Multiple body chunks arrive in order
subtest 'Multiple http.response.body events work correctly' => sub {
my $streaming_app = async sub {
my ($scope, $receive, $send) = @_;
die "Unsupported: $scope->{type}" if $scope->{type} ne 'http';
while (1) {
my $event = await $receive->();
last if $event->{type} ne 'http.request';
last unless $event->{more};
}
await $send->({
t/02-streaming.t view on Meta::CPAN
status => 200,
headers => [['content-type', 'text/plain']],
});
await $send->({ type => 'http.response.body', body => "First\n", more => 1 });
await $send->({ type => 'http.response.body', body => "Second\n", more => 1 });
await $send->({ type => 'http.response.body', body => "Third\n", more => 0 });
};
my $server = PAGI::Server->new(
app => $streaming_app,
host => '127.0.0.1',
port => 0,
quiet => 1,
);
$loop->add($server);
$server->listen->get;
my $port = $server->port;
t/02-streaming.t view on Meta::CPAN
is($response->code, 200, 'Response status is 200');
is($response->header('Transfer-Encoding'), 'chunked', 'Response uses chunked encoding');
my $body = $response->decoded_content;
like($body, qr/Body content/, 'Body content is correct');
$server->shutdown->get;
$loop->remove($server);
};
# Test 5: Client disconnect detection
subtest 'Client disconnect stops streaming app' => sub {
my $disconnect_detected = 0;
my $chunks_sent = 0;
my $slow_streaming_app = async sub {
my ($scope, $receive, $send) = @_;
die "Unsupported: $scope->{type}" if $scope->{type} ne 'http';
my $loop = IO::Async::Loop->new;
# Drain request body
while (1) {
my $event = await $receive->();
last if $event->{type} ne 'http.request';
last unless $event->{more};
t/02-streaming.t view on Meta::CPAN
await $send->({ type => 'http.response.body', body => "Chunk $i\n", more => ($i < 5) ? 1 : 0 });
# Wait between chunks (allows disconnect to be detected)
if ($i < 5 && $loop) {
await $loop->delay_future(after => 0.2);
}
}
};
my $server = PAGI::Server->new(
app => $slow_streaming_app,
host => '127.0.0.1',
port => 0,
quiet => 1,
);
$loop->add($server);
$server->listen->get;
my $port = $server->port;
t/02-streaming.t view on Meta::CPAN
ok($sock, 'Connected to server');
# Send HTTP request
print $sock "GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n";
# Read just the headers and first chunk
my $data = '';
$sock->blocking(0);
# Give server time to start streaming
my $timeout = time + 2;
while (time < $timeout) {
$loop->loop_once(0.1);
my $buf;
my $n = $sock->sysread($buf, 4096);
if (defined $n && $n > 0) {
$data .= $buf;
# Once we have some body data, disconnect
if ($data =~ /Chunk 1/) {
last;
t/10-http-compliance.t view on Meta::CPAN
})->()->get;
};
# Test 18: send() after the response is complete raises, per mandatory
# sequencing (PAGI spec compliance) -- not the genuine post-disconnect no-op,
# which lives in t/52-mandatory-validation.t's /post-close case.
subtest 'send() after response completion raises "response already complete"' => sub {
# Exercise the _create_send implementation directly by verifying that a
# send after the response has completed fails the returned Future.
# First verify by examining the send() return behavior in a streaming test
# where the client disconnects mid-stream
my $app_completed = 0;
my $error_during_send = 0;
my $test_app = async sub {
my ($scope, $receive, $send) = @_;
# Handle lifespan scope
if ($scope->{type} eq 'lifespan') {
while (1) {
my $event = await $receive->();
t/40-event-validation.t view on Meta::CPAN
};
# =============================================================================
# Sequence State Machines
# =============================================================================
subtest 'advance_http transition matrix' => sub {
my $adv = \&PAGI::Server::EventValidator::advance_http;
is( $adv->('initial', { type => 'http.response.start', status => 200 }), 'started', 'start -> started');
is( $adv->('initial', { type => 'http.response.start', status => 200, trailers => 1 }), 'started_t', 'start+trailers -> started_t');
is( $adv->('started', { type => 'http.response.body', body => 'x', more => 1 }), 'started_i', 'streaming chunk marks inline bytes delivered');
is( $adv->('started', { type => 'http.response.body', body => 'x' }), 'complete', 'terminal body -> complete');
is( $adv->('started', { type => 'http.response.body', file => '/tmp/f' }), 'complete', 'file body -> complete');
is( $adv->('started', { type => 'http.response.body', fh => \*STDOUT, more => 1 }), 'complete', 'fh body is always terminal regardless of more');
is( $adv->('started_t', { type => 'http.response.body', body => 'x', more => 1 }), 'started_t_i', 'streaming chunk marks inline bytes delivered, trailers still declared');
is( $adv->('started_t', { type => 'http.response.body', body => 'x', more => 0 }), 'awaiting_trailers', 'terminal body with declared trailers -> awaiting_trailers');
is( $adv->('awaiting_trailers', { type => 'http.response.trailers', headers => [] }), 'complete', 'trailers -> complete');
is( $adv->('started', { type => 'http.fullflush' }), 'started', 'fullflush leaves started unchanged');
is( $adv->('started_t', { type => 'http.fullflush' }), 'started_t', 'fullflush leaves started_t unchanged');
is( $adv->('awaiting_trailers', { type => 'http.fullflush' }), 'awaiting_trailers', 'fullflush leaves awaiting_trailers unchanged');
like( dies { $adv->('initial', { type => 'http.response.body', body => 'x' }) }, qr/before http\.response\.start/, 'body before start');
like( dies { $adv->('initial', { type => 'http.response.trailers' }) }, qr/before http\.response\.start/, 'trailers before start');
like( dies { $adv->('started', { type => 'http.response.start', status => 200 }) }, qr/duplicate http\.response\.start/, 'duplicate start');
like( dies { $adv->('started_t', { type => 'http.response.start', status => 200 }) }, qr/duplicate http\.response\.start/, 'duplicate start after trailers declared');
like( dies { $adv->('started', { type => 'http.response.trailers' }) }, qr/not declared/, 'undeclared trailers');
t/40-event-validation.t view on Meta::CPAN
is( $adv->('started_i', { type => 'http.response.body', body => 'x', more => 0 }),
'complete', 'and terminate normally');
is( $adv->('started_t_i', { type => 'http.response.body', body => 'x', more => 0 }),
'awaiting_trailers', 'reaching the trailers phase as before');
is( $adv->('started_i', { type => 'http.fullflush' }), 'started_i',
'fullflush leaves the inline marker alone');
};
subtest 'advance_sse close is idempotent, streams stay exclusive' => sub {
my $adv = \&PAGI::Server::EventValidator::advance_sse;
is( $adv->('initial', { type => 'sse.start' }), 'streaming', 'start -> streaming');
is( $adv->('streaming', { type => 'sse.send', data => 'x' }), 'streaming', 'send keeps streaming');
is( $adv->('streaming', { type => 'sse.comment', comment => 'x' }), 'streaming', 'comment keeps streaming');
is( $adv->('streaming', { type => 'sse.keepalive', interval => 15 }), 'streaming', 'keepalive keeps streaming');
is( $adv->('streaming', { type => 'http.fullflush' }), 'streaming', 'fullflush leaves streaming unchanged');
like( dies { $adv->('initial', { type => 'http.fullflush' }) }, qr/before sse\.start/, 'fullflush before start');
is( $adv->('streaming', { type => 'sse.close' }), 'closed', 'close -> closed');
is( $adv->('closed', { type => 'sse.close' }), 'closed', 'second close idempotent');
is( $adv->('declining', { type => 'sse.http.response.body', more => 1 }), 'declining', 'decline body chunk keeps declining');
is( $adv->('declining', { type => 'sse.http.response.body' }), 'decline_complete', 'terminal decline body -> decline_complete');
like( dies { $adv->('closed', { type => 'sse.send', data => 'x' }) }, qr/after sse\.close/, 'send after close');
is( $adv->('initial', { type => 'sse.http.response.start', status => 404 }), 'declining', 'decline start');
like( dies { $adv->('initial', { type => 'sse.send', data => 'x' }) }, qr/before sse\.start/, 'send before start');
like( dies { $adv->('streaming', { type => 'sse.http.response.start', status => 404 }) }, qr/after sse\.start/, 'decline after start');
like( dies { $adv->('streaming', { type => 'sse.start' }) }, qr/duplicate sse\.start/, 'duplicate start');
like( dies { $adv->('streaming', { type => 'sse.http.response.body', body => 'x' }) }, qr/after sse\.start/, 'decline body while streaming croaks');
like( dies { $adv->('declining', { type => 'sse.send', data => 'x' }) }, qr/after sse\.http\.response\.start/, 'stream event while declining');
like( dies { $adv->('decline_complete', { type => 'sse.close' }) }, qr/decline response already complete/, 'anything after decline complete');
};
subtest 'advance_websocket denial and accept are exclusive' => sub {
my $adv = \&PAGI::Server::EventValidator::advance_websocket;
is( $adv->('connecting', { type => 'websocket.accept' }), 'accepted', 'accept');
is( $adv->('connecting', { type => 'websocket.http.response.start', status => 401 }), 'denial', 'denial start');
is( $adv->('connecting', { type => 'websocket.close' }), 'closed', 'close while connecting');
is( $adv->('accepted', { type => 'websocket.send', text => 'x' }), 'accepted', 'send keeps accepted');
t/53-trailers-framing.t view on Meta::CPAN
is( $T::CL_DISC_REASON, 'server_error',
'/cl-trailers: app-return after the failed trailers send reports on_disconnect(server_error)' );
ok(
(scalar grep { /returned with an incomplete response/i } @cl_warnings),
'/cl-trailers: incomplete-response warning logged'
);
# --- /chunked-trailers: control -- raw socket, trailers actually on the wire
# Net::Async::HTTP doesn't expose trailers (see t/02-streaming.t), so read
# the raw chunked framing off the wire ourselves. Socket-pump idiom lifted
# from t/52-mandatory-validation.t's $ws_handshake_and_drain; the chunk
# grammar being asserted against (hex chunk-size line, CRLF-terminated
# chunks, "0\r\n" + trailer headers + CRLF as the terminator) is the same
# grammar t/16-chunked-validation.t exercises against the parser.
my $raw_request = sub {
my ($port, $method, $path) = @_;
my $sock = IO::Socket::INET->new(
PeerAddr => '127.0.0.1',
PeerPort => $port,
t/http2/07-websocket.t view on Meta::CPAN
method => 'CONNECT',
path => '/ws/chat',
scheme => 'https',
authority => 'localhost',
headers => [
[':protocol', 'websocket'],
['sec-websocket-version', '13'],
['sec-websocket-protocol', 'chat, superchat'],
['origin', 'https://localhost'],
],
body => sub { return undef }, # streaming: keep open
);
$client_sock->syswrite($client->mem_send);
exchange_frames($client, $client_sock);
ok(scalar @scopes >= 1, 'App was called');
if (@scopes) {
my $scope = $scopes[0];
is($scope->{type}, 'websocket', 'scope type is websocket');
t/http2/07-websocket.t view on Meta::CPAN
# Send Extended CONNECT
my $ws_stream_id = $client->submit_request(
method => 'CONNECT',
path => '/ws/echo',
scheme => 'https',
authority => 'localhost',
headers => [
[':protocol', 'websocket'],
['sec-websocket-version', '13'],
],
body => sub { return undef }, # streaming: keep open
);
$client_sock->syswrite($client->mem_send);
# Wait for 200 response
exchange_frames($client, $client_sock);
is($response_headers{':status'}, '200', 'Got 200 for WebSocket accept');
# Send a WebSocket text frame via HTTP/2 DATA
my $ws_frame = Protocol::WebSocket::Frame->new(
buffer => 'Hello WebSocket',
t/http2/07-websocket.t view on Meta::CPAN
# Open WebSocket
my $ws_stream_id = $client->submit_request(
method => 'CONNECT',
path => '/ws/close-test',
scheme => 'https',
authority => 'localhost',
headers => [
[':protocol', 'websocket'],
['sec-websocket-version', '13'],
],
body => sub { return undef }, # streaming: keep open
);
$client_sock->syswrite($client->mem_send);
exchange_frames($client, $client_sock);
is($response_headers{':status'}, '200', 'WebSocket accepted');
# Send close frame (code=1000, reason="normal closure")
my $close_frame = Protocol::WebSocket::Frame->new(
type => 'close',
buffer => pack('n', 1000) . 'normal closure',
t/http2/11-streaming.t view on Meta::CPAN
plan skip_all => "Server integration tests not supported on Windows" if $^O eq 'MSWin32';
BEGIN {
require PAGI::Server::Protocol::HTTP2;
PAGI::Server::Protocol::HTTP2->available
or plan(skip_all => 'HTTP/2 not available (Net::HTTP2::nghttp2 0.008+ required)');
}
# ============================================================
# Test: HTTP/2 Streaming Responses
# ============================================================
# Verifies that HTTP/2 streaming responses (more => 1) send
# DATA frames incrementally rather than accumulating in memory.
use PAGI::Server::Connection;
use PAGI::Server;
use PAGI::Server::Protocol::HTTP1;
use PAGI::Server::Protocol::HTTP2;
my $loop = IO::Async::Loop->new;
my $protocol = PAGI::Server::Protocol::HTTP1->new;
t/http2/11-streaming.t view on Meta::CPAN
$loop->loop_once(0.1);
my $buf = '';
$client_sock->sysread($buf, 16384);
$client->mem_recv($buf) if length($buf);
my $out = $client->mem_send;
$client_sock->syswrite($out) if length($out);
}
}
# ============================================================
# Basic streaming: 3 chunks + final
# ============================================================
subtest 'basic streaming response delivers all data' => sub {
my $app = async sub {
my ($scope, $receive, $send) = @_;
await $receive->();
await $send->({
type => 'http.response.start',
status => 200,
headers => [['content-type', 'text/plain']],
});
for my $i (1..3) {
await $send->({
t/http2/11-streaming.t view on Meta::CPAN
on_stream_close => sub {
$stream_closed = 1;
return 0;
},
);
h2c_handshake($client, $client_sock);
$client->submit_request(
method => 'GET',
path => '/streaming',
scheme => 'http',
authority => 'localhost',
);
$client_sock->syswrite($client->mem_send);
exchange_frames($client, $client_sock, 20);
is($response_body, 'chunk1chunk2chunk3final', 'All streaming chunks received');
ok($stream_closed, 'Stream was closed');
$stream_io->close_now;
$loop->remove($server);
};
# ============================================================
# Incremental delivery: data arrives before more => 0
# ============================================================
# This is the KEY test that proves the bug. With the old code,
# data only arrives after more => 0 because everything is
# accumulated in $body_chunks. With the fix, data arrives
# incrementally.
subtest 'streaming data arrives incrementally (not buffered until EOF)' => sub {
# Use futures to coordinate: app waits for client to confirm
# receipt of each chunk before sending the next one.
my @chunk_received_at; # Track when each data callback fires
my $app = async sub {
my ($scope, $receive, $send) = @_;
await $receive->();
await $send->({
type => 'http.response.start',
status => 200,
t/http2/11-streaming.t view on Meta::CPAN
ok(scalar @data_events > 1,
'Data arrived in multiple chunks (not accumulated)')
or diag "Got " . scalar(@data_events) . " data events: " .
join(', ', map { "'$_'" } @data_events);
$stream_io->close_now;
$loop->remove($server);
};
# ============================================================
# Empty final chunk: streaming + empty more => 0
# ============================================================
subtest 'streaming with empty final chunk' => sub {
my $app = async sub {
my ($scope, $receive, $send) = @_;
await $receive->();
await $send->({
type => 'http.response.start',
status => 200,
headers => [['content-type', 'text/plain']],
});
await $send->({
type => 'http.response.body',
t/http2/11-streaming.t view on Meta::CPAN
$client->submit_request(
method => 'GET',
path => '/empty-final',
scheme => 'http',
authority => 'localhost',
);
$client_sock->syswrite($client->mem_send);
exchange_frames($client, $client_sock, 20);
is($response_body, 'only-chunk', 'Body is just the streaming chunk');
ok($stream_closed, 'Stream was closed');
$stream_io->close_now;
$loop->remove($server);
};
# ============================================================
# Backpressure: many large chunks don't cause unbounded growth
# ============================================================
subtest 'streaming with many chunks completes without accumulation' => sub {
my $chunk_count = 50;
my $chunk_size = 8192;
my $chunk_data = 'X' x $chunk_size;
my $app = async sub {
my ($scope, $receive, $send) = @_;
await $receive->();
await $send->({
type => 'http.response.start',
status => 200,
t/http2/11-streaming.t view on Meta::CPAN
on_stream_close => sub {
$stream_closed = 1;
return 0;
},
);
h2c_handshake($client, $client_sock);
$client->submit_request(
method => 'GET',
path => '/large-streaming',
scheme => 'http',
authority => 'localhost',
);
$client_sock->syswrite($client->mem_send);
# Many rounds needed: 50 x 8KB = 400KB, flow control window is 65535
exchange_frames($client, $client_sock, 100);
is($received_bytes, $chunk_count * $chunk_size,
'All streaming data received');
ok($stream_closed, 'Stream was closed');
$stream_io->close_now;
$loop->remove($server);
};
# ============================================================
# Single-shot (non-streaming): more => 0 as first body event
# ============================================================
subtest 'non-streaming response (more => 0 only) still works' => sub {
my $app = async sub {
my ($scope, $receive, $send) = @_;
await $receive->();
await $send->({
type => 'http.response.start',
status => 200,
headers => [['content-type', 'text/plain']],
});
await $send->({
type => 'http.response.body',
t/http2/11-streaming.t view on Meta::CPAN
is($response_headers{':status'}, '200', 'Got 200 status');
is($response_body, 'single-shot', 'Body received correctly');
ok($stream_closed, 'Stream was closed');
$stream_io->close_now;
$loop->remove($server);
};
# ============================================================
# Connection close during streaming: no crashes
# ============================================================
subtest 'connection close during streaming does not crash' => sub {
my $send_started = 0;
my $send_error = 0;
my $app = async sub {
my ($scope, $receive, $send) = @_;
await $receive->();
await $send->({
type => 'http.response.start',
status => 200,
headers => [['content-type', 'text/plain']],
t/http2/11-streaming.t view on Meta::CPAN
# Close the client side to simulate disconnect
close($client_sock);
# Let the event loop process the disconnection
for (1..10) {
$loop->loop_once(0.1);
}
# If we got here without crashing, the test passes
pass('No crash on connection close during streaming');
$stream_io->close_now;
$loop->remove($server);
};
# ============================================================
# EOF race: empty final body with eof_pending
# ============================================================
# Regression: if data_callback is called when @data_queue is
# empty but $eof_pending is true, it should return ('', 1)
# to signal EOF, not undef (defer).
subtest 'EOF signaling when data_callback invoked after queue drained' => sub {
# App sends streaming chunks, then final empty body (more => 0).
# The key scenario: the sentinel empty string in @data_queue
# gets consumed, then data_callback is called again.
my $app = async sub {
my ($scope, $receive, $send) = @_;
await $receive->();
await $send->({
type => 'http.response.start',
status => 200,
headers => [['content-type', 'text/plain']],
});
t/http2/12-error-handling.t view on Meta::CPAN
use PAGI::Server::Connection;
use PAGI::Server;
use PAGI::Server::Protocol::HTTP1;
use PAGI::Server::Protocol::HTTP2;
my $loop = IO::Async::Loop->new;
my $protocol = PAGI::Server::Protocol::HTTP1->new;
# ============================================================
# Helpers (same pattern as 11-streaming.t)
# ============================================================
sub create_test_server {
my (%args) = @_;
my $server = PAGI::Server->new(
app => $args{app} // sub { },
host => '127.0.0.1',
port => 0,
quiet => 1,
http2 => 1,
t/http2/12-error-handling.t view on Meta::CPAN
return 0;
},
on_stream_close => sub {
$stream_closed = 1;
return 0;
},
);
h2c_handshake($client, $client_sock);
# Send POST with content-length > max_body_size using a streaming body
# The server should reject based on content-length header alone,
# before any body data arrives
$client->submit_request(
method => 'POST',
path => '/upload-cl',
scheme => 'http',
authority => 'localhost',
headers => [
['content-type', 'application/octet-stream'],
['content-length', '50000'],
],
body => sub { return undef }, # streaming: keep open
);
$client_sock->syswrite($client->mem_send);
exchange_frames($client, $client_sock, 20);
is($response_headers{':status'}, '413', 'Server responded with 413 based on content-length');
ok(!$app_called, 'App was never called');
$stream_io->close_now;
$loop->remove($server);
t/http2/12-error-handling.t view on Meta::CPAN
is($submit_response_calls, 1,
'only ONE submit_response reached nghttp2 for this stream -- the 413, not a second (200) on top of it')
or diag('submitted statuses: ' . join(', ', @submitted_statuses));
is($submitted_statuses[0], 413, 'the one submit_response that landed was the 413');
$stream_io->close_now;
$loop->remove($server);
};
# ============================================================
# RST_STREAM from client during streaming response
# ============================================================
subtest 'RST_STREAM from client does not crash server' => sub {
my $send_started = 0;
my $app = async sub {
my ($scope, $receive, $send) = @_;
await $receive->();
await $send->({
type => 'http.response.start',
status => 200,
headers => [['content-type', 'text/plain']],
});
await $send->({
type => 'http.response.body',
body => 'chunk1',
more => 1,
});
$send_started = 1;
# Keep streaming — client will RST_STREAM
for my $i (2..10) {
eval {
await $send->({
type => 'http.response.body',
body => "chunk$i",
more => ($i < 10) ? 1 : 0,
});
};
last if $@; # Stream may be reset
}
t/http2/12-error-handling.t view on Meta::CPAN
$stream_id = $f->{stream_id};
}
return 0;
},
);
h2c_handshake($client, $client_sock);
$client->submit_request(
method => 'GET',
path => '/streaming-rst',
scheme => 'http',
authority => 'localhost',
);
$client_sock->syswrite($client->mem_send);
# Wait for streaming to start
for (1..15) {
$loop->loop_once(0.1);
my $buf = '';
$client_sock->sysread($buf, 16384);
$client->mem_recv($buf) if length($buf);
my $out = $client->mem_send;
$client_sock->syswrite($out) if length($out);
last if $send_started;
}
t/http2/12-error-handling.t view on Meta::CPAN
# Send Extended CONNECT for WebSocket
my $ws_stream_id = $client->submit_request(
method => 'CONNECT',
path => '/ws/test',
scheme => 'http',
authority => 'localhost',
headers => [
[':protocol', 'websocket'],
['sec-websocket-version', '13'],
],
body => sub { return undef }, # streaming: keep open
);
$client_sock->syswrite($client->mem_send);
# Exchange until WebSocket is accepted
for (1..15) {
$loop->loop_once(0.1);
my $buf = '';
$client_sock->sysread($buf, 16384);
$client->mem_recv($buf) if length($buf);
my $out = $client->mem_send;
t/http2/13-sse-detection.t view on Meta::CPAN
use PAGI::Server::Connection;
use PAGI::Server;
use PAGI::Server::Protocol::HTTP1;
use PAGI::Server::Protocol::HTTP2;
my $loop = IO::Async::Loop->new;
my $protocol = PAGI::Server::Protocol::HTTP1->new;
# ============================================================
# Helpers (same pattern as t/http2/11-streaming.t)
# ============================================================
sub create_test_server {
my (%args) = @_;
my $server = PAGI::Server->new(
app => $args{app} // sub { },
host => '127.0.0.1',
port => 0,
quiet => 1,
http2 => 1,
t/http2/14-sse-events.t view on Meta::CPAN
use PAGI::Server::Connection;
use PAGI::Server;
use PAGI::Server::Protocol::HTTP1;
use PAGI::Server::Protocol::HTTP2;
my $loop = IO::Async::Loop->new;
my $protocol = PAGI::Server::Protocol::HTTP1->new;
# ============================================================
# Helpers (same pattern as t/http2/11-streaming.t)
# ============================================================
sub create_test_server {
my (%args) = @_;
my $server = PAGI::Server->new(
app => $args{app} // sub { },
host => '127.0.0.1',
port => 0,
quiet => 1,
http2 => 1,
t/http2/14-sse-events.t view on Meta::CPAN
exchange_frames($client, $client_sock, 20);
# Verify headers
is($response_headers{':status'}, '200', 'Status 200');
like($response_headers{'content-type'}, qr{text/event-stream}, 'Content-Type correct');
is($response_headers{'cache-control'}, 'no-cache', 'Cache-Control set');
ok(defined $response_headers{'date'} && length $response_headers{'date'},
'Date header present by default (design 11.4)');
# design doc section 11.4: HTTP/2 never emits HTTP/1-only Connection or
# chunked-framing headers (it uses DATA frames via submit_response_streaming).
ok(!exists $response_headers{'connection'}, 'no Connection header on HTTP/2 SSE responses');
ok(!exists $response_headers{'transfer-encoding'}, 'no Transfer-Encoding header on HTTP/2 SSE responses');
# Verify SSE event format
like($response_body, qr/event: update\n/, 'Named event field present');
like($response_body, qr/data: payload1\n/, 'Data field for event 1');
like($response_body, qr/id: 1\n/, 'ID field for event 1');
like($response_body, qr/data: payload2\n/, 'Data field for event 2');
like($response_body, qr/retry: 5000\n/, 'Retry field present');
t/http2/16-sse-cleanup.t view on Meta::CPAN
my $client = create_client();
h2c_handshake($client, $client_sock);
my $stream_id = $client->submit_request(
method => 'GET',
path => '/events',
scheme => 'http',
authority => 'localhost',
headers => [['accept', 'text/event-stream']],
body => sub { return undef }, # streaming: keep open, fed manually below
);
$client_sock->syswrite($client->mem_send);
# Wait for SSE to start with keepalive armed.
for (1..20) {
$loop->loop_once(0.1);
my $buf = '';
$client_sock->sysread($buf, 16384);
$client->mem_recv($buf) if length($buf);
my $out = $client->mem_send;
t/http2/16-sse-cleanup.t view on Meta::CPAN
my $client = create_client();
h2c_handshake($client, $client_sock);
my $stream_id = $client->submit_request(
method => 'GET',
path => '/events-early-receive',
scheme => 'http',
authority => 'localhost',
headers => [['accept', 'text/event-stream']],
body => sub { return undef }, # streaming: keep open, fed manually below
);
$client_sock->syswrite($client->mem_send);
local $SIG{__WARN__} = sub { push @app_warnings, $_[0] };
# Poll (bounded) until the app is dispatched and parked on receive()
# (body_pending armed but not yet resolved).
my $dispatched = 0;
for (1..20) {
$loop->loop_once(0.1);
t/http2/18-transport-leak.t view on Meta::CPAN
my $stream_closed = 0;
my $client = create_client(on_stream_close => sub { $stream_closed = 1; return 0 });
# h2c handshake.
$client->send_connection_preface;
$client_sock->syswrite($client->mem_send);
pump($client, $client_sock);
$client->submit_request(
method => 'GET',
path => '/streaming',
scheme => 'http',
authority => 'localhost',
);
$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');
t/http2/19-transport-callbacks.t view on Meta::CPAN
plan skip_all => "Server integration tests not supported on Windows" if $^O eq 'MSWin32';
BEGIN {
require PAGI::Server::Protocol::HTTP2;
PAGI::Server::Protocol::HTTP2->available
or plan(skip_all => 'HTTP/2 not available (Net::HTTP2::nghttp2 0.008+ required)');
}
# ============================================================
# Test: pagi.transport on_high_water / on_drain fire on a real HTTP/2 stream
# ============================================================
# A single streaming chunk larger than the high-water mark fires on_high_water
# at the synchronous post-push poke (before nghttp2 pulls from the per-stream
# queue). nghttp2 then pulls the queue down past the low-water mark (the client
# uses the default 64KB window and consumes), firing on_drain (deferred via
# loop->later). Both must be observed by the time the stream closes.
use PAGI::Server::Connection;
use PAGI::Server;
use PAGI::Server::Protocol::HTTP1;
use PAGI::Server::Protocol::HTTP2;
t/http2/19-transport-callbacks.t view on Meta::CPAN
$loop->loop_once(0.02);
my $buf = '';
$client_sock->sysread($buf, 65536);
$client->mem_recv($buf) if length($buf);
my $out = $client->mem_send;
$client_sock->syswrite($out) if length($out);
last if $cond && $cond->();
}
}
subtest 'on_high_water and on_drain fire on an h2 streaming response' => sub {
my ($hit_high, $hit_drain) = (0, 0);
my $app = async sub {
my ($scope, $receive, $send) = @_;
await $receive->();
my $t = $scope->{'pagi.transport'};
$t->on_high_water(sub { $hit_high++ });
$t->on_drain(sub { $hit_drain++ });
t/http2/19-transport-callbacks.t view on Meta::CPAN
my ($conn, $stream_io, $client_sock, $server) = create_h2c_connection(app => $app);
my $stream_closed = 0;
my $client = create_client(on_stream_close => sub { $stream_closed = 1; return 0 });
$client->send_connection_preface;
$client_sock->syswrite($client->mem_send);
pump($client, $client_sock);
$client->submit_request(
method => 'GET', path => '/streaming', scheme => 'http', authority => 'localhost',
);
$client_sock->syswrite($client->mem_send);
# Pump until the stream closes (and at least until both callbacks fire).
pump($client, $client_sock, sub { $stream_closed && $hit_high && $hit_drain });
ok($stream_closed, 'stream completed and closed');
ok($hit_high, 'on_high_water fired when the queue exceeded the high mark');
ok($hit_drain, 'on_drain fired once nghttp2 drained the queue below the low mark');
t/http2/20-sse-transport.t view on Meta::CPAN
plan skip_all => "Server integration tests not supported on Windows" if $^O eq 'MSWin32';
BEGIN {
require PAGI::Server::Protocol::HTTP2;
PAGI::Server::Protocol::HTTP2->available
or plan(skip_all => 'HTTP/2 not available (Net::HTTP2::nghttp2 0.008+ required)');
}
# ============================================================
# Test: pagi.transport on a real SSE-over-HTTP/2 stream
# ============================================================
# SSE-over-h2 must provide the same pagi.transport handle as HTTP/2 streaming
# and HTTP/1.1: the app cannot tell which transport carries its events.
# Subtest 1 exercises on_high_water / on_drain on a real stream; subtest 2
# proves the handle (and its $ss reference cycle) is collected at teardown.
use PAGI::Server::Connection;
use PAGI::Server;
use PAGI::Server::Protocol::HTTP1;
use PAGI::Server::Protocol::HTTP2;
my $loop = IO::Async::Loop->new;
t/http2/21-http-date-header.t view on Meta::CPAN
on_stream_close => sub { $stream_closed = 1; return 0 },
);
$client->send_connection_preface;
$client_sock->syswrite($client->mem_send);
pump($client, $client_sock);
$client->submit_request(
method => 'POST', path => '/upload', scheme => 'http', authority => 'localhost',
headers => [['content-type', 'application/octet-stream'], ['content-length', '50000']],
body => sub { return undef }, # streaming: keep open
);
$client_sock->syswrite($client->mem_send);
pump($client, $client_sock, sub { $stream_closed });
is($headers{':status'}, '413', 'content-length precheck rejected with 413');
ok(!$app_called, 'app was never called');
ok(defined $headers{date}, 'HTTP/2 413 (precheck) response carries a Date header');
$stream_io->close_now;
$loop->remove($server);
t/http2/21-http-date-header.t view on Meta::CPAN
on_header => sub { my ($sid, $n, $v) = @_; $headers{lc $n} = $v; return 0 },
);
$client->send_connection_preface;
$client_sock->syswrite($client->mem_send);
pump($client, $client_sock);
$client->submit_request(
method => 'CONNECT', path => '/ws/test', scheme => 'https', authority => 'localhost',
headers => [[':protocol', 'websocket'], ['sec-websocket-version', '13']],
body => sub { return undef }, # streaming: keep open
);
$client_sock->syswrite($client->mem_send);
pump($client, $client_sock, sub { defined $headers{':status'} && $headers{':status'} eq '401' });
is($headers{':status'}, '401', 'custom denial status used');
ok(defined $headers{date}, 'HTTP/2 WebSocket denial response carries a Date header');
$stream_io->close_now;
$loop->remove($server);
};
( run in 1.997 second using v1.01-cache-2.11-cpan-5c0b1e786e0 )