ds: pass $self to code references

[public-inbox.git] / lib / PublicInbox / HTTP.pm

diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
index afa71ea53f3cb3ddda35d96a11117b104ead6dc8..e132c6105f4ac9f66f495bd21a45753125e0b63c 100644 (file)
--- a/lib/PublicInbox/HTTP.pm
+++ b/lib/PublicInbox/HTTP.pm
@@ -20,12 +20,14 @@ use HTTP::Date qw(time2str);
 use IO::Handle;
 require PublicInbox::EvCleanup;
 PublicInbox::DS->import(qw(msg_more write_in_full));
+use PublicInbox::Syscall qw(EPOLLIN EPOLLONESHOT);
 use constant {
        CHUNK_START => -1,   # [a-f0-9]+\r\n
        CHUNK_END => -2,     # \r\n
        CHUNK_ZEND => -3,    # \r\n
        CHUNK_MAX_HDR => 256,
 };
+use Errno qw(EAGAIN);
 
 my $pipelineq = [];
 my $pipet;
@@ -56,7 +58,7 @@ sub http_date () {
 sub new ($$$) {
        my ($class, $sock, $addr, $httpd) = @_;
        my $self = fields::new($class);
-       $self->SUPER::new($sock, PublicInbox::DS::EPOLLIN());
+       $self->SUPER::new($sock, EPOLLIN | EPOLLONESHOT);
        $self->{httpd} = $httpd;
        $self->{rbuf} = '';
        ($self->{remote_addr}, $self->{remote_port}) =
@@ -80,11 +82,10 @@ sub event_step { # called by PublicInbox::DS
                return $self->close if $r == 0;
                return rbuf_process($self);
        }
-       return if $!{EAGAIN}; # no need to call watch_read(1) again
 
        # common for clients to break connections without warning,
        # would be too noisy to log here:
-       return $self->close;
+       $! == EAGAIN ? $self->watch_in1 : $self->close;
 }
 
 sub rbuf_process {
@@ -100,7 +101,7 @@ sub rbuf_process {
                        ($r == -2 && length($self->{rbuf}) > 0x4000)) {
                return quit($self, 400);
        }
-       return $self->watch_read(1) if $r < 0; # incomplete
+       return $self->watch_in1 if $r < 0; # incomplete
        $self->{rbuf} = substr($self->{rbuf}, $r);
 
        my $len = input_prepare($self, \%env);
@@ -117,8 +118,7 @@ sub read_input ($) {
 
        # 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'};
 
@@ -143,7 +143,6 @@ sub read_input ($) {
 
 sub app_dispatch {
        my ($self, $input) = @_;
-       $self->watch_read(0);
        my $env = $self->{env};
        $env->{REMOTE_ADDR} = $self->{remote_addr};
        $env->{REMOTE_PORT} = $self->{remote_port};
@@ -236,7 +235,7 @@ sub identity_wcb ($) {
 sub next_request ($) {
        my ($self) = @_;
        if ($self->{rbuf} eq '') { # wait for next request
-               $self->watch_read(1);
+               $self->watch_in1;
        } else { # avoid recursion for pipelined requests
                push @$pipelineq, $self;
                $pipet ||= PublicInbox::EvCleanup::asap(*process_pipelineq);
@@ -246,10 +245,9 @@ sub next_request ($) {
 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);
        }
 }
 
@@ -279,7 +277,7 @@ sub getline_cb ($$$) {
                }
        }
 
-       $self->{forward} = $self->{pull} = undef;
+       delete @$self{qw(forward pull)};
        # avoid recursion
        if ($forward) {
                eval { $forward->close };
@@ -358,9 +356,9 @@ sub write_err {
 sub recv_err {
        my ($self, $r, $len) = @_;
        return $self->close if (defined $r && $r == 0);
-       if ($!{EAGAIN}) {
+       if ($! == EAGAIN) {
                $self->{input_left} = $len;
-               return;
+               return $self->watch_in1;
        }
        err($self, "error reading for input: $! ($len bytes remaining)");
        quit($self, 500);
@@ -370,8 +368,7 @@ sub read_input_chunked { # unlikely...
        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
@@ -441,16 +438,16 @@ sub quit {
 }
 
 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:
@@ -459,4 +456,11 @@ sub busy () {
        ($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;