www_coderepo: implement /$CODE_REPO/atom/ endpoint

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

diff --git a/lib/PublicInbox/Qspawn.pm b/lib/PublicInbox/Qspawn.pm
index 79db9e7618fc183ca2d9a35ddfbcdda7ebc2a9d9..ab0ef25c5611f3e645a4f4e593ae143ee94291a6 100644 (file)
--- a/lib/PublicInbox/Qspawn.pm
+++ b/lib/PublicInbox/Qspawn.pm
@@ -1,4 +1,4 @@
-# Copyright (C) 2016-2021 all contributors <meta@public-inbox.org>
+# Copyright (C) all contributors <meta@public-inbox.org>
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 
 # Like most Perl modules in public-inbox, this is internal and
@@ -26,6 +26,7 @@
 
 package PublicInbox::Qspawn;
 use strict;
+use v5.10.1;
 use PublicInbox::Spawn qw(popen_rd);
 use PublicInbox::GzipFilter;
 
@@ -38,7 +39,8 @@ my $def_limiter;
 # $cmd is the command to spawn
 # $cmd_env is the environ for the child process (not PSGI env)
 # $opt can include redirects and perhaps other process spawning options
-sub new ($$$;) {
+# {qsp_err} is an optional error buffer callers may access themselves
+sub new {
        my ($class, $cmd, $cmd_env, $opt) = @_;
        bless { args => [ $cmd, $cmd_env, $opt ] }, $class;
 }
@@ -93,18 +95,19 @@ sub finalize ($$) {
        }
 
        if ($err) {
-               if (defined $self->{err}) {
-                       $self->{err} .= "; $err";
-               } else {
-                       $self->{err} = $err;
-               }
-               if ($env && $self->{cmd}) {
-                       warn join(' ', @{$self->{cmd}}) . ": $err";
+               utf8::decode($err);
+               if (my $dst = $self->{qsp_err}) {
+                       $$dst .= $$dst ? " $err" : "; $err";
                }
+               warn "@{$self->{cmd}}: $err" if $self->{cmd};
        }
        if ($qx_cb) {
                eval { $qx_cb->($qx_buf, $qx_arg) };
-       } elsif (my $wcb = delete $env->{'qspawn.wcb'}) {
+               return unless $@;
+               warn "E: $@"; # hope qspawn.wcb can handle it
+       }
+       return if $self->{passed}; # another command chained it
+       if (my $wcb = delete $env->{'qspawn.wcb'}) {
                # have we started writing, yet?
                require PublicInbox::WwwStatic;
                $wcb->(PublicInbox::WwwStatic::r(500));
@@ -185,7 +188,7 @@ sub event_step {
        my ($self, $err) = @_; # $err: $!
        warn "psgi_{return,qx} $err" if defined($err);
        finish($self);
-       my ($fh, $qx_fh) = delete(@$self{qw(fh qx_fh)});
+       my ($fh, $qx_fh) = delete(@$self{qw(qfh qx_fh)});
        $fh->close if $fh; # async-only (psgi_return)
 }
 
@@ -223,31 +226,43 @@ sub psgi_return_init_cb {
        my ($self) = @_;
        my $r = rd_hdr($self) or return;
        my $env = $self->{psgi_env};
-       my $filter = delete $env->{'qspawn.filter'} //
-               PublicInbox::GzipFilter::qsp_maybe($r->[1], $env);
+       my $filter;
+       if (ref($r) eq 'ARRAY' && Scalar::Util::blessed($r->[2]) &&
+                       $r->[2]->can('attach')) {
+               $filter = pop @$r;
+       }
+       $filter //= delete($env->{'qspawn.filter'}) // (ref($r) eq 'ARRAY' ?
+               PublicInbox::GzipFilter::qsp_maybe($r->[1], $env) : undef);
 
        my $wcb = delete $env->{'qspawn.wcb'};
        my $async = delete $self->{async}; # PublicInbox::HTTPD::Async
-       if (scalar(@$r) == 3) { # error
+       if (ref($r) ne 'ARRAY' || scalar(@$r) == 3) { # error
                if ($async) { # calls rpipe->close && ->event_step
                        $async->close; # PublicInbox::HTTPD::Async::close
-               } else {
-                       $self->{rpipe}->close;
+               } else { # generic PSGI:
+                       delete($self->{rpipe})->close;
                        event_step($self);
+                       waitpid_err($self);
                }
-               $wcb->($r);
+               if (ref($r) eq 'ARRAY') { # error
+                       $wcb->($r)
+               } elsif (ref($r) eq 'CODE') { # chain another command
+                       $r->($wcb);
+                       $self->{passed} = 1;
+               }
+               # else do nothing
        } elsif ($async) {
                # done reading headers, handoff to read body
                my $fh = $wcb->($r); # scalar @$r == 2
                $fh = $filter->attach($fh) if $filter;
-               $self->{fh} = $fh;
+               $self->{qfh} = $fh;
                $async->async_pass($env->{'psgix.io'}, $fh,
                                        delete($self->{hdr_buf}));
        } else { # for synchronous PSGI servers
                require PublicInbox::GetlineBody;
+               my $buf = delete $self->{hdr_buf};
                $r->[2] = PublicInbox::GetlineBody->new($self->{rpipe},
-                                       \&event_step, $self,
-                                       ${$self->{hdr_buf}}, $filter);
+                                       \&event_step, $self, $$buf, $filter);
                $wcb->($r);
        }
 }