``````package Mojo::Content;
use Mojo::Base 'Mojo::EventEmitter';

use Carp qw(croak);
use Compress::Raw::Zlib qw(WANT_GZIP Z_STREAM_END);
use Scalar::Util qw(looks_like_number);

has [qw(auto_decompress auto_relax relaxed skip_body)];
has max_buffer_size   => sub { \$ENV{MOJO_MAX_BUFFER_SIZE}   || 262144 };
has max_leftover_size => sub { \$ENV{MOJO_MAX_LEFTOVER_SIZE} || 262144 };

my \$BOUNDARY_RE = qr!multipart.*boundary\s*=\s*(?:"([^"]+)"|([\w'(),.:?\-+/]+))!i;

sub body_contains { croak 'Method "body_contains" not implemented by subclass' }
sub body_size     { croak 'Method "body_size" not implemented by subclass' }

sub boundary { (shift->headers->content_type // '') =~ \$BOUNDARY_RE ? \$1 // \$2 : undef }

sub charset {
my \$type = shift->headers->content_type // '';
return \$type =~ /charset\s*=\s*"?([^"\s;]+)"?/i ? \$1 : undef;
}

sub clone {
my \$self = shift;
return undef if \$self->is_dynamic;
}

sub generate_body_chunk {
my (\$self, \$offset) = @_;

\$self->emit(drain => \$offset) unless length(\$self->{body_buffer} //= '');
return delete \$self->{body_buffer} if length \$self->{body_buffer};
return ''                          if \$self->{eof};

return looks_like_number \$len && \$len == \$offset ? '' : undef;
}

sub get_body_chunk { croak 'Method "get_body_chunk" not implemented by subclass' }

sub is_compressed { lc(shift->headers->content_encoding // '') eq 'gzip' }

sub is_dynamic { !!\$_{dynamic} }

sub is_finished { (shift->{state} // '') eq 'finished' }

sub is_limit_exceeded { !!shift->{limit} }

sub is_multipart {undef}

sub is_parsing_body { (shift->{state} // '') eq 'body' }

sub leftovers { shift->{buffer} }

sub parse {
my \$self = shift;

\$self->_parse_until_body(@_);
return \$self if \$self->{state} eq 'headers';

# Chunked content
\$self->{real_size} //= 0;
if (\$self->is_chunked && \$self->{state} ne 'headers') {
\$self->_parse_chunked;
\$self->{state} = 'finished' if (\$self->{chunk_state} // '') eq 'finished';
}

# Not chunked, pass through to second buffer
else {
\$self->{real_size} += length \$self->{pre_buffer};
my \$limit = \$self->is_finished && length(\$self->{buffer}) > \$self->max_leftover_size;
\$self->{buffer} .= \$self->{pre_buffer} unless \$limit;
\$self->{pre_buffer} = '';
}

# No content
if (\$self->skip_body) {
\$self->{state} = 'finished';
return \$self;
}

# Relaxed parsing
my \$len     = \$headers->content_length // '';
if (\$self->auto_relax && !length \$len) {
my \$connection = lc(\$headers->connection // '');
\$self->relaxed(1) if \$connection eq 'close' || !\$connection;
}

# Chunked or relaxed content
if (\$self->is_chunked || \$self->relaxed) {
\$self->_decompress(\$self->{buffer} //= '');
\$self->{size} += length \$self->{buffer};
\$self->{buffer} = '';
return \$self;
}

# Normal content
\$len = 0 unless looks_like_number \$len;
if ((my \$need = \$len - (\$self->{size} ||= 0)) > 0) {
my \$len   = length \$self->{buffer};
my \$chunk = substr \$self->{buffer}, 0, \$need > \$len ? \$len : \$need, '';
\$self->_decompress(\$chunk);
\$self->{size} += length \$chunk;
}
\$self->{state} = 'finished' if \$len <= \$self->progress;

return \$self;
}

sub parse_body {
my \$self = shift;
\$self->{state} = 'body';
return \$self->parse(@_);
}

sub progress {
my \$self = shift;
return 0 unless my \$state = \$self->{state};
return 0 unless \$state eq 'body' || \$state eq 'finished';
return \$self->{raw_size} - (\$self->{header_size} || 0);
}

sub write {
my (\$self, \$chunk, \$cb) = @_;

\$self->{dynamic} = 1;
\$self->{body_buffer} .= \$chunk if defined \$chunk;
\$self->once(drain => \$cb)      if \$cb;
\$self->{eof} = 1               if defined \$chunk && !length \$chunk;

return \$self;
}

sub write_chunk {
my (\$self, \$chunk, \$cb) = @_;

@{\$self}{qw(chunked dynamic)} = (1, 1);

\$self->{body_buffer} .= \$self->_build_chunk(\$chunk) if defined \$chunk;
\$self->once(drain => \$cb)                           if \$cb;
\$self->{eof} = 1                                    if defined \$chunk && !length \$chunk;

return \$self;
}

sub _build_chunk {
my (\$self, \$chunk) = @_;

# End
return "\x0d\x0a0\x0d\x0a\x0d\x0a" unless length \$chunk;

# First chunk has no leading CRLF
my \$crlf = \$self->{chunks}++ ? "\x0d\x0a" : '';
return \$crlf . sprintf('%x', length \$chunk) . "\x0d\x0a\$chunk";
}

sub _decompress {
my (\$self, \$chunk) = @_;

# No compression
return \$self->emit(read => \$chunk) unless \$self->auto_decompress && \$self->is_compressed;

# Decompress
\$self->{post_buffer} .= \$chunk;
my \$gz     = \$self->{gz} //= Compress::Raw::Zlib::Inflate->new(WindowBits => WANT_GZIP);
my \$status = \$gz->inflate(\\$self->{post_buffer}, my \$out);
\$self->emit(read => \$out) if defined \$out;

# Replace Content-Encoding with Content-Length

# Check buffer size
@\$self{qw(state limit)} = ('finished', 1) if length(\$self->{post_buffer} // '') > \$self->max_buffer_size;
}

my \$self = shift;
return \$self;
}

sub _parse_chunked {
my \$self = shift;

while (my \$len = length \$self->{pre_buffer}) {

# Start new chunk (ignore the chunk extension)
unless (\$self->{chunk_len}) {
last unless \$self->{pre_buffer} =~ s/^(?:\x0d?\x0a)?([0-9a-fA-F]+).*\x0a//;
next if \$self->{chunk_len} = hex \$1;

# Last chunk
last;
}

# Remove as much as possible from payload
\$len = \$self->{chunk_len} if \$self->{chunk_len} < \$len;
\$self->{buffer} .= substr \$self->{pre_buffer}, 0, \$len, '';
\$self->{real_size} += \$len;
\$self->{chunk_len} -= \$len;
}

# Check buffer size
@\$self{qw(state limit)} = ('finished', 1) if length(\$self->{pre_buffer} // '') > \$self->max_buffer_size;
}

my \$self = shift;

\$self->{chunk_state} = 'finished';

# Take care of leftover and replace Transfer-Encoding with Content-Length
}

my \$self = shift;

\$self->{state} = 'body';

# Take care of leftovers
my \$leftovers = \$self->{pre_buffer} = \$headers->leftovers;
\$self->{header_size} = \$self->{raw_size} - length \$leftovers;
}

sub _parse_until_body {
my (\$self, \$chunk) = @_;

\$self->{raw_size} += length(\$chunk //= '');
\$self->{pre_buffer} .= \$chunk;
\$self->emit('body')   if \$self->{state} ne 'headers' && !\$self->{body}++;
}

1;

=encoding utf8

Mojo::Content - HTTP content base class

package Mojo::Content::MyContent;
use Mojo::Base 'Mojo::Content';

sub body_contains  {...}
sub body_size      {...}
sub get_body_chunk {...}

L<Mojo::Content> is an abstract base class for HTTP content containers, based on L<RFC
7230|https://tools.ietf.org/html/rfc7230> and L<RFC 7231|https://tools.ietf.org/html/rfc7231>, like
L<Mojo::Content::MultiPart> and L<Mojo::Content::Single>.

L<Mojo::Content> inherits all events from L<Mojo::EventEmitter> and can emit the following new ones.

\$content->on(body => sub (\$content) {...});

Emitted once all headers have been parsed and the body starts.

\$content->on(body => sub (\$content) {
});

\$content->on(drain => sub (\$content, \$offset) {...});

Emitted once all data has been written.

\$content->on(drain => sub (\$content) {
\$content->write_chunk(time);
});

\$content->on(read => sub (\$content, \$bytes) {...});

Emitted when a new chunk of content arrives.

\$content->on(read => sub (\$content, \$bytes) {
say "Streaming: \$bytes";
});

L<Mojo::Content> implements the following attributes.

my \$bool = \$content->auto_decompress;
\$content = \$content->auto_decompress(\$bool);

Decompress content automatically if L</"is_compressed"> is true.

my \$bool = \$content->auto_relax;
\$content = \$content->auto_relax(\$bool);

Try to detect when relaxed parsing is necessary.

my \$size = \$content->max_buffer_size;
\$content = \$content->max_buffer_size(1024);

Maximum size in bytes of buffer for content parser, defaults to the value of the C<MOJO_MAX_BUFFER_SIZE> environment
variable or C<262144> (256KiB).

my \$size = \$content->max_leftover_size;
\$content = \$content->max_leftover_size(1024);

Maximum size in bytes of buffer for pipelined HTTP requests, defaults to the value of the C<MOJO_MAX_LEFTOVER_SIZE>
environment variable or C<262144> (256KiB).

my \$bool = \$content->relaxed;
\$content = \$content->relaxed(\$bool);

Activate relaxed parsing for responses that are terminated with a connection close.

my \$bool = \$content->skip_body;
\$content = \$content->skip_body(\$bool);

Skip body parsing and finish after headers.

L<Mojo::Content> inherits all methods from L<Mojo::EventEmitter> and implements the following new ones.

my \$bool = \$content->body_contains('foo bar baz');

Check if content contains a specific string. Meant to be overloaded in a subclass.

my \$size = \$content->body_size;

Content size in bytes. Meant to be overloaded in a subclass.

my \$boundary = \$content->boundary;

Extract multipart boundary from C<Content-Type> header.

my \$charset = \$content->charset;

my \$clone = \$content->clone;

Return a new L<Mojo::Content> object cloned from this content if possible, otherwise return C<undef>.

my \$bytes = \$content->generate_body_chunk(0);

Generate dynamic content.

my \$bytes = \$content->get_body_chunk(0);

Get a chunk of content starting from a specific position. Meant to be overloaded in a subclass.

Get a chunk of the headers starting from a specific position. Note that this method finalizes the content.

Size of headers in bytes. Note that this method finalizes the content.

my \$bool = \$content->headers_contain('foo bar baz');

Check if headers contain a specific string. Note that this method finalizes the content.

my \$bool = \$content->is_chunked;

Check if C<Transfer-Encoding> header indicates chunked transfer encoding.

my \$bool = \$content->is_compressed;

Check C<Content-Encoding> header for C<gzip> value.

my \$bool = \$content->is_dynamic;

Check if content will be dynamically generated, which prevents L</"clone"> from working.

my \$bool = \$content->is_finished;

Check if parser is finished.

my \$bool = \$content->is_limit_exceeded;

Check if buffer has exceeded L</"max_buffer_size">.

my \$bool = \$content->is_multipart;

False, this is not a L<Mojo::Content::MultiPart> object.

my \$bool = \$content->is_parsing_body;

Check if body parsing started yet.

my \$bytes = \$content->leftovers;

Get leftover data from content parser.

\$content
= \$content->parse("Content-Length: 12\x0d\x0a\x0d\x0aHello World!");

Parse content chunk.

\$content = \$content->parse_body('Hi!');

Parse body chunk and skip headers.

my \$size = \$content->progress;

\$content = \$content->write;
\$content = \$content->write('');
\$content = \$content->write(\$bytes);
\$content = \$content->write(\$bytes => sub {...});

Write dynamic content non-blocking, the optional drain callback will be executed once all data has been written.
Calling this method without a chunk of data will finalize the L</"headers"> and allow for dynamic content to be written
later. You can write an empty chunk of data at any time to end the stream.

# Make sure previous chunk of data has been written before continuing
\$content->write('He' => sub (\$content) {
\$content->write('llo!' => sub (\$content) {
\$content->write('');
});
});

\$content = \$content->write_chunk;
\$content = \$content->write_chunk('');
\$content = \$content->write_chunk(\$bytes);
\$content = \$content->write_chunk(\$bytes => sub {...});

Write dynamic content non-blocking with chunked transfer encoding, the optional drain callback will be executed once
all data has been written. Calling this method without a chunk of data will finalize the L</"headers"> and allow for
dynamic content to be written later. You can write an empty chunk of data at any time to end the stream.

# Make sure previous chunk of data has been written before continuing
\$content->write_chunk('He' => sub (\$content) {
\$content->write_chunk('llo!' => sub (\$content) {
\$content->write_chunk('');
});
});
{"ft_lang_label":"__label__en","ft_lang_prob":0.5043257,"math_prob":0.6243572,"size":14002,"snap":"2021-43-2021-49","text_gpt3_token_len":3820,"char_repetition_ratio":0.20852979,"word_repetition_ratio":0.1571346,"special_character_ratio":0.3230967,"punctuation_ratio":0.18449198,"nsfw_num_words":0,"has_unicode_error":false,"math_prob_llama3":0.9503196,"pos_list":[0],"im_url_duplicate_count":[null],"WARC_HEADER":"{\"WARC-Type\":\"response\",\"WARC-Date\":\"2021-10-19T10:03:52Z\",\"WARC-Record-ID\":\"<urn:uuid:cfefe14c-8d5a-4c92-a248-fe5c28d36726>\",\"Content-Length\":\"32063\",\"Content-Type\":\"application/http; msgtype=response\",\"WARC-Warcinfo-ID\":\"<urn:uuid:7028c568-c218-418e-ad0d-0f6351c05673>\",\"WARC-Concurrent-To\":\"<urn:uuid:691995c7-f7bc-4ec6-be3c-1f464646f9eb>\",\"WARC-IP-Address\":\"151.101.250.217\",\"WARC-Target-URI\":\"https://web-stage.metacpan.org/release/SRI/Mojolicious-9.19/source/lib/Mojo/Content.pm\",\"WARC-Payload-Digest\":\"sha1:HCQXN2FEUKCYD37QT7T5ETDL2VR5AH37\",\"WARC-Block-Digest\":\"sha1:7TLYNCT2BYB7S4KXSBTHCMKQKLVLA6DT\",\"WARC-Identified-Payload-Type\":\"text/html\",\"warc_filename\":\"/cc_download/warc_2021/CC-MAIN-2021-43/CC-MAIN-2021-43_segments_1634323585246.50_warc_CC-MAIN-20211019074128-20211019104128-00454.warc.gz\"}"}