PAGI-Tools
view release on metacpan or search on metacpan
- PAGI::Utils::to_app coerces coderefs, component objects, and class names; every
composition point accepts anything it accepts.
- builder's enable() accepts configured middleware instances.
- PAGI::App::Router: coderef route middleware can transform the channel.
- PAGI::Endpoint::Router route middleware are value-flow (await $next->() returns
the handler's response); App::Router dispatch returns the matched route's value.
- PAGI::Context: $ctx->text/html/json/redirect shorthands,
assert_http/websocket/sse guards, on_default catch-all, raw_send accessor.
- Backpressure/lifecycle helpers: on_complete, buffered_amount/high_water_mark/
low_water_mark, on_high_water/on_drain/is_writable, WebSocket->deny().
- PAGI::Request->multipart_stream: pull-based streaming multipart parsing with
app-controlled sinks; mutually exclusive with the buffered body methods.
- PAGI::App::Redirect and ::NotFound build a PAGI::Response value (modules kept
for the dynamic case).
- PAGI::Lifespan coerces its app arg via to_app, and reports shutdown-handler
failures instead of swallowing them.
- Spec: URLMap sets root_path; WrapCGI builds SCRIPT_NAME from root_path.
- Canonical scope-adding middleware route through the modify_scope helper.
- Documented the PAGI::Response subclassing seam for framework authors.
- Docs: recipes are now PAGI::Tools::Cookbook; Tutorial/Cookbook examples
corrected against the current API; chat-showcase example uses PAGI::App::Router
examples/README.md view on Meta::CPAN
```
Then launch any example with:
```
pagi-server examples/<name>/app.pl --port 5000
```
Examples assume you understand the core spec
(see the [PAGI project](https://github.com/jjn1056/pagi) for spec documents)
plus the relevant protocol documents.
Note: Low-level protocol examples (hello-http, streaming-response, websocket-echo
handshake, SSE broadcaster, lifespan-state, extension-fullflush, tls-introspection,
job-runner, utf8) shipped with the `PAGI-Server` distribution — they demonstrate
raw PAGI protocol details that belong alongside the server implementation.
## Example List
1. `09-psgi-bridge` - wraps a PSGI app for PAGI use (via `PAGI::App::WrapPSGI`)
2. `10-chat-showcase` - WebSocket chat demo with multiple clients
3. `13-contact-form` - form parsing and file uploads
4. `14-lifespan-utils` - lifespan hooks via `PAGI::Utils`
examples/sse-dashboard/README.md view on Meta::CPAN
# SSE Dashboard Example
Live dashboard using PAGI::SSE for real-time metrics streaming.
## Run
```bash
pagi-server --app examples/sse-dashboard/app.pl --port 5000
```
Visit http://localhost:5000/
## Features
- Real-time server metrics streaming
- Automatic keepalive for proxy compatibility
- Reconnection support via `Last-Event-ID`
- Multiple event types (`connected`, `reconnected`, `metrics`)
- Subscriber tracking
## API
- `SSE /events` - Metrics stream (2-second updates)
- `GET /*` - Static files from `public/`
examples/sse-dashboard/app.pl view on Meta::CPAN
#!/usr/bin/env perl
#
# Live Dashboard using PAGI::SSE
#
# Demonstrates real-time server metrics streaming with:
# - Automatic keepalive for proxy compatibility
# - Reconnection support via Last-Event-ID
# - Multiple event types
#
# Run: pagi-server --app examples/sse-dashboard/app.pl --port 5000
# Open: http://localhost:5000/
#
use strict;
use warnings;
lib/PAGI/App/File.pm view on Meta::CPAN
)->to_app;
=head1 DESCRIPTION
PAGI::App::File serves static files from a configured root directory.
=head2 Features
=over 4
=item * Efficient streaming (no memory bloat for large files)
=item * ETag caching with If-None-Match support (304 Not Modified)
=item * Range requests (HTTP 206 Partial Content)
=item * Automatic MIME type detection for common file types
=item * Index file resolution (index.html, index.htm)
=back
lib/PAGI/App/File.pm view on Meta::CPAN
status => 206,
headers => [
['content-type', $content_type],
['content-length', $length],
['content-range', "bytes $start-$end/$size"],
['accept-ranges', 'bytes'],
['etag', $etag],
],
});
# Use file response with offset/length for efficient streaming
if ($method eq 'HEAD') {
await $send->({ type => 'http.response.body', body => '', more => 0 });
}
else {
await $send->({
type => 'http.response.body',
file => $file_path,
offset => $start,
length => $length,
});
lib/PAGI/App/File.pm view on Meta::CPAN
type => 'http.response.start',
status => 200,
headers => [
['content-type', $content_type],
['content-length', $size],
['accept-ranges', 'bytes'],
['etag', $etag],
],
});
# Use file response for efficient streaming (sendfile or worker pool)
if ($method eq 'HEAD') {
await $send->({ type => 'http.response.body', body => '', more => 0 });
}
else {
await $send->({
type => 'http.response.body',
file => $file_path,
});
}
};
lib/PAGI/App/WrapPSGI.pm view on Meta::CPAN
last unless $event->{more};
}
# Create psgi.input
open my $input, '<', \$body or die $!;
$env->{'psgi.input'} = $input;
# Call PSGI app
my $response = $psgi_app->($env);
# Handle response - could be arrayref or coderef (streaming)
if (ref $response eq 'CODE') {
# Delayed/streaming response
await $self->_handle_streaming_response($send, $response);
} else {
await $self->_send_response($send, $response);
}
};
}
sub _build_env {
my ($self, $scope) = @_;
my %env = (
lib/PAGI/App/WrapPSGI.pm view on Meta::CPAN
SCRIPT_NAME => $scope->{root_path},
PATH_INFO => $scope->{path},
QUERY_STRING => $scope->{query_string},
SERVER_PROTOCOL => 'HTTP/' . $scope->{http_version},
'psgi.version' => [1, 1],
'psgi.url_scheme' => $scope->{scheme},
'psgi.errors' => \*STDERR,
'psgi.multithread' => 0,
'psgi.multiprocess' => 0,
'psgi.run_once' => 0,
'psgi.streaming' => 1,
'psgi.nonblocking' => 1,
);
# Add headers
for my $header (@{$scope->{headers}}) {
my ($name, $value) = @$header;
my $key = uc($name);
$key =~ s/-/_/g;
if ($key eq 'CONTENT_TYPE') {
$env{CONTENT_TYPE} = $value;
lib/PAGI/App/WrapPSGI.pm view on Meta::CPAN
local $/;
my $content = <$body>;
await $send->({
type => 'http.response.body',
body => $content // '',
more => 0,
});
}
}
# Handle PSGI delayed/streaming response pattern
async sub _handle_streaming_response {
my ($self, $send, $responder_callback) = @_;
my @body_chunks;
my $response_started = 0;
my $writer;
# Create a writer object for streaming
my $create_writer = sub {
my ($send_ref, $status, $headers) = @_;
return {
write => sub {
my ($chunk) = @_;
push @body_chunks, $chunk;
},
close => sub {
# Mark as closed - will be handled after responder returns
},
lib/PAGI/App/WrapPSGI.pm view on Meta::CPAN
});
} else {
await $send->({
type => 'http.response.body',
body => $body // '',
more => 0,
});
}
}
# Simple writer class for streaming responses
package PAGI::App::WrapPSGI::Writer;
sub write {
my ($self, $chunk) = @_;
push @{$self->{chunks}}, $chunk;
}
sub close {
my ($self) = @_;
# Nothing special needed - chunks are already collected
lib/PAGI/Middleware/ContentLength.pm view on Meta::CPAN
return async sub {
my ($scope, $receive, $send) = @_;
# Skip for non-HTTP requests
if ($scope->{type} ne 'http') {
await $app->($scope, $receive, $send);
return;
}
my @buffered_events;
my $has_content_length = 0;
my $is_streaming = 0;
my $status;
my @headers;
# Create intercepting send to buffer response
my $wrapped_send = async sub {
my ($event) = @_;
my $type = $event->{type};
if ($type eq 'http.response.start') {
$status = $event->{status};
@headers = @{$event->{headers} // []};
# Check if Content-Length already present
for my $h (@headers) {
if (lc($h->[0]) eq 'content-length') {
$has_content_length = 1;
last;
}
# If Transfer-Encoding is chunked, don't add Content-Length
if (lc($h->[0]) eq 'transfer-encoding' && lc($h->[1]) eq 'chunked') {
$is_streaming = 1;
last;
}
}
# If already has Content-Length or is streaming, pass through
if ($has_content_length || $is_streaming || $self->{auto_chunked}) {
await $send->($event);
return;
}
# Buffer the start event to add Content-Length later
push @buffered_events, $event;
}
elsif ($type eq 'http.response.body') {
# If we're passing through (has Content-Length or streaming)
if ($has_content_length || $is_streaming || $self->{auto_chunked}) {
await $send->($event);
return;
}
# Check if this is a streaming response (more => 1)
if ($event->{more}) {
$is_streaming = 1;
# Flush buffered events and switch to pass-through
for my $buffered (@buffered_events) {
await $send->($buffered);
}
@buffered_events = ();
await $send->($event);
return;
}
lib/PAGI/Middleware/ContentLength.pm view on Meta::CPAN
else {
# Pass through other events (trailers, etc.)
await $send->($event);
}
};
# Run the inner app
await $app->($scope, $receive, $wrapped_send);
# If we have buffered events, calculate Content-Length and send
if (@buffered_events && !$has_content_length && !$is_streaming) {
# Calculate total body length
my $body_length = 0;
for my $event (@buffered_events) {
if ($event->{type} eq 'http.response.body') {
$body_length += length($event->{body} // '');
}
}
# Send start with Content-Length
for my $event (@buffered_events) {
lib/PAGI/Middleware/ContentLength.pm view on Meta::CPAN
}
1;
__END__
=head1 NOTES
=over 4
=item * For streaming responses (multiple body events with more => 1),
this middleware switches to pass-through mode to avoid buffering.
=item * Responses that already have Content-Length are passed through unchanged.
=item * Responses with Transfer-Encoding: chunked are passed through unchanged.
=item * SSE and WebSocket responses should not use this middleware.
=back
lib/PAGI/Middleware/ETag.pm view on Meta::CPAN
use PAGI::Middleware::Builder;
my $app = builder {
enable 'ETag';
$my_app;
};
=head1 DESCRIPTION
PAGI::Middleware::ETag generates ETag headers for responses based on
the response body content. Works best with buffered (non-streaming) responses.
=head1 CONFIGURATION
=over 4
=item * weak (default: 0)
If true, generate weak ETags (W/"...").
=back
lib/PAGI/Middleware/ETag.pm view on Meta::CPAN
return async sub {
my ($scope, $receive, $send) = @_;
if ($scope->{type} ne 'http') {
await $app->($scope, $receive, $send);
return;
}
my @body_parts;
my $original_headers;
my $status;
my $is_streaming = 0;
my $wrapped_send = async sub {
my ($event) = @_;
if ($event->{type} eq 'http.response.start') {
$status = $event->{status};
$original_headers = $event->{headers};
# Check if already has ETag
for my $h (@{$original_headers // []}) {
if (lc($h->[0]) eq 'etag') {
# Already has ETag, pass through
await $send->($event);
$is_streaming = 1; # Flag to pass through body
return;
}
}
}
elsif ($event->{type} eq 'http.response.body') {
if ($is_streaming) {
await $send->($event);
return;
}
push @body_parts, $event->{body} // '';
# If streaming, can't generate ETag
if ($event->{more}) {
$is_streaming = 1;
await $send->({
type => 'http.response.start',
status => $status,
headers => $original_headers,
});
for my $part (@body_parts) {
await $send->({
type => 'http.response.body',
body => $part,
more => 1,
lib/PAGI/Middleware/ETag.pm view on Meta::CPAN
@body_parts = ();
}
}
else {
await $send->($event);
}
};
await $app->($scope, $receive, $wrapped_send);
return if $is_streaming;
# Generate ETag from body
my $body = join('', @body_parts);
my $etag = $self->_generate_etag($body);
# Add ETag to headers
my @new_headers = @{$original_headers // []};
push @new_headers, ['ETag', $etag];
await $send->({
lib/PAGI/Middleware/GZip.pm view on Meta::CPAN
for my $h (@{$event->{headers} // []}) {
if (lc($h->[0]) eq 'content-type') {
$content_type = $h->[1];
last;
}
}
$response_started = 1;
# Don't send yet - buffer to compress
}
elsif ($event->{type} eq 'http.response.body') {
# If we're already in streaming mode, pass through all chunks
if ($headers_sent) {
await $send->($event);
return;
}
push @body_parts, $event->{body} // '';
# If streaming (more => 1), switch to pass-through mode
if ($event->{more}) {
if (!$headers_sent) {
await $send->({
type => 'http.response.start',
status => 200,
headers => $original_headers,
});
$headers_sent = 1;
}
await $send->($event);
}
}
else {
await $send->($event);
}
};
await $app->($scope, $receive, $wrapped_send);
# If headers already sent (streaming), we're done
return if $headers_sent;
# Combine body
my $body = join('', @body_parts);
# Decide whether to compress
my $should_compress = $self->_should_compress($content_type, length($body));
if ($should_compress && length($body) > 0) {
my $compressed;
lib/PAGI/Middleware/Head.pm view on Meta::CPAN
elsif ($type eq 'http.response.body') {
# Suppress body content but preserve the event structure
# Send an empty body with more => 0 to complete the response
if (!$event->{more}) {
await $send->({
type => 'http.response.body',
body => '',
more => 0,
});
}
# Otherwise, skip the event entirely (streaming chunks)
}
elsif ($type eq 'http.response.trailers') {
# Skip trailers for HEAD requests
}
else {
# Pass through other events
await $send->($event);
}
};
lib/PAGI/Middleware/Static.pm view on Meta::CPAN
# For HEAD requests, don't send body
if ($scope->{method} eq 'HEAD') {
await $send->({
type => 'http.response.body',
body => '',
more => 0,
});
return;
}
# Use file response for efficient streaming (sendfile or worker pool)
# This also enables XSendfile middleware to intercept the response
if ($is_range) {
await $send->({
type => 'http.response.body',
file => $file_path,
offset => $start,
length => $body_size,
});
}
else {
lib/PAGI/Middleware/Static.pm view on Meta::CPAN
=head1 CACHING
The middleware generates ETags based on file path, size, and modification
time. Clients can use If-None-Match to receive 304 Not Modified responses
when the file hasn't changed.
=head1 RANGE REQUESTS
The middleware supports HTTP Range requests for partial content, useful
for resumable downloads and media streaming. Only single byte ranges
are supported (not multi-range requests).
=head1 SEE ALSO
L<PAGI::Middleware> - Base class for middleware
=cut
lib/PAGI/Request.pm view on Meta::CPAN
return PAGI::Response->new($self->{scope});
}
# Application state (injected by PAGI::Lifespan, read-only)
sub state {
my $self = shift;
return $self->{scope}{state} // {};
}
# Body streaming - mutually exclusive with buffered body methods
sub body_stream {
my ($self, %opts) = @_;
croak "Body already consumed; streaming not available" if $self->{scope}{'pagi.request.body.read'};
croak "Body streaming already started" if $self->{scope}{'pagi.request.body.stream.created'};
$self->{scope}{'pagi.request.body.stream.created'} = 1;
my $max_bytes = $opts{max_bytes};
my $limit_name = defined $max_bytes ? 'max_bytes' : undef;
if (!defined $max_bytes) {
my $cl = $self->content_length;
if (defined $cl) {
$max_bytes = $cl;
$limit_name = 'content-length';
lib/PAGI/Request.pm view on Meta::CPAN
boundary => $boundary,
map { defined $opts{$_} ? ($_ => $opts{$_}) : () }
qw(max_files max_fields max_field_size max_file_size max_request_body),
);
}
# Read raw body bytes (async, cached in scope)
async sub body {
my $self = shift;
croak "Body streaming already started; buffered helpers unavailable"
if $self->{scope}{'pagi.request.body.stream.created'};
# Return cached body if already read
return $self->{scope}{'pagi.request.body'} if $self->{scope}{'pagi.request.body.read'};
my $receive = $self->{receive};
die "No receive callback provided" unless $receive;
my $body = '';
while (1) {
lib/PAGI/Request.pm view on Meta::CPAN
# Raw singular accessor
async sub raw_form_param {
my ($self, $name) = @_;
return await $self->form_param($name, raw => 1);
}
# Parse multipart form (internal, cached in scope)
async sub _parse_multipart_form {
my ($self, %opts) = @_;
croak "Body streaming already started; buffered helpers unavailable"
if $self->{scope}{'pagi.request.body.stream.created'};
# Already parsed?
return $self->{scope}{'pagi.request.form'}
if $self->{scope}{'pagi.request.form'} && $self->{scope}{'pagi.request.uploads'};
# Extract boundary from content-type
my $ct = $self->header('content-type') // '';
my ($boundary) = $ct =~ /boundary=([^;\s]+)/;
$boundary =~ s/^["']|["']$//g if $boundary; # Strip quotes
lib/PAGI/Request.pm view on Meta::CPAN
=head2 body_stream
my $stream = $req->body_stream;
my $stream = $req->body_stream(
max_bytes => 10 * 1024 * 1024, # 10MB limit
decode => 'UTF-8', # Decode to UTF-8
strict => 1, # Strict UTF-8 decoding
);
Returns a L<PAGI::Request::BodyStream> for streaming body consumption. This is
useful for processing large request bodies incrementally without loading them
entirely into memory.
B<Options:>
=over 4
=item * C<max_bytes> - Maximum body size. Defaults to Content-Length header if present.
=item * C<decode> - Encoding to decode chunks to (typically 'UTF-8').
=item * C<strict> - If true, throw on invalid UTF-8. Default: false (use replacement chars).
=back
B<Important:> Body streaming is mutually exclusive with buffered body methods
(C<body>, C<text>, C<json>, C<form_params>). Once you start streaming, you cannot use
those methods, and vice versa.
Example:
# Stream large upload to file
my $stream = $req->body_stream(max_bytes => 100 * 1024 * 1024);
await $stream->stream_to_file('/uploads/data.bin');
See L<PAGI::Request::BodyStream> for full documentation.
lib/PAGI/Request.pm view on Meta::CPAN
my $stream = $req->multipart_stream;
my $stream = $req->multipart_stream(
max_files => 1000,
max_fields => 1000,
max_field_size => 1024 * 1024,
max_file_size => 100 * 1024 * 1024,
max_request_body => 1024 * 1024 * 1024,
);
Returns a L<PAGI::Request::MultipartStream> for pull-based streaming of a
C<multipart/form-data> request body. You pull one part at a time and choose
where each one goes:
while (defined(my $part = await $stream->next)) {
if ($part->is_file) {
await $part->stream_to_file($path);
}
else {
my $value = await $part->value; # raw bytes; you decode
}
lib/PAGI/Request.pm view on Meta::CPAN
This is useful for racing against other async operations.
=head2 buffered_amount, high_water_mark, low_water_mark
my $pending = $req->buffered_amount; # bytes queued, not yet on the wire
my $ceiling = $req->high_water_mark; # backpressure ceiling (or undef)
my $floor = $req->low_water_mark; # backpressure floor (or undef)
Outbound flow-control introspection, delegated to the server-provided
C<pagi.transport> handle (see L<PAGI::Spec::Www/"Transport Flow Control">). For a
streaming response, use C<buffered_amount> to conflate or shed load instead of
only blocking on drain; when the server does not provide the handle,
C<buffered_amount> returns C<0> and the watermarks return C<undef>.
=head2 on_high_water, on_drain, is_writable
$req->on_high_water(sub { $source->pause }); # backpressure engaged
$req->on_drain(sub { $source->resume }); # backpressure cleared
last unless $req->is_writable; # below the high mark?
Backpressure controls delegated to the C<pagi.transport> handle. C<on_high_water>
lib/PAGI/Request/BodyStream.pm view on Meta::CPAN
=head1 NAME
PAGI::Request::BodyStream - Streaming body consumption for PAGI requests
=head1 SYNOPSIS
use PAGI::Request::BodyStream;
use Future::AsyncAwait;
# Basic streaming
my $stream = PAGI::Request::BodyStream->new(receive => $receive);
while (!$stream->is_done) {
my $chunk = await $stream->next_chunk;
last unless defined $chunk;
print "Got chunk: ", length($chunk), " bytes\n";
}
# With size limit
my $stream = PAGI::Request::BodyStream->new(
lib/PAGI/Request/BodyStream.pm view on Meta::CPAN
await $stream->stream_to_file('/tmp/upload.dat');
# Stream to custom sink
await $stream->stream_to(async sub ($chunk) {
# Process chunk
print STDERR "Processing: ", length($chunk), " bytes\n";
});
=head1 DESCRIPTION
PAGI::Request::BodyStream provides streaming body consumption for large request
bodies. This is useful when you need to process request data incrementally
without loading the entire body into memory.
The stream is pull-based: you call C<next_chunk()> to receive the next chunk
of data. The stream handles:
=over 4
=item * Size limits with customizable error messages
=item * UTF-8 decoding with proper handling of incomplete sequences at chunk boundaries
=item * Client disconnect detection
=item * Convenient file streaming with C<stream_to_file()>
=back
B<Important>: Streaming is mutually exclusive with buffered body methods like
C<body()>, C<json()>, C<form()> in L<PAGI::Request>. Once you start streaming,
you cannot use those methods.
=head1 CONSTRUCTOR
=head2 new
my $stream = PAGI::Request::BodyStream->new(
receive => $receive, # Required: PAGI receive callback
max_bytes => 10485760, # Optional: max body size
decode => 'UTF-8', # Optional: decode to UTF-8
lib/PAGI/Request/BodyStream.pm view on Meta::CPAN
sub is_done {
my ($self) = @_;
return $self->{_done};
}
=head2 error
my $error = $stream->error;
Returns any error that occurred during streaming, or undef.
=cut
sub error {
my ($self) = @_;
return $self->{_error};
}
=head2 stream_to_file
lib/PAGI/Request/MultipartStream.pm view on Meta::CPAN
$PAGI::Request::MultipartStream::VERSION = '0.002001';
use strict;
use warnings;
use Future::AsyncAwait;
use Carp qw(croak);
use HTTP::MultiPartParser;
=head1 NAME
PAGI::Request::MultipartStream - Pull-based streaming multipart/form-data engine
=head1 SYNOPSIS
use PAGI::Request::MultipartStream;
use Future::AsyncAwait;
# Usually obtained via $req->multipart_stream, not constructed directly:
my $stream = $req->multipart_stream;
while (defined(my $part = await $stream->next)) {
if ($part->is_file) {
await $part->stream_to_file($path);
}
else {
my $value = await $part->value; # raw bytes; you decode
}
}
=head1 DESCRIPTION
A pull-based streaming parser for C<multipart/form-data> request bodies. Each
part of the body is exposed in turn as a L<PAGI::Request::Part> via C<next>,
and B<the application decides where each part goes>: you choose its sink (a
file, an object store, an async transform) per part, rather than accepting the
buffered, spool-each-upload-to-a-temp-file behaviour of C<form_params> and
C<upload> in L<PAGI::Request>.
Because you own the sink, it can be fully asynchronous: C<stream_to> awaits a
sink that returns a Future, so a slow downstream naturally backpressures the
read. This is what the buffered multipart path cannot offer -- its spool to a
temp file is blocking.
lib/PAGI/Request/MultipartStream.pm view on Meta::CPAN
my $stream = PAGI::Request::MultipartStream->new(
receive => $receive, # required: PAGI receive callback
boundary => $boundary, # required: multipart boundary
max_files => 1000, # optional limits (defaults shown)
max_fields => 1000,
max_field_size => 1024 * 1024,
max_file_size => 100 * 1024 * 1024,
max_request_body => 1024 * 1024 * 1024,
);
Creates a new streaming multipart engine. Most applications do not call this
directly -- they obtain a ready-built stream from
L<PAGI::Request/multipart_stream>, which extracts the boundary from the
request's C<Content-Type> and passes through the same limit options.
C<receive> and C<boundary> are required. The remaining options cap the body to
bound memory and resource use:
=over 4
=item * C<max_files> - Maximum number of file parts. Default: 1000.
lib/PAGI/Request/MultipartStream.pm view on Meta::CPAN
package PAGI::Request::Part;
use strict;
use warnings;
use Future::AsyncAwait;
use Carp qw(croak);
use Fcntl qw(O_WRONLY O_CREAT O_EXCL O_NOFOLLOW);
=head1 NAME
PAGI::Request::Part - A single part of a streaming multipart request
=head1 DESCRIPTION
A value object representing one part yielded by
L<PAGI::Request::MultipartStream>. It carries the part's metadata (name,
filename, headers) and provides the methods that consume the part's body: pull
it chunk by chunk, buffer it whole, or drain it to a sink of your choosing.
A part's body must be consumed before the next part is fetched. Calling
C<< $stream->next >> while a part is only partially read drains the rest of
lib/PAGI/Response.pm view on Meta::CPAN
await $res->respond($send);
The single send primitive for a detached response value. Reads the
accumulated status, headers, and body from C<$self> and emits the
appropriate PAGI protocol events via C<$send>.
C<$send> must be a coderef (the PAGI send callback). C<respond> does
B<not> mutate the response object, so the same response value can be
passed to C<respond> multiple times for different connections.
For streaming responses (set up via the C<_stream> slot), C<respond>
sends the start event, runs the stream callback with a
L<PAGI::Response::Writer>, and ensures the writer is closed.
Returns a L<Future>.
=head2 to_app
my $app = $res->to_app;
Returns a PAGI application coderef C<sub ($scope, $receive, $send)> that
lib/PAGI/Response.pm view on Meta::CPAN
=head2 stream
$res->stream(async sub {
my ($writer) = @_;
await $writer->write("chunk1");
await $writer->write("chunk2");
await $writer->close();
});
PAGI::Response->stream($callback);
Store a streaming callback. When the response is sent via L</respond>, the callback
receives a L<PAGI::Response::Writer> and streams chunks. Returns C<$self>.
=head2 writer
my $writer = await $res->writer($send);
my $writer = await $res->writer($send, on_close => sub { cleanup() });
my $writer = await $res->writer($send, on_close => async sub { await cleanup() });
Returns a L<PAGI::Response::Writer> directly, sending headers immediately.
Unlike C<stream()>, the writer is not scoped to a callback — you own it
lib/PAGI/Response.pm view on Meta::CPAN
PAGI::Response->send_file('/path/to/file.pdf');
# Partial file (for range requests)
$res->send_file('/path/to/video.mp4',
offset => 1024, # Start from byte 1024
length => 65536, # Send 64KB
);
Set the response to serve a file. Stats the file and sets Content-Type,
Content-Length, and Content-Disposition at call time. The PAGI protocol's
C<file> key is used for efficient server-side streaming (file not read into
memory) when L</respond> is called. Returns C<$self>.
For production, use L<PAGI::Middleware::XSendfile> to delegate file serving
to your reverse proxy.
B<Options:>
=over 4
=item * C<filename> - Set Content-Disposition attachment filename
lib/PAGI/Response.pm view on Meta::CPAN
sub stream {
my ($proto, $callback) = @_;
my $self = $proto->_self_or_new;
$self->{_stream} = $callback;
return $self;
}
async sub writer {
my ($self, $send, %opts) = @_;
croak("send must be a coderef") unless ref($send) eq 'CODE';
# A writer takes over the connection for live streaming; it can only be
# taken once on a given response value. (The cross-stack "did a response
# start" fact lives on pagi.connection; this is a local single-takeover guard.)
croak("Response already sent") if $self->{_writer_started};
$self->{_writer_started} = 1;
# Send headers
await $send->({
type => 'http.response.start',
status => $self->status,
headers => $self->_render_headers(undef),
lib/PAGI/Response.pm view on Meta::CPAN
# Store the file send descriptor; respond() handles the actual emission.
# offset/length are stored only when they narrow the full-file default.
my $file_desc = { path => $path };
$file_desc->{offset} = $offset if $offset > 0;
$file_desc->{length} = $length if $length < $max_length;
$self->{_file} = $file_desc;
return $self;
}
# Writer class for streaming responses
package PAGI::Response::Writer {
use strict;
use warnings;
use Future::AsyncAwait;
use Carp qw(croak);
use Scalar::Util qw(blessed);
sub new {
my ($class, $send, %opts) = @_;
my $self = bless {
lib/PAGI/Test/Client.pm view on Meta::CPAN
=back
=head1 LIMITATIONS
=over 4
=item *
HTTP request bodies are delivered as a single C<http.request> event. This
client does B<not> currently simulate multi-event request body streaming or
disconnects mid-request-body.
=item *
HTTP response trailers (C<http.response.trailers>) are not exposed through
L<PAGI::Test::Response>. If your application depends on trailer semantics,
test it against L<PAGI::Server>.
=item *
lib/PAGI/Tools/Cookbook.pod view on Meta::CPAN
$router->sse('/events/:channel' => async sub {
my ($scope, $receive, $send) = @_;
my $sse = PAGI::SSE->new($scope, $receive, $send);
# Access channel parameter
my $channel = $scope->{path_params}{channel};
await $sse->start;
await $sse->send_event(data => "Subscribed to: $channel");
# ... streaming logic ...
});
$router->to_app;
=head2 Route-Level Middleware
Apply middleware to specific routes by passing an arrayref:
use strict;
use warnings;
lib/PAGI/Tools/Cookbook.pod view on Meta::CPAN
async sub handle_channel {
my ($self, $ctx) = @_;
my $sse = $ctx->sse;
# Access path parameter
my $channel = $sse->path_param('channel');
await $sse->start;
await $sse->send_event(data => "Channel: $channel");
# ... streaming logic ...
}
1;
=head3 Initializing Resources
C<PAGI::Endpoint::Router> calls your C<routes> method once, when the app is
built, on the instance that lives for the app's lifetime. That makes C<routes>
the natural place to initialize per-worker resources and seed state: whatever
you store in C<< $self->state >> is visible to every handler.
lib/PAGI/Tools/Cookbook.pod view on Meta::CPAN
-------------------------- ------------- -------------------------------------
has_status build Was a status code set explicitly?
has_header($name) build Was this header set?
has_content_type build Was Content-Type set?
has_body_source build Was a body (bytes/file/stream) set?
is_sent send Has the response been emitted?
(Future from respond) finished Has every byte been written?
$writer->bytes_written finished* How many bytes streamed so far?
$writer->is_closed finished* Has the stream writer been closed?
The C<$writer> signals (*) apply only while streaming, on the
L<PAGI::Response::Writer> handed to your C<stream> callback or returned by
C<writer>. They describe the live stream; the response value itself only carries
the build and send signals.
=head2 has_body_source is intent, not bytes
The signal most likely to surprise you is C<has_body_source>. It reports that a
body B<source> is registered — not that bytes exist, and not that anything was
sent. For a stream this matters: the callback is stored at build time and does
not run until C<respond>, so a freshly-registered stream has produced zero bytes
lib/PAGI/Tools/Cookbook.pod view on Meta::CPAN
=head2 Deciding "did the handler produce a response?"
A framework that lets handlers populate C<< $ctx->res >> and then auto-sends it
needs to know, after the handler returns, whether to send, skip, or fall through
to the next route. That is a three-way decision, and it is a precedence ladder —
not a single predicate:
# After running the handler, which mutated $ctx->res:
if ($ctx->res->is_sent) {
# The handler already took the connection itself — e.g. it called
# ->writer($send) for live streaming, or used SSE/WebSocket. Do nothing;
# sending again would be a double-send.
}
elsif ($ctx->res->has_body_source || $ctx->res->has_status) {
# The handler registered a body, or set a status (a bare 204 / redirect
# has no body but IS a response). Send it once, through the guard.
await $ctx->respond($ctx->res);
}
else {
# The handler touched nothing send-able — fall through to the next match
# (or a 404).
}
Three things make this ladder correct:
=over 4
=item * B<Check C<is_sent> first.> A handler that called C<< ->writer($send) >>
has already emitted headers (C<writer> marks the response sent), but
C<has_body_source> stays false because the live-writer path bypasses the body
slots. Without the C<is_sent> guard you would mistake an actively-streaming
response for "nothing produced".
=item * B<C<|| has_status> is required.> A C<302> redirect or a C<204> can have
no body source, yet it is a real response. C<has_status> catches the
status-only case so you do not 404 a legitimate empty-body response.
=item * B<Never inspect the private slots.> Use C<has_body_source>, not
C<< exists $res->{_stream} >> and friends — those are private and may change
(see L<PAGI::Response/SUBCLASSING (FRAMEWORK INTEGRATION)>). The predicate is
the supported way to ask.
lib/PAGI/Tools/Tutorial.pod view on Meta::CPAN
await PAGI::Response->text('Hello, world!')->respond($send);
};
C<PAGI::Response> assembles the two events for you, and C<respond> sends them.
Returning JSON is just as short:
await PAGI::Response->json({ hello => 'world' })->respond($send);
From here the tutorial builds up: routing requests to handlers (PART 2),
reading form and JSON input (PART 3), the full response builder (PART 4),
real-time and streaming (PART 5), middleware (PART 6), and composing larger
applications (PART 7).
=head1 PART 2: ROUTING
=head2 2.1 PAGI::App::Router - Basic Routing
L<PAGI::App::Router> provides lightweight functional routing:
use PAGI::App::Router;
use PAGI::Response;
lib/PAGI/Tools/Tutorial.pod view on Meta::CPAN
}
=back
=head1 PART 7: COMPOSING APPLICATIONS
PAGI ships with several ready-to-use applications for common tasks. These can be mounted with routers or used standalone.
=head2 7.1 PAGI::App::File - Static Files
L<PAGI::App::File> serves static files with security, caching, and streaming:
use PAGI::App::File;
my $static = PAGI::App::File->new(root => './public');
$router->mount('/static' => $static);
Features:
=over 4
=item * Efficient streaming (large files don't consume memory)
=item * ETag caching with 304 Not Modified support
=item * HTTP Range requests for resume support
=item * Automatic MIME type detection
=item * Security: path traversal protection
=back
t/35-gzip-concurrency.t view on Meta::CPAN
# Each result should be unique and contain its request ID
for my $i (1..5) {
like($results[$i-1], qr/^response-$i-x+$/, "request $i got correct response");
}
};
# =============================================================================
# Test: Streaming responses bypass compression
# =============================================================================
subtest 'streaming responses bypass compression' => sub {
my $gzip = PAGI::Middleware::GZip->new(min_size => 10);
# App that streams response in multiple chunks
my $app = async sub {
my ($scope, $receive, $send) = @_;
await $send->({ type => 'http.response.start', status => 200, headers => [['content-type', 'text/plain']] });
await $send->({ type => 'http.response.body', body => 'chunk1', more => 1 });
await $send->({ type => 'http.response.body', body => 'chunk2', more => 1 });
await $send->({ type => 'http.response.body', body => 'chunk3', more => 0 });
};
t/35-gzip-concurrency.t view on Meta::CPAN
{ type => 'http', headers => [['accept-encoding', 'gzip']] },
async sub { { type => 'http.disconnect' } },
async sub { push @events, $_[0] },
);
});
# Streaming should pass through without compression
is(scalar @events, 4, 'four events sent (start + 3 body chunks)');
is($events[0]{type}, 'http.response.start', 'first event is response start');
# Check that Content-Encoding: gzip is NOT present (streaming bypasses compression)
my $has_gzip = grep { $_->[0] eq 'Content-Encoding' && $_->[1] eq 'gzip' } @{$events[0]{headers}};
ok(!$has_gzip, 'streaming response not compressed');
# Body chunks should be uncompressed
is($events[1]{body}, 'chunk1', 'first chunk intact');
is($events[2]{body}, 'chunk2', 'second chunk intact');
is($events[3]{body}, 'chunk3', 'third chunk intact');
};
# =============================================================================
# Test: No client gzip support
# =============================================================================
t/35-gzip-concurrency.t view on Meta::CPAN
async sub { { type => 'http.disconnect' } },
async sub { push @events, $_[0] },
);
});
is(scalar @events, 2, 'two events sent');
is($events[1]{body}, 'x' x 100, 'body is uncompressed');
};
# =============================================================================
# Test: State isolation between streaming and non-streaming requests
# =============================================================================
subtest 'state isolation: streaming then non-streaming' => sub {
my $gzip = PAGI::Middleware::GZip->new(min_size => 10);
# First: streaming request
my $streaming_app = async sub {
my ($scope, $receive, $send) = @_;
await $send->({ type => 'http.response.start', status => 200, headers => [['content-type', 'text/plain']] });
await $send->({ type => 'http.response.body', body => 'stream', more => 1 });
await $send->({ type => 'http.response.body', body => 'end', more => 0 });
};
my $wrapped_streaming = $gzip->wrap($streaming_app);
my @streaming_events;
run_async(async sub {
await $wrapped_streaming->(
{ type => 'http', headers => [['accept-encoding', 'gzip']] },
async sub { { type => 'http.disconnect' } },
async sub { push @streaming_events, $_[0] },
);
});
# Second: non-streaming request (should compress)
my $buffered_app = async sub {
my ($scope, $receive, $send) = @_;
await $send->({ type => 'http.response.start', status => 200, headers => [['content-type', 'text/plain']] });
await $send->({ type => 'http.response.body', body => 'x' x 100, more => 0 });
};
my $wrapped_buffered = $gzip->wrap($buffered_app);
my @buffered_events;
run_async(async sub {
await $wrapped_buffered->(
{ type => 'http', headers => [['accept-encoding', 'gzip']] },
async sub { { type => 'http.disconnect' } },
async sub { push @buffered_events, $_[0] },
);
});
# Streaming should NOT be compressed
my $streaming_has_gzip = grep { $_->[0] eq 'Content-Encoding' && $_->[1] eq 'gzip' } @{$streaming_events[0]{headers}};
ok(!$streaming_has_gzip, 'streaming response not compressed');
# Buffered should BE compressed
my $buffered_has_gzip = grep { $_->[0] eq 'Content-Encoding' && $_->[1] eq 'gzip' } @{$buffered_events[0]{headers}};
ok($buffered_has_gzip, 'buffered response IS compressed');
# Verify the buffered body decompresses correctly
my $compressed = $buffered_events[1]{body};
my $uncompressed;
gunzip(\$compressed, \$uncompressed) or die "gunzip failed: $GunzipError";
is($uncompressed, 'x' x 100, 'buffered body decompresses correctly');
t/middleware/01-content-length.t view on Meta::CPAN
my ($event) = @_; push @sent, $event },
);
});
# Count Content-Length headers
my @lengths = grep { lc($_->[0]) eq 'content-length' } @{$sent[0]{headers}};
is scalar(@lengths), 1, 'only one Content-Length header';
is $lengths[0][1], '5', 'original Content-Length preserved';
};
subtest 'ContentLength passes through streaming responses' => sub {
my $mw = PAGI::Middleware::ContentLength->new;
# App that streams response (more => 1)
my $app = async sub {
my ($scope, $receive, $send) = @_;
await $send->({
type => 'http.response.start',
status => 200,
headers => [['content-type', 'text/plain']],
});
t/middleware/01-content-length.t view on Meta::CPAN
await $wrapped->(
{ type => 'http', path => '/' },
async sub { { type => 'http.disconnect' } },
async sub {
my ($event) = @_; push @sent, $event },
);
});
is scalar(@sent), 3, 'three events sent (start + 2 body)';
# No Content-Length should be added for streaming
my @lengths = grep { lc($_->[0]) eq 'content-length' } @{$sent[0]{headers}};
is scalar(@lengths), 0, 'no Content-Length for streaming response';
};
subtest 'ContentLength skips non-HTTP requests' => sub {
my $mw = PAGI::Middleware::ContentLength->new;
my $app_called = 0;
my $app = async sub {
my ($scope, $receive, $send) = @_;
$app_called = 1;
await $send->({
t/middleware/02-head.t view on Meta::CPAN
async sub { { type => 'http.disconnect' } },
async sub {
my ($event) = @_; push @sent, $event },
);
});
is $received_method, 'GET', 'GET request passed through unchanged';
is $sent[1]{body}, 'Hello', 'body is preserved for GET';
};
subtest 'Head middleware handles streaming responses' => sub {
my $mw = PAGI::Middleware::Head->new;
my $app = async sub {
my ($scope, $receive, $send) = @_;
await $send->({
type => 'http.response.start',
status => 200,
headers => [['content-type', 'text/plain']],
});
await $send->({
t/middleware/15-xsendfile.t view on Meta::CPAN
is scalar(@sent), 2, 'two events sent';
# No X-Sendfile header
my $xsendfile = find_header($sent[0]{headers}, 'X-Sendfile');
is $xsendfile, undef, 'no X-Sendfile header for regular body';
is $sent[1]{body}, 'Hello, World!', 'body passed through unchanged';
};
subtest 'streaming response passes through' => sub {
my $mw = PAGI::Middleware::XSendfile->new(type => 'X-Sendfile');
my $app = async sub {
my ($scope, $receive, $send) = @_;
await $send->({
type => 'http.response.start',
status => 200,
headers => [],
});
await $send->({
t/request-body-stream.t view on Meta::CPAN
# Mock receive function
sub mock_receive {
my @chunks = @_;
my $i = 0;
return sub {
my $chunk = $chunks[$i++];
return Future->done($chunk);
};
}
subtest 'basic streaming' => sub {
my $receive = mock_receive(
{ type => 'http.request', body => 'Hello', more => 1 },
{ type => 'http.request', body => ' World', more => 0 },
);
my $stream = PAGI::Request::BodyStream->new(receive => $receive);
my $chunk1 = $stream->next_chunk->get;
is $chunk1, 'Hello', 'first chunk';
ok !$stream->is_done, 'not done yet';
t/request-body-stream.t view on Meta::CPAN
};
subtest 'body_stream mutual exclusivity' => sub {
use PAGI::Request;
# Streaming then buffered should fail
my $scope1 = { method => 'POST', path => '/', headers => [] };
my $receive1 = sub { Future->done({ type => 'http.request', body => 'test', more => 0 }) };
my $req1 = PAGI::Request->new($scope1, $receive1);
$req1->body_stream;
like dies { $req1->body->get }, qr/streaming/, 'body after stream fails';
# Buffered then streaming should fail
my $scope2 = { method => 'POST', path => '/', headers => [] };
my $receive2 = sub { Future->done({ type => 'http.request', body => 'x', more => 0 }) };
my $req2 = PAGI::Request->new($scope2, $receive2);
$req2->body->get;
like dies { $req2->body_stream }, qr/consumed|read/, 'stream after body fails';
};
done_testing;
t/request/11-multipart-handler.t view on Meta::CPAN
max_field_size => 50 * 1024, # 50KB limit for form fields
);
like(
dies { (async sub { await $handler->parse })->()->get },
qr/Form field too large/,
'dies when exceeding max_field_size'
);
};
subtest 'chunked input streaming' => sub {
my $boundary = '----Chunked';
my $full_body = build_multipart($boundary,
{ name => 'title', data => 'Hello' },
{ name => 'doc', filename => 'test.txt', content_type => 'text/plain', data => 'file content here' },
);
# Split body into chunks
my @chunks;
my $chunk_size = 30;
for (my $i = 0; $i < length($full_body); $i += $chunk_size) {
t/sse/11-async-close.t view on Meta::CPAN
if ($close_f) {
$gate->done;
Future->wrap($close_f)->get;
ok($cleanup_ran, 'async on_close ran to completion via close()');
}
};
# Same defect, seen through a real application: an SSE app that streams an event,
# registers an async cleanup (e.g. release a subscription / close a DB handle),
# then explicitly closes the stream -- a completely ordinary "done streaming"
# flow. The app uses the forward-looking idiom `await $sse->close`.
subtest 'an SSE app that closes after streaming, with async cleanup, does not crash' => sub {
my @sent;
my $send = sub { push @sent, $_[0]; Future->done };
my $receive = sub { Future->new }; # client never disconnects on its own
my $gate = Future->new; # the async cleanup's I/O
my $cleanup_ran = 0;
my $app = async sub {
my ($scope, $receive, $send) = @_;
my $sse = PAGI::SSE->new($scope, $receive, $send);
t/sse/11-async-close.t view on Meta::CPAN
await $sse->close; # app is done; close the stream
};
my $app_f = $app->({ type => 'sse' }, $receive, $send);
# Release the cleanup's I/O so a correctly-awaiting close() can finish.
$gate->done;
my $lived = eval { $app_f->get; 1 };
my $err = $@;
ok($lived, 'SSE app that closes after streaming did not crash')
or diag("app died: $err");
# Guarded by $lived so a resumed orphan coroutine can't fake a green here.
ok($lived && $cleanup_ran, 'async on_close cleanup completed as part of close()');
};
done_testing;
( run in 1.459 second using v1.01-cache-2.11-cpan-995e09ba956 )