repobrowse: stream git patch output with Danga::Socket

This should allow better concurrency in case git-format-patch
needs to take a long time for merge commits.

diff --git a/lib/PublicInbox/RepobrowseGitPatch.pm b/lib/PublicInbox/RepobrowseGitPatch.pm
index b67fb86f0b34cfcbee740c3f45459cbaeb954211..b3cf17fe65624e97f3732ab6da5fc6a1c1a4432a 100644 (file)
--- a/lib/PublicInbox/RepobrowseGitPatch.pm
+++ b/lib/PublicInbox/RepobrowseGitPatch.pm
@@ -26,22 +26,54 @@ sub call_git_patch {
        if (defined(my $expath = $req->{expath})) {
                push @cmd, $expath;
        }
-       my $fp = $git->popen(@cmd);
-       my ($buf, $n);
+       my $rpipe = $git->popen(@cmd);
+       my $env = $req->{cgi}->env;
+       my $err = $env->{'psgi.errors'};
+       my ($buf, $n, $res, $vin, $fh);
+       my $end = sub {
+               if ($fh) {
+                       $fh->close;
+                       $fh = undef;
+               } elsif ($res) {
+                       $res->($self->r(500));
+               }
+               if ($rpipe) {
+                       $rpipe->close; # _may_ be Danga::Socket::close
+                       $rpipe = undef;
+               }
+       };
+       my $fail = sub {
+               if ($!{EAGAIN} || $!{EINTR}) {
+                       select($vin, undef, undef, undef) if defined $vin;
+                       # $vin is undef on async, so this is a noop on EAGAIN
+                       return;
+               }
+               my $e = $!;
+               $end->();
+               $err->print("git format-patch ($git->{git_dir}): $e\n");
+       };
+       my $cb = sub {
+               $n = $rpipe->sysread($buf, 8192);
+               return $fail->() unless defined $n;
+               return $end->() if $n == 0;
+               if ($res) {
+                       my $h = ['Content-Type', 'text/plain; charset=UTF-8'];
+                       $fh = $res->([200, $h]);
+                       $res = undef;
+               }
+               $fh->write($buf) if $fh;
+       };
 
-       $n = read($fp, $buf, 8192);
-       return unless (defined $n && $n > 0);
-       sub {
-               my ($res) = @_; # Plack callback
-               my $fh = $res->([200, [
-                       'Content-Type' => 'text/plain; charset=UTF-8']]);
-               $fh->write($buf);
-               while (1) {
-                       $n = read($fp, $buf, 8192);
-                       last unless (defined $n && $n > 0);
-                       $fh->write($buf);
+       if (my $async = $env->{'pi-httpd.async'}) {
+               $rpipe = $async->($rpipe, $cb);
+               sub { ($res) = @_ } # let Danga::Socket handle the rest.
+       } else { # synchronous loop for other PSGI servers
+               $vin = '';
+               vec($vin, fileno($rpipe), 1) = 1;
+               sub {
+                       ($res) = @_;
+                       while ($rpipe) { $cb->() }
                }
-               $fh->close;
        }
 }