qspawn: disambiguate command vs PSGI env

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

diff --git a/lib/PublicInbox/Qspawn.pm b/lib/PublicInbox/Qspawn.pm
index cb3dc516304d74e70467c8ba0aac43e8d0004c66..ba980e7397f4510c874cf5ea0da101ac61f816b5 100644 (file)
--- a/lib/PublicInbox/Qspawn.pm
+++ b/lib/PublicInbox/Qspawn.pm
@@ -36,17 +36,17 @@ my $def_limiter;
 
 # declares a command to spawn (but does not spawn it).
 # $cmd is the command to spawn
-# $env is the environ for the child process
+# $cmd_env is the environ for the child process (not PSGI env)
 # $opt can include redirects and perhaps other process spawning options
 sub new ($$$;) {
-       my ($class, $cmd, $env, $opt) = @_;
-       bless { args => [ $cmd, $env, $opt ] }, $class;
+       my ($class, $cmd, $cmd_env, $opt) = @_;
+       bless { args => [ $cmd, $cmd_env, $opt ] }, $class;
 }
 
 sub _do_spawn {
        my ($self, $start_cb, $limiter) = @_;
        my $err;
-       my ($cmd, $env, $opts) = @{$self->{args}};
+       my ($cmd, $cmd_env, $opts) = @{$self->{args}};
        my %opts = %{$opts || {}};
        $self->{limiter} = $limiter;
        foreach my $k (PublicInbox::Spawn::RLIMITS()) {
@@ -55,7 +55,7 @@ sub _do_spawn {
                }
        }
 
-       ($self->{rpipe}, $self->{pid}) = popen_rd($cmd, $env, \%opts);
+       ($self->{rpipe}, $self->{pid}) = popen_rd($cmd, $cmd_env, \%opts);
 
        # drop any IO handles opt was holding open via $opt->{hold}
        # No need to hold onto the descriptor once the child process has it.
@@ -94,7 +94,7 @@ sub waitpid_err ($$) {
                $err = "W: waitpid($xpid, 0) => $pid: $!";
        } # else should not be called with pid == 0
 
-       my $env = delete $self->{env};
+       my $env = delete $self->{psgi_env};
 
        # done, spawn whatever's in the queue
        my $limiter = $self->{limiter};
@@ -106,17 +106,21 @@ sub waitpid_err ($$) {
                }
        }
 
-       return unless $err;
-       $self->{err} = $err;
-       if ($env && !$env->{'qspawn.quiet'}) {
-               log_err($env, join(' ', @{$self->{args}}) . ": $err");
+       if ($err) {
+               $self->{err} = $err;
+               if ($env && !$env->{'qspawn.quiet'}) {
+                       log_err($env, join(' ', @{$self->{args}}) . ": $err");
+               }
+       }
+       if (my $fin_cb = delete $self->{fin_cb}) {
+               eval { $fin_cb->() }
        }
 }
 
-sub do_waitpid ($;$) {
-       my ($self, $env) = @_;
+sub do_waitpid ($;$$) {
+       my ($self, $fin_cb) = @_;
        my $pid = $self->{pid};
-       $self->{env} = $env;
+       $self->{fin_cb} = $fin_cb;
        # PublicInbox::DS may not be loaded
        eval { PublicInbox::DS::dwaitpid($pid, \&waitpid_err, $self) };
        # done if we're running in PublicInbox::DS::EventLoop
@@ -128,9 +132,11 @@ sub do_waitpid ($;$) {
 }
 
 sub finish ($;$) {
-       my ($self, $env) = @_;
+       my ($self, $fin_cb) = @_;
        if (delete $self->{rpipe}) {
-               do_waitpid($self, $env);
+               do_waitpid($self, $fin_cb);
+       } elsif ($fin_cb) {
+               eval { $fin_cb->() };
        }
 }
 
@@ -148,15 +154,15 @@ sub start {
 # $env is the PSGI env.  As with ``/qx; only use this when output is small
 # and safe to slurp.
 sub psgi_qx {
-       my ($self, $env, $limiter, $qx_cb) = @_;
+       my ($self, $env, $limiter, $qx_cb, $cb_arg) = @_;
+       $self->{psgi_env} = $env;
        my $scalar = '';
        open(my $qx, '+>', \$scalar) or die; # PerlIO::scalar
        my $end = sub {
                my $err = $_[0]; # $!
                log_err($env, "psgi_qx: $err") if defined($err);
-               finish($self, $env);
-               eval { $qx_cb->(\$scalar) };
-               $qx = $scalar = undef;
+               finish($self, sub { $qx_cb->(\$scalar, $cb_arg) });
+               $qx = undef;
        };
        my $rpipe; # comes from popen_rd
        my $async = $env->{'pi-httpd.async'};
@@ -179,7 +185,7 @@ reread:
                ($rpipe) = @_; # popen_rd result
                if ($async) {
                # PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
-                       $async = $async->($rpipe, $cb, $end);
+                       $async = $async->($rpipe, $cb, undef, $end);
                        # $cb will call ->async_pass or ->close
                } else { # generic PSGI
                        $cb->() while $qx;
@@ -224,11 +230,12 @@ sub filter_fh ($$) {
 #              immediately (or streamed via ->getline (pull-based)).
 sub psgi_return {
        my ($self, $env, $limiter, $parse_hdr) = @_;
+       $self->{psgi_env} = $env;
        my ($fh, $rpipe);
        my $end = sub {
                my $err = $_[0]; # $!
                log_err($env, "psgi_return: $err") if defined($err);
-               finish($self, $env);
+               finish($self);
                $fh->close if $fh; # async-only
        };
 
@@ -291,7 +298,7 @@ sub psgi_return {
                ($rpipe) = @_;
                if ($async) {
                        # PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
-                       $async = $async->($rpipe, $cb, $end);
+                       $async = $async->($rpipe, $cb, undef, $end);
                        # $cb will call ->async_pass or ->close
                } else { # generic PSGI
                        $cb->() while $rd_hdr;