use strict;
use warnings;
use PublicInbox::Spawn qw(popen_rd);
+my $def_limiter;
sub new ($$$;) {
my ($class, $cmd, $env, $opt) = @_;
}
}
+sub psgi_return {
+ my ($self, $env, $limiter, $parse_hdr) = @_;
+ my ($fh, $rpipe);
+ my $end = sub {
+ if (my $err = $self->finish) {
+ $err = join(' ', @{$self->{args}->{cmd}}).": $err\n";
+ $env->{'psgi.errors'}->print($err);
+ }
+ $fh->close if $fh; # async-only
+ };
+
+ # Danga::Socket users, we queue up the read_enable callback to
+ # fire after pending writes are complete:
+ my $buf = '';
+ my $rd_hdr = sub {
+ my $r = sysread($rpipe, $buf, 1024, length($buf));
+ return if !defined($r) && ($!{EINTR} || $!{EAGAIN});
+ $parse_hdr->($r, $buf);
+ };
+ my $res;
+ my $async = $env->{'pi-httpd.async'};
+ my $cb = sub {
+ my $r = $rd_hdr->() or return;
+ $rd_hdr = undef;
+ if (scalar(@$r) == 3) { # error
+ if ($async) {
+ $async->close; # calls rpipe->close
+ } else {
+ $rpipe->close;
+ $end->();
+ }
+ $res->($r);
+ } elsif ($async) {
+ $fh = $res->($r); # scalar @$r == 2
+ $async->async_pass($env->{'psgix.io'}, $fh, \$buf);
+ } else { # for synchronous PSGI servers
+ require PublicInbox::GetlineBody;
+ $r->[2] = PublicInbox::GetlineBody->new($rpipe, $end,
+ $buf);
+ $res->($r);
+ }
+ };
+ $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
+ sub {
+ ($res) = @_;
+ $self->start($limiter, sub { # may run later, much later...
+ ($rpipe) = @_;
+ if ($async) {
+ # PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
+ $async = $async->($rpipe, $cb, $end);
+ } else { # generic PSGI
+ $cb->() while $rd_hdr;
+ }
+ });
+ };
+}
+
package PublicInbox::Qspawn::Limiter;
use strict;
use warnings;
use strict;
use warnings;
use base qw(PublicInbox::RepobrowseBase);
+use PublicInbox::Qspawn;
# try to be educational and show the command-line used in the signature
my @CMD = qw(format-patch -M --stdout);
# limit scope, don't take extra args to avoid wasting server
# resources buffering:
my $range = "$id~1..$id^0";
- my @cmd = (@CMD, $sig." $range", $range, '--');
+ my @cmd = ('git', "--git-dir=$git->{git_dir}", @CMD,
+ $sig." $range", $range, '--');
if (defined(my $expath = $req->{expath})) {
- push @cmd, $expath;
+ push @cmd, $expath if $expath ne '';
}
- my $rpipe = $git->popen(@cmd);
- my $err = $env->{'psgi.errors'};
- my ($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(my $buf, 65536);
- 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 (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->() }
- }
- }
+ # FIXME: generalize this with other qspawn users
+ my $qsp = PublicInbox::Qspawn->new(\@cmd);
+ $qsp->psgi_return($env, undef, sub {
+ my ($r) = @_;
+ my $h = ['Content-Type', 'text/plain; charset=UTF-8'];
+ $r ? [ 200, $h ] : [ 500, $h, [ "format-patch error\n" ] ];
+ });
}
1;