lib/PublicInbox/OpPipe.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 # bytecode dispatch pipe, reads a byte, runs a sub
   5 # byte => [ sub, @operands ]
   6 package PublicInbox::OpPipe;
   7 use strict;
   8 use v5.10.1;
   9 use parent qw(PublicInbox::DS);
  10 use PublicInbox::Syscall qw(EPOLLIN);
  11
  12 sub new {
  13         my ($cls, $rd, $op_map, $in_loop) = @_;
  14         my $self = bless { sock => $rd, op_map => $op_map }, $cls;
  15         # 1031: F_SETPIPE_SZ, 4096: page size
  16         fcntl($rd, 1031, 4096) if $^O eq 'linux';
  17         if ($in_loop) { # iff using DS->EventLoop
  18                 $rd->blocking(0);
  19                 $self->SUPER::new($rd, EPOLLIN);
  20         }
  21         $self;
  22 }
  23
  24 sub event_step {
  25         my ($self) = @_;
  26         my $rd = $self->{sock};
  27         my $byte;
  28         until (defined(sysread($rd, $byte, 1))) {
  29                 return if $!{EAGAIN};
  30                 next if $!{EINTR};
  31                 die "read \$rd: $!";
  32         }
  33         my $op = $self->{op_map}->{$byte} or die "BUG: unknown byte `$byte'";
  34         if ($byte eq '') { # close on EOF
  35                 $rd->blocking ? delete($self->{sock}) : $self->close;
  36         }
  37         my ($sub, @args) = @$op;
  38         $sub->(@args);
  39 }
  40
  41 1;