+sub _psgi_finish ($$) {
+ my ($self, $env) = @_;
+ my $err = $self->finish;
+ if ($err && !$env->{'qspawn.quiet'}) {
+ $err = join(' ', @{$self->{args}->[0]}).": $err\n";
+ $env->{'psgi.errors'}->print($err);
+ }
+}
+
+sub psgi_qx {
+ my ($self, $env, $limiter, $qx_cb) = @_;
+ my $qx = PublicInbox::Qspawn::Qx->new;
+ my $end = sub {
+ _psgi_finish($self, $env);
+ eval { $qx_cb->($qx) };
+ $qx = undef;
+ };
+ my $rpipe;
+ my $async = $env->{'pi-httpd.async'};
+ my $cb = sub {
+ my $r = sysread($rpipe, my $buf, 8192);
+ if ($async) {
+ $async->async_pass($env->{'psgix.io'}, $qx, \$buf);
+ } elsif (defined $r) {
+ $r ? $qx->write($buf) : $end->();
+ } else {
+ return if $!{EAGAIN} || $!{EINTR}; # loop again
+ $end->();
+ }
+ };
+ $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
+ $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 $qx;
+ }
+ });
+}
+