view release on metacpan or  search on metacpan

t/request-body-stream.t  view on Meta::CPAN

        { 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';

    my $chunk2 = $stream->next_chunk->get;
    is $chunk2, ' World', 'second chunk';
    ok $stream->is_done, 'done after last chunk';

    is $stream->bytes_read, 11, 'bytes_read correct';
};

subtest 'max_bytes limit' => sub {
    my $receive = mock_receive(
        { type => 'http.request', body => 'Hello World', more => 0 },
    );

    my $stream = PAGI::Request::BodyStream->new(
        receive => $receive,
        max_bytes => 5,
    );

    like dies { $stream->next_chunk->get }, qr/max_bytes exceeded/, 'throws on limit exceeded';
};

subtest 'UTF-8 decoding' => sub {
    my $receive = mock_receive(
        { type => 'http.request', body => "caf\xc3\xa9", more => 0 },
    );

    my $stream = PAGI::Request::BodyStream->new(
        receive => $receive,
        decode => 'UTF-8',
    );

    my $chunk = $stream->next_chunk->get;
    is $chunk, "café", 'UTF-8 decoded';
};

subtest 'disconnect handling' => sub {
    my $receive = mock_receive(
        { type => 'http.request', body => 'Hello', more => 1 },
        { type => 'http.disconnect' },
    );

    my $stream = PAGI::Request::BodyStream->new(receive => $receive);

    my $chunk1 = $stream->next_chunk->get;
    is $chunk1, 'Hello', 'first chunk';

    my $chunk2 = $stream->next_chunk->get;
    is $chunk2, undef, 'undef on disconnect';
    ok $stream->is_done, 'done after disconnect';
};

subtest 'UTF-8 boundary handling - split multi-byte sequence' => sub {
    # Split café (caf\xc3\xa9) across chunks: "caf\xc3" and "\xa9"
    my $receive = mock_receive(
        { type => 'http.request', body => "caf\xc3", more => 1 },
        { type => 'http.request', body => "\xa9", more => 0 },
    );

    my $stream = PAGI::Request::BodyStream->new(
        receive => $receive,
        decode => 'UTF-8',
    );

    my $chunk1 = $stream->next_chunk->get;
    is $chunk1, "caf", 'first chunk without incomplete sequence';

    my $chunk2 = $stream->next_chunk->get;
    is $chunk2, "é", 'second chunk completes the sequence';

    ok $stream->is_done, 'stream done';
};

subtest 'empty chunks are handled' => sub {
    my $receive = mock_receive(
        { type => 'http.request', body => '', more => 1 },
        { type => 'http.request', body => 'data', more => 1 },
        { type => 'http.request', body => '', more => 0 },
    );

    my $stream = PAGI::Request::BodyStream->new(receive => $receive);

    my $chunk1 = $stream->next_chunk->get;
    is $chunk1, '', 'empty first chunk';

    my $chunk2 = $stream->next_chunk->get;
    is $chunk2, 'data', 'non-empty chunk';

    my $chunk3 = $stream->next_chunk->get;
    is $chunk3, '', 'empty last chunk';

    ok $stream->is_done, 'stream done';
    is $stream->bytes_read, 4, 'bytes_read only counts non-empty';
};

subtest 'stream_to custom sink' => 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 $collected = '';
    my $bytes = $stream->stream_to(sub {
        my ($chunk) = @_;
        $collected .= $chunk;
    })->get;

    is $collected, 'Hello World', 'all chunks collected';
    is $bytes, 11, 'bytes_processed correct';
    ok $stream->is_done, 'stream done';
};