qspawn: drop "qspawn.filter" support, for now

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

diff --git a/lib/PublicInbox/Qspawn.pm b/lib/PublicInbox/Qspawn.pm
index 33e20147bbb9bad6acc18c621bc42a0e91c69d54..d1a34beac89ffd89c38083a0526be6e534395b8d 100644 (file)
--- a/lib/PublicInbox/Qspawn.pm
+++ b/lib/PublicInbox/Qspawn.pm
@@ -27,7 +27,6 @@ package PublicInbox::Qspawn;
 use strict;
 use warnings;
 use PublicInbox::Spawn qw(popen_rd);
-require Plack::Util;
 
 # n.b.: we get EAGAIN with public-inbox-httpd, and EINTR on other PSGI servers
 use Errno qw(EAGAIN EINTR);
@@ -149,6 +148,37 @@ sub start ($$$) {
        }
 }
 
+sub psgi_qx_init_cb {
+       my ($self) = @_;
+       my $async = delete $self->{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 ? $qx_fh->write($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};
+       }
+}
+
 # Similar to `backtick` or "qx" ("perldoc -f qx"), it calls $qx_cb with
 # the stdout of the given command when done; but respects the given limiter
 # $env is the PSGI env.  As with ``/qx; only use this when output is small
@@ -162,44 +192,8 @@ sub psgi_qx {
        $self->{qx_arg} = $qx_arg;
        $self->{qx_fh} = $qx_fh;
        $self->{qx_buf} = \$qx_buf;
-       my $async = $env->{'pi-httpd.async'};
-       my $cb = sub {
-               my ($r, $buf);
-reread:
-               $r = sysread($self->{rpipe}, $buf, 65536);
-               if ($async) {
-                       $async->async_pass($env->{'psgix.io'}, $qx_fh, \$buf);
-               } elsif (defined $r) {
-                       $r ? $qx_fh->write($buf) : event_step($self, undef);
-               } else {
-                       return if $! == EAGAIN; # try again when notified
-                       goto reread if $! == EINTR;
-                       event_step($self, $!);
-               }
-       };
        $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
-       $self->start($limiter, sub { # start_cb, may run later, much later...
-               if ($async) {
-               # PublicInbox::HTTPD::Async->new(rpipe, $cb, $end_obj)
-                       $async = $async->($self->{rpipe}, $cb, undef, $self);
-                       # $cb will call ->async_pass or ->close
-               } else { # generic PSGI
-                       $cb->() while $self->{qx_fh};
-               }
-       });
-}
-
-# create a filter for "push"-based streaming PSGI writes used by HTTPD::Async
-sub filter_fh ($$) {
-       my ($fh, $filter) = @_;
-       Plack::Util::inline_object(
-               close => sub {
-                       $fh->write($filter->(undef));
-                       $fh->close;
-               },
-               write => sub {
-                       $fh->write($filter->($_[0]));
-               });
+       start($self, $limiter, \&psgi_qx_start);
 }
 
 # this is called on pipe EOF to reap the process, may be called
@@ -243,7 +237,6 @@ 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'};
        my $wcb = delete $env->{'qspawn.wcb'};
        my $async = delete $self->{async};
        if (scalar(@$r) == 3) { # error
@@ -258,7 +251,6 @@ sub psgi_return_init_cb {
        } elsif ($async) {
                # done reading headers, handoff to read body
                my $fh = $wcb->($r); # scalar @$r == 2
-               $fh = filter_fh($fh, $filter) if $filter;
                $self->{fh} = $fh;
                $async->async_pass($env->{'psgix.io'}, $fh,
                                        delete($self->{hdr_buf}));
@@ -266,7 +258,7 @@ sub psgi_return_init_cb {
                require PublicInbox::GetlineBody;
                $r->[2] = PublicInbox::GetlineBody->new($self->{rpipe},
                                        \&event_step, $self,
-                                       ${$self->{hdr_buf}}, $filter);
+                                       ${$self->{hdr_buf}});
                $wcb->($r);
        }
 
@@ -296,9 +288,6 @@ sub psgi_return_start { # may run later, much later...
 #                          psgi_return will return an anonymous
 #                          sub for the PSGI server to call
 #
-#   $env->{'qspawn.filter'} - filter callback, receives a string as input,
-#                             undef on EOF
-#
 # $limiter - the Limiter object to use (uses the def_limiter if not given)
 #
 # $parse_hdr - Initial read function; often for parsing CGI header output.