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, args => \@args }, __PACKAGE__;
  14         eval { $self->SUPER::new($in, EPOLLIN|EPOLLET) };
  15         return $self->requeue if $@; # regular file
  16         $in->blocking(0); # pipe or socket
  17 }
  18
  19 sub event_step {
  20         my ($self) = @_;
  21         my $r = sysread($self->{sock} // return, my $rbuf, 65536);
  22         if ($r) {
  23                 $self->{cb}->(@{$self->{args} // []}, $rbuf);
  24                 return $self->requeue; # may be regular file or pipe
  25         }
  26         if (defined($r)) { # EOF
  27                 $self->{cb}->(@{$self->{args} // []}, '');
  28         } elsif ($!{EAGAIN}) {
  29                 return;
  30         } else { # another error
  31                 $self->{cb}->(@{$self->{args} // []}, undef)
  32         }
  33         $self->{sock}->blocking ? delete($self->{sock}) : $self->close
  34 }
  35
  36 1;