+# middlewares such as Deflater may write empty strings
+sub chunked_wcb ($) {
+ my ($self) = @_;
+ sub {
+ return if $_[0] eq '';
+ more($self, sprintf("%x\r\n", bytes::length($_[0])));
+ more($self, $_[0]);
+
+ # use $self->write("\n\n") if you care about real-time
+ # streaming responses, public-inbox WWW does not.
+ more($self, "\r\n");
+ }
+}
+
+sub identity_wcb ($) {
+ my ($self) = @_;
+ sub { $self->write(\($_[0])) if $_[0] ne '' }
+}
+
+sub next_request ($) {
+ my ($self) = @_;
+ $self->watch_write(0);
+ if ($self->{rbuf} eq '') { # wait for next request
+ $self->watch_read(1);
+ } else { # avoid recursion for pipelined requests
+ push @$pipelineq, $self;
+ $pipet ||= PublicInbox::EvCleanup::asap(*process_pipelineq);
+ }
+}
+
+sub response_done_cb ($$) {
+ my ($self, $alive) = @_;
+ sub {
+ my $env = $self->{env};
+ $self->{env} = undef;
+ $self->write("0\r\n\r\n") if $alive == 2;
+ $self->write(sub{$alive ? next_request($self) : $self->close});
+ }
+}
+
+sub getline_cb ($$$) {
+ my ($self, $write, $close) = @_;
+ local $/ = \8192;
+ my $forward = $self->{forward};
+ # limit our own running time for fairness with other
+ # clients and to avoid buffering too much:
+ if ($forward) {
+ my $buf = eval { $forward->getline };
+ if (defined $buf) {
+ $write->($buf); # may close in PublicInbox::DS::write
+ unless ($self->{closed}) {
+ my $next = $self->{pull};
+ if ($self->{write_buf_size}) {
+ $self->write($next);
+ } else {
+ PublicInbox::EvCleanup::asap($next);
+ }
+ return;
+ }
+ } elsif ($@) {
+ err($self, "response ->getline error: $@");
+ $forward = undef;
+ $self->close;
+ }
+ }
+
+ $self->{forward} = $self->{pull} = undef;
+ # avoid recursion
+ if ($forward) {
+ eval { $forward->close };
+ if ($@) {
+ err($self, "response ->close error: $@");
+ $self->close; # idempotent
+ }
+ }
+ $close->();
+}
+
+sub getline_response ($$$) {
+ my ($self, $write, $close) = @_;
+ my $pull = $self->{pull} = sub { getline_cb($self, $write, $close) };
+ $pull->();
+}
+