+sub psgi_return_init_cb {
+ my ($self) = @_;
+ my $r = rd_hdr($self) or return;
+ my $env = $self->{psgi_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 (ref($r) ne 'ARRAY' || scalar(@$r) == 3) { # error
+ if ($async) { # calls rpipe->close && ->event_step
+ $async->close; # PublicInbox::HTTPD::Async::close
+ } else { # generic PSGI:
+ delete($self->{rpipe})->close;
+ event_step($self);
+ waitpid_err($self);
+ }
+ 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->{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, $$buf, $filter);
+ $wcb->($r);
+ }
+}
+
+sub psgi_return_start { # may run later, much later...
+ my ($self) = @_;
+ if (my $cb = $self->{psgi_env}->{'pi-httpd.async'}) {
+ # PublicInbox::HTTPD::Async->new(rpipe, $cb, $cb_arg, $end_obj)
+ $self->{async} = $cb->($self->{rpipe},
+ \&psgi_return_init_cb, $self, $self);
+ } else { # generic PSGI
+ psgi_return_init_cb($self) while $self->{parse_hdr};
+ }
+}
+