PAGI-Server
view release on metacpan or search on metacpan
- 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).
META.yml
Makefile.PL
README
README.md
SECURITY.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/Compliance.pod
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
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
Also included: `backpressure-test` - demonstrates backpressure handling (unnumbered utility example)
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
HTTP/1.1 for http/websocket/sse; on HTTP/2 for http streaming and sse. B<Not yet>
on WebSocket-over-HTTP/2 (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>.
=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
both HTTP/1.1 and HTTP/2.
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 HTTP/2 C<http> (streaming responses)
and C<sse> scopes, matching its HTTP/1.1 coverage of C<http>, C<sse>, and
C<websocket>.
B<Known gap:> WebSocket over HTTP/2 (RFC 8441) does B<not> yet provide
C<pagi.transport>. The HTTP/2 WebSocket send path hands each frame to nghttp2
through its push-style C<submit_data> entry point, which keeps no per-stream
send queue for the handle to measure, and the binding exposes no per-stream
buffered-byte API. Closing the gap means converting the HTTP/2 WebSocket send
path to the same pull-based data-provider model that HTTP/2 streaming and SSE
use. Until then a WebSocket application running over HTTP/2 sees
C<pagi.transport> as absent: the L<PAGI::WebSocket> helpers degrade quietly
(C<buffered_amount> reports C<0>, C<is_writable> stays true, and the callbacks
are no-ops). WebSocket over HTTP/1.1 is unaffected and provides the handle
normally.
The long-term goal is parity: C<pagi.transport> for every streaming scope type
over every transport, so an application need not know whether it runs over
HTTP/1.1 or HTTP/2.
=head2 Enabling HTTP/2
# TLS mode (h2 via ALPN)
pagi-server --http2 --ssl-cert cert.pem --ssl-key key.pem --app myapp.pl
# Cleartext mode (h2c)
pagi-server --http2 --app myapp.pl
lib/PAGI/Server/Connection.pm view on Meta::CPAN
use Encode;
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 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;
# =============================================================================
# Unrecognized Event Type Handler (PAGI spec compliance)
# =============================================================================
# Per main.mkdn: "Servers must raise exceptions if... The type field is unrecognized"
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;
# 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
}
}
elsif ($type eq 'http.response.body') {
my $ss = $weak_self->{h2_streams}{$stream_id} or return;
return unless $ss->{response_started};
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;
$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 {
# 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;
}
}
}
else {
lib/PAGI/Server/Connection.pm view on Meta::CPAN
[_validate_header_name($_->[0]), _validate_header_value($_->[1])]
} @$extra;
}
$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 streaming body that defers
$weak_self->{h2_session}->submit_response($stream_id,
status => 200,
headers => \@headers,
body => sub { return undef }, # defer until submit_data
);
$weak_self->_h2_write_pending;
# Process any data that arrived before accept
if (length($ss->{body}) > 0) {
my $buffered = $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
};
}
sub _h2_create_sse_send {
my ($self, $stream_id, $stream_state) = @_;
weaken(my $weak_self = $self);
# 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
if (!$has_content_type) {
push @final_headers, ['content-type', 'text/event-stream'];
}
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). Keepalive
# bytes are counted in the per-stream backlog so buffered_amount stays
# accurate, but they do not poke the watermark callbacks — a server
lib/PAGI/Server/Connection.pm view on Meta::CPAN
SSE events (C<sse.start>, C<sse.send>, C<sse.comment>, C<sse.keepalive>)
work transparently over both HTTP/1.1 and HTTP/2. Applications do not need
to change their SSE handling code based on protocol version.
=head2 How It Works
When a request arrives with C<Accept: text/event-stream>, the connection
detects it as SSE 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 applies at the B<connection level>,
not per-stream. Over HTTP/1.1, this is a non-issue since each SSE
stream occupies its own TCP connection. Over HTTP/2, where multiple
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
$loop->remove($server);
})->()->get;
};
# Test 18: send() after disconnect is a no-op per spec
# This tests the unit behavior of the send function when connection is closed
subtest 'send() after disconnect returns completed Future' => sub {
# Test the _create_send implementation directly by verifying
# that it returns immediately when closed flag is set
# 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/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);
};
# ============================================================
# 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/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/22-denial-response.t view on Meta::CPAN
# Extended CONNECT (RFC 8441) — matches t/http2/07-websocket.t submit_request
$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);
exchange_frames($client, $client_sock, 20);
ok(defined $captured_scope, 'app was called with a websocket scope');
ok(
$captured_scope && $captured_scope->{extensions}{'websocket.http.response'},
'extension websocket.http.response is advertised on h2 ws scope',
);
is($headers{':status'}, '401', 'h2 denial uses custom 401, not 200 or 403');
t/integration/app-file.t view on Meta::CPAN
my $response = $http->GET("http://127.0.0.1:$port$path")->get;
ok($response->code >= 400, "URL-encoded traversal blocked for $path (status: " . $response->code . ")");
}
$loop->remove($http);
$server->shutdown->get;
$loop->remove($server);
};
# =============================================================================
# Test: Large file streaming (verifies file response integration)
# =============================================================================
subtest 'Large file streaming' => sub {
# Create a temp directory with a large file
# Size is 65KB - just over the 64KB sync_file_threshold to trigger async path
# Kept small to minimize buffer pressure on constrained CI environments (e.g., FreeBSD)
my $test_dir = File::Temp::tempdir(CLEANUP => 1);
my $large_content = "X" x (65 * 1024); # 65KB file (exceeds 64KB threshold)
open my $fh, '>', "$test_dir/large.bin" or die;
print $fh $large_content;
close $fh;
my $server = create_server(root => $test_dir);
t/integration/app-file.t view on Meta::CPAN
is($response->code, 200, 'GET /large.bin returns 200');
is(length($response->content), length($large_content), 'Large file length matches');
is($response->content, $large_content, 'Large file content matches');
$loop->remove($http);
$server->shutdown->get;
$loop->remove($server);
};
# =============================================================================
# Test: Large file Range request streaming
# =============================================================================
subtest 'Large file Range request' => sub {
# Size is 65KB - just over the 64KB sync_file_threshold to trigger async path
my $test_dir = File::Temp::tempdir(CLEANUP => 1);
my $large_content = "X" x (65 * 1024); # 65KB file (exceeds 64KB threshold)
open my $fh, '>', "$test_dir/large.bin" or die;
print $fh $large_content;
close $fh;
my $server = create_server(root => $test_dir);
t/integration/psgi-bridge.t view on Meta::CPAN
my $response = $http->GET("http://127.0.0.1:$port/")->get;
is($response->code, 200, 'Response status is 200 OK');
is($response->decoded_content, 'Hello from filehandle', 'Filehandle body is read correctly');
$server->shutdown->get;
$loop->remove($server);
$loop->remove($http);
};
# Test 6: PSGI streaming response (coderef body) works
subtest 'PSGI streaming response (coderef body) works' => sub {
my $streaming_psgi = sub {
my ($env) = @_;
return sub {
my ($responder) = @_;
# Immediate response with streaming body
my $writer = $responder->([200, ['Content-Type' => 'text/plain']]);
$writer->write("Streaming ");
$writer->write("response");
$writer->close;
};
};
my $wrapper = PAGI::App::WrapPSGI->new(psgi_app => $streaming_psgi);
my $pagi_app = $wrapper->to_app;
# Wrap with lifespan handler
my $wrapped_app = async sub {
my ($scope, $receive, $send) = @_;
if ($scope->{type} eq 'lifespan') {
while (1) {
my $event = await $receive->();
if ($event->{type} eq 'lifespan.startup') {
await $send->({ type => 'lifespan.startup.complete' });
t/integration/response-integration.t view on Meta::CPAN
my $response = await $http->GET("http://127.0.0.1:$port/");
is $response->code, 200, 'status 200';
my $cookie = $response->header('Set-Cookie');
like $cookie, qr/session=abc123/, 'cookie value';
like $cookie, qr/Path=\//, 'cookie path';
like $cookie, qr/HttpOnly/, 'httponly flag';
})->get;
};
subtest 'streaming response' => sub {
my $app = async sub {
my ($scope, $receive, $send) = @_;
my $res = PAGI::Response->new($scope);
$res->content_type('text/plain')
->stream(async sub {
my ($writer) = @_;
await $writer->write("chunk1");
await $writer->write("chunk2");
await $writer->write("chunk3");
});
( run in 2.125 seconds using v1.01-cache-2.11-cpan-995e09ba956 )