view release on metacpan or  search on metacpan

t/stream_incremental.t  view on Meta::CPAN

    . 'with all of it, delivered when the daemon closed';
  is_deeply $handler->{summary}->(), { delivered => 3, stopped => 0 },
    'and the summary counts three units rather than one';
};

subtest 'the header block and the first bytes of the body in one arrival'
  => sub {
  # The case that made this a whole-transport change rather than a local one.
  # The head used to be read with <$sock>, which reads ahead: the bytes past
  # the blank line were already inside PerlIO's buffer, where nothing could
  # get at them. A body reader switched to sysread on its own would have
  # dropped exactly this.
  my @got;
  my $fh = burst_handle($HEAD . 'body starts here', ' and continues');
  my $handler = $client->_stream_handler('GET /v1.41/probe', 'on_chunk',
    sub { push @got, $_[0] }, 1);

  $client->_read_streaming_response($fh, 'GET', $handler, {});

  is_deeply \@got, ['body starts here', ' and continues'],
    'the read-ahead past the headers is the start of the body, and it is '
    . 'neither dropped nor delivered twice';
};

subtest 'a frame split across two arrivals is delivered once, whole' => sub {
  # The carry buffer's job, asserted here because karr k60 changes how often
  # it is asked to do it: it now sees the split the daemon actually made
  # rather than 64K blocks.
  my @got;
  my $whole = frame('hello world');
  my $fh = burst_handle($HEAD, substr($whole, 0, 5), substr($whole, 5));
  my $handler = $client->_stream_handler('GET /v1.41/probe', 'on_frame',
    sub { push @got, $_[0] }, 1);

  $client->_read_streaming_response($fh, 'GET', $handler, {});

  is_deeply \@got, [ { stream => 'stdout', data => 'hello world' } ],
    'the 8-byte header was split too, and the frame still came out once';
};

subtest 'a chunked stream still delivers per read, not per chunk' => sub {
  # The chunked path was never the broken one -- it asks for exactly the chunk
  # size, which is exactly what is there -- but it must not have regressed:
  # the engine is free to send an hour of log output as one chunk, so waiting
  # for a whole chunk would be the same bug in a different place.
  my @got;
  my $head = "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n";
  my $fh = burst_handle($head . "12\r\nfirst ", 'second ', "third\r\n",
    "0\r\n\r\n");
  my $handler = $client->_stream_handler('GET /v1.41/probe', 'on_chunk',
    sub { push @got, $_[0] }, 1);

  $client->_read_streaming_response($fh, 'GET', $handler, {});

  is join('', @got), 'first second third', 'the whole chunk arrived';
  cmp_ok scalar @got, '>=', 3,
    'in at least as many calls as it arrived in (' . scalar(@got) . ')';
};

subtest 'one byte per read: every structure boundary is fragmented' => sub {
  # The cheapest way to say "no reader assumes anything arrives whole". At one
  # byte per read the status line, each header line, the blank line, the chunk
  # header, the chunk data, the CRLF after it and the terminating zero chunk
  # are each split across several reads -- and the 8-byte frame header inside
  # the payload is split eight ways.
  my $wire = "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n";
  my $body = frame('one') . frame('two');
  $wire .= sprintf("%x\r\n", length $body) . $body . "\r\n0\r\n\r\n";

  my @got;
  my $fh = burst_handle(split //, $wire);
  my $handler = $client->_stream_handler('GET /v1.41/probe', 'on_frame',
    sub { push @got, $_[0] }, 1);

  $client->_read_streaming_response($fh, 'GET', $handler, {});

  is_deeply \@got,
    [ { stream => 'stdout', data => 'one' },
      { stream => 'stdout', data => 'two' } ],
    'both frames come out whole';

  # And the same wire read whole, so the buffered path is held to it too.
  my $buffered = $client->_read_response(burst_handle(split //, $wire), 'GET');
  is $buffered->[3], $body, 'the buffered reader reassembles it as well';
  is $buffered->[0], 200, 'and got the status line out of one-byte reads too';
};

subtest 'the caller can stop mid-stream with bytes still buffered' => sub {
  my @got;
  my $fh = burst_handle($HEAD, 'first', 'second', 'third');
  my $handler = $client->_stream_handler('GET /v1.41/probe', 'on_chunk',
    sub { push @got, $_[0]; $_[1]->() if @got == 2 }, 1);

  $client->_read_streaming_response($fh, 'GET', $handler, {});

  is_deeply \@got, ['first', 'second'], 'it stopped where it said to';
  is_deeply $handler->{summary}->(), { delivered => 2, stopped => 1 },
    'and said so';
};

# ---------------------------------------------------------------------------
# The real socket. A tied handle proves the loop; only a writer on the other
# end of a real socket proves that a read comes back with what has arrived
# rather than waiting for the rest.
#
# Runs the same scenario -- a peer that answers, writes three framed messages
# 0.15s apart and then closes -- and reports what the callback saw and when.
sub paced_writer {
  my ($option, $cb) = @_;

  socketpair(my $ours, my $theirs, AF_UNIX, SOCK_STREAM, PF_UNSPEC)
    or return;
  my $pid = fork();
  return unless defined $pid;

  unless ($pid) {
    # The daemon side. The pauses are what a coalescing reader swallows.
    close $ours;
    syswrite $theirs, $HEAD;
    for my $text ('one', 'two', 'three') {
      sleep 0.15;