use HTTP::Date qw(time2str);
use IO::Handle;
require PublicInbox::EvCleanup;
-PublicInbox::DS->import(qw(msg_more write_in_full));
+PublicInbox::DS->import(qw(msg_more));
use PublicInbox::Syscall qw(EPOLLIN EPOLLONESHOT);
use constant {
CHUNK_START => -1, # [a-f0-9]+\r\n
CHUNK_ZEND => -3, # \r\n
CHUNK_MAX_HDR => 256,
};
+use Errno qw(EAGAIN);
my $pipelineq = [];
my $pipet;
# otherwise we can be buffering infinitely w/o backpressure
return read_input($self) if defined $self->{env};
-
- my $off = length($self->{rbuf});
- my $r = sysread($self->{sock}, $self->{rbuf}, 8192, $off);
- if (defined $r) {
- return $self->close if $r == 0;
- return rbuf_process($self);
- }
-
- return $self->watch_in1 if $!{EAGAIN};
-
- # common for clients to break connections without warning,
- # would be too noisy to log here:
- return $self->close;
+ my $rbuf = \($self->{rbuf});
+ my $off = bytes::length($$rbuf);
+ $self->do_read($rbuf, 8192, $off) and rbuf_process($self);
}
sub rbuf_process {
# (they are rarely-used and git (as of 2.7.2) does not use them)
if ($r == -1 || $env{HTTP_TRAILER} ||
# this length-check is necessary for PURE_PERL=1:
- ($r == -2 && length($self->{rbuf}) > 0x4000)) {
+ ($r == -2 && bytes::length($self->{rbuf}) > 0x4000)) {
return quit($self, 400);
}
return $self->watch_in1 if $r < 0; # incomplete
$len ? read_input($self) : app_dispatch($self);
}
+# IO::Handle::write returns boolean, this returns bytes written:
+sub xwrite ($$$) {
+ my ($fh, $rbuf, $max) = @_;
+ my $w = bytes::length($$rbuf);
+ $w = $max if $w > $max;
+ $fh->write($$rbuf, $w) or return;
+ $w;
+}
+
sub read_input ($) {
my ($self) = @_;
my $env = $self->{env};
# env->{CONTENT_LENGTH} (identity)
my $sock = $self->{sock};
- my $len = $self->{input_left};
- $self->{input_left} = undef;
+ my $len = delete $self->{input_left};
my $rbuf = \($self->{rbuf});
my $input = $env->{'psgi.input'};
while ($len > 0) {
if ($$rbuf ne '') {
- my $w = write_in_full($input, $rbuf, $len, 0);
+ my $w = xwrite($input, $rbuf, $len);
return write_err($self, $len) unless $w;
$len -= $w;
die "BUG: $len < 0 (w=$w)" if $len < 0;
sub response_done_cb ($$) {
my ($self, $alive) = @_;
sub {
- my $env = $self->{env};
- $self->{env} = undef;
+ my $env = delete $self->{env};
$self->write(\"0\r\n\r\n") if $alive == 2;
- $self->write(sub{$alive ? next_request($self) : $self->close});
+ $self->write($alive ? \&next_request : \&close);
}
}
}
}
- $self->{forward} = $self->{pull} = undef;
+ delete @$self{qw(forward pull)};
# avoid recursion
if ($forward) {
eval { $forward->close };
}
}
+sub input_tmpfile ($) {
+ open($_[0], '+>', undef);
+ $_[0]->autoflush(1);
+}
+
sub input_prepare {
my ($self, $env) = @_;
my $input;
quit($self, 413);
return;
}
- open($input, '+>', undef);
+ input_tmpfile($input);
} elsif (env_chunked($env)) {
$len = CHUNK_START;
- open($input, '+>', undef);
+ input_tmpfile($input);
} else {
$input = $null_io;
}
sub recv_err {
my ($self, $r, $len) = @_;
return $self->close if (defined $r && $r == 0);
- if ($!{EAGAIN}) {
+ if ($! == EAGAIN) {
$self->{input_left} = $len;
return $self->watch_in1;
}
my ($self) = @_;
my $input = $self->{env}->{'psgi.input'};
my $sock = $self->{sock};
- my $len = $self->{input_left};
- $self->{input_left} = undef;
+ my $len = delete $self->{input_left};
my $rbuf = \($self->{rbuf});
while (1) { # chunk start
if ($len == CHUNK_ZEND) {
$$rbuf =~ s/\A\r\n//s and
return app_dispatch($self, $input);
- return quit($self, 400) if length($$rbuf) > 2;
+ return quit($self, 400) if bytes::length($$rbuf) > 2;
}
if ($len == CHUNK_END) {
if ($$rbuf =~ s/\A\r\n//s) {
$len = CHUNK_START;
- } elsif (length($$rbuf) > 2) {
+ } elsif (bytes::length($$rbuf) > 2) {
return quit($self, 400);
}
}
if (($len + -s $input) > $MAX_REQUEST_BUFFER) {
return quit($self, 413);
}
- } elsif (length($$rbuf) > CHUNK_MAX_HDR) {
+ } elsif (bytes::length($$rbuf) > CHUNK_MAX_HDR) {
return quit($self, 400);
}
# will break from loop since $len >= 0
}
if ($len < 0) { # chunk header is trickled, read more
- my $off = length($$rbuf);
+ my $off = bytes::length($$rbuf);
my $r = sysread($sock, $$rbuf, 8192, $off);
return recv_err($self, $r, $len) unless $r;
# (implicit) goto chunk_start if $r > 0;
# drain the current chunk
until ($len <= 0) {
if ($$rbuf ne '') {
- my $w = write_in_full($input, $rbuf, $len, 0);
+ my $w = xwrite($input, $rbuf, $len);
return write_err($self, "$len chunk") if !$w;
$len -= $w;
if ($len == 0) {
}
sub close {
- my $self = shift;
- my $forward = $self->{forward};
- my $env = $self->{env};
- delete $env->{'psgix.io'} if $env; # prevent circular references
- $self->{pull} = $self->{forward} = $self->{env} = undef;
- if ($forward) {
+ my $self = $_[0];
+ if (my $env = delete $self->{env}) {
+ delete $env->{'psgix.io'}; # prevent circular references
+ }
+ delete $self->{pull};
+ if (my $forward = delete $self->{forward}) {
eval { $forward->close };
err($self, "forward ->close error: $@") if $@;
}
- $self->SUPER::close(@_);
+ $self->SUPER::close; # PublicInbox::DS::close
}
# for graceful shutdown in PublicInbox::Daemon:
($self->{rbuf} ne '' || $self->{env} || $self->{wbuf});
}
+# fires after pending writes are complete:
+sub restart_pass ($) {
+ $_[0]->{forward}->restart_read; # see PublicInbox::HTTPD::Async
+}
+
+sub enqueue_restart_pass ($) { $_[0]->write(\&restart_pass) }
+
1;