lib/PublicInbox/InputPipe.pm

   1 # Copyright (C) 2021 all contributors <meta@public-inbox.org>
   2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
   3
   4 # for reading pipes and sockets off the DS event loop
   5 package PublicInbox::InputPipe;
   6 use strict;
   7 use v5.10.1;
   8 use parent qw(PublicInbox::DS);
   9 use PublicInbox::Syscall qw(EPOLLIN EPOLLET);
  10
  11 sub consume {
  12         my ($in, $cb, @args) = @_;
  13         my $self = bless { cb => $cb, sock => $in, args => \@args },__PACKAGE__;
  14         if ($PublicInbox::DS::in_loop) {
  15                 eval { $self->SUPER::new($in, EPOLLIN|EPOLLET) };
  16                 return $in->blocking(0) unless $@; # regular file sets $@
  17         }
  18         event_step($self) while $self->{sock};
  19 }
  20
  21 sub event_step {
  22         my ($self) = @_;
  23         my ($r, $rbuf);
  24         while (($r = sysread($self->{sock}, $rbuf, 65536))) {
  25                 $self->{cb}->(@{$self->{args} // []}, $rbuf);
  26         }
  27         if (defined($r)) { # EOF
  28                 $self->{cb}->(@{$self->{args} // []}, '');
  29         } elsif ($!{EAGAIN}) {
  30                 return;
  31         } else {
  32                 $self->{cb}->(@{$self->{args} // []}, undef)
  33         }
  34         $self->{sock}->blocking ? delete($self->{sock}) : $self->close
  35 }
  36
  37 1;