+sub psgi_qx_init_cb {
+ my ($self) = @_;
+ my $async = delete $self->{async}; # PublicInbox::HTTPD::Async
+ my ($r, $buf);
+ my $qx_fh = $self->{qx_fh};
+reread:
+ $r = sysread($self->{rpipe}, $buf, 65536);
+ if ($async) {
+ $async->async_pass($self->{psgi_env}->{'psgix.io'},
+ $qx_fh, \$buf);
+ } elsif (defined $r) {
+ $r ? (print $qx_fh $buf) : event_step($self, undef);
+ } else {
+ return if $! == EAGAIN; # try again when notified
+ goto reread if $! == EINTR;
+ event_step($self, $!);
+ }
+}
+
+sub psgi_qx_start {
+ my ($self) = @_;
+ if (my $async = $self->{psgi_env}->{'pi-httpd.async'}) {
+ # PublicInbox::HTTPD::Async->new(rpipe, $cb, cb_arg, $end_obj)
+ $self->{async} = $async->($self->{rpipe},
+ \&psgi_qx_init_cb, $self, $self);
+ # init_cb will call ->async_pass or ->close
+ } else { # generic PSGI
+ psgi_qx_init_cb($self) while $self->{qx_fh};
+ }
+}
+