PAGI-Tools
view release on metacpan or search on metacpan
lib/PAGI/Request/MultipartStream.pm view on Meta::CPAN
package PAGI::Request::MultipartStream;
$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.
Internally this drives L<HTTP::MultiPartParser> on demand, bridging its
push-based callbacks onto an internal event queue that C<next> and the part
methods consume.
B<Mutually exclusive with the buffered body methods.> An HTTP request body can
only be consumed once. Once you create a multipart stream you cannot also call
C<body>/C<text>/C<json>/C<form_params>/C<uploads>, and a stream cannot be
created if the body was already read; see L<PAGI::Request/multipart_stream>.
=cut
our $MAX_FILES = 1000;
our $MAX_FIELDS = 1000;
our $MAX_FIELD_SIZE = 1024 * 1024; # buffered per-field cap
our $MAX_FILE_SIZE = 100 * 1024 * 1024;
our $MAX_REQUEST_BODY = 1024 * 1024 * 1024; # defense-in-depth; server max_body_size is primary
=head1 CONSTRUCTOR
=head2 new
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.
=item * C<max_fields> - Maximum number of non-file (field) parts. Default: 1000.
=item * C<max_field_size> - Maximum size, in bytes, of any single field part.
Default: 1 MiB (1024 * 1024).
=item * C<max_file_size> - Maximum size, in bytes, of any single file part.
Default: 100 MiB (100 * 1024 * 1024).
=item * C<max_request_body> - Maximum total bytes read from the request body.
Default: 1 GiB (1024 * 1024 * 1024). This is a per-stream defence-in-depth
cap; the PAGI server's C<max_body_size> is the primary aggregate limit on the
request body.
=back
=cut
sub new {
my ($class, %args) = @_;
croak "receive is required" unless $args{receive};
croak "boundary is required" unless defined $args{boundary} && length $args{boundary};
my $self = bless {
receive => $args{receive},
boundary => $args{boundary},
max_files => $args{max_files} // $MAX_FILES,
max_fields => $args{max_fields} // $MAX_FIELDS,
max_field_size => $args{max_field_size} // $MAX_FIELD_SIZE,
max_file_size => $args{max_file_size} // $MAX_FILE_SIZE,
max_request_body => $args{max_request_body} // $MAX_REQUEST_BODY,
_queue => [], # FIFO: ['part',\%meta] | ['body',$chunk]
_file_count => 0,
_field_count => 0,
_bytes_total => 0,
_cur_is_file => 0,
_cur_bytes => 0,
_cur_name => undef,
_current => undef, # current Part
_exhausted => 0,
_parser_finished => 0, # guard: finish() is called at most once
_failed => undef, # sticky failure message (poisons the stream)
}, $class;
$self->{_parser} = $self->_build_parser;
return $self;
}
# Parse the on_header arrayref of header lines into
# {name,filename,content_type,encoding,headers}. is_file := defined(filename).
sub _disposition {
my ($lines) = @_;
lib/PAGI/Request/MultipartStream.pm view on Meta::CPAN
my $part = await $stream->next;
Returns a Future resolving to the next L<PAGI::Request::Part>, or C<undef> when
the stream is exhausted (end of body).
Advancing past a part whose body you have not fully consumed auto-drains the
remainder of that part first, so you can always loop on C<next> without
reading every part. To discard a part deliberately (and signal that intent),
call C<< $part->skip >>.
Croaks if a size or count limit is breached, or if the upload is truncated
(see L</LIMITS AND ERRORS>).
=cut
async sub next {
my ($self) = @_;
croak $self->{_failed} if $self->{_failed};
if ($self->{_current} && !$self->{_current}{_done}) { await $self->{_current}->skip; } # auto-drain
while (1) {
croak $self->{_failed} if $self->{_failed};
shift @{$self->{_queue}} while @{$self->{_queue}} && $self->{_queue}[0][0] eq 'body'; # defensive
if (@{$self->{_queue}} && $self->{_queue}[0][0] eq 'part') {
my (undef, $meta) = @{ shift @{$self->{_queue}} };
$self->{_current} = PAGI::Request::Part->new(stream => $self, meta => $meta);
return $self->{_current};
}
last unless await $self->_pump;
}
croak $self->{_failed} if $self->{_failed}; # truncation surfaces via _failed (set by finish)
return undef;
}
# Next body chunk for the current part: the chunk, or undef when the part ends.
async sub _next_chunk {
my ($self) = @_;
while (1) {
croak $self->{_failed} if $self->{_failed};
if (@{$self->{_queue}}) {
my $kind = $self->{_queue}[0][0];
if ($kind eq 'body') { my $ev = shift @{$self->{_queue}}; return $ev->[1]; }
return undef if $kind eq 'part'; # next part began -> current done
}
if (!(await $self->_pump)) {
croak $self->{_failed} if $self->{_failed}; # truncation surfaces via _failed (set by finish)
return undef; # clean EOF (complete body, then disconnect)
}
}
}
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
the current part automatically.
=head1 CONSTRUCTOR
=head2 new
my $part = PAGI::Request::Part->new(stream => $stream, meta => \%meta);
Constructs a part bound to its owning stream. Parts are normally created by
L<PAGI::Request::MultipartStream/next>, not by application code.
=head1 METHODS
=head2 name
my $name = $part->name;
The part's form field name, taken from its C<Content-Disposition> header.
=head2 filename
my $filename = $part->filename;
The part's filename from C<Content-Disposition>, or C<undef> for non-file
(field) parts.
=head2 content_type
my $type = $part->content_type;
The part's C<Content-Type> header. Defaults to C<text/plain> if the part sent
no C<Content-Type>.
=head2 encoding
my $encoding = $part->encoding;
The part's C<Content-Transfer-Encoding> header, or C<undef> if not present.
=head2 headers
my $headers = $part->headers;
A hashref of all the part's headers, keyed by lower-cased header name.
=head2 is_file
if ($part->is_file) { ... }
True if the part has a filename (i.e. is a file upload), false otherwise.
( run in 0.732 second using v1.01-cache-2.11-cpan-6aa56a78535 )