1 files changed, 35 insertions, 32 deletions

diff --git a/lib/PublicInbox/PktOp.pm b/lib/PublicInbox/PktOp.pm
index 10942dd1..1bcdd799 100644
--- a/lib/PublicInbox/PktOp.pm
+++ b/lib/PublicInbox/PktOp.pm
@@ -1,4 +1,4 @@
-# Copyright (C) 2021 all contributors <meta@public-inbox.org>
+# Copyright (C) all contributors <meta@public-inbox.org>
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 
 # op dispatch socket, reads a message, runs a sub
@@ -6,19 +6,19 @@
 # Used for lei_xsearch and maybe other things
 # "command" => [ $sub, @fixed_operands ]
 package PublicInbox::PktOp;
-use strict;
-use v5.10.1;
+use v5.12;
 use parent qw(PublicInbox::DS);
-use Errno qw(EAGAIN EINTR);
-use PublicInbox::Syscall qw(EPOLLIN EPOLLET);
-use Socket qw(AF_UNIX MSG_EOR SOCK_SEQPACKET);
+use Errno qw(EAGAIN ECONNRESET);
+use PublicInbox::Syscall qw(EPOLLIN);
+use Socket qw(AF_UNIX SOCK_SEQPACKET);
 use PublicInbox::IPC qw(ipc_freeze ipc_thaw);
+use Scalar::Util qw(blessed);
 
 sub new {
         my ($cls, $r) = @_;
         my $self = bless { sock => $r }, $cls;
         $r->blocking(0);
-        $self->SUPER::new($r, EPOLLIN|EPOLLET);
+        $self->SUPER::new($r, EPOLLIN);
 }
 
 # returns a blessed objects as the consumer and producer
@@ -31,39 +31,42 @@ sub pair {
 
 sub pkt_do { # for the producer to trigger event_step in consumer
         my ($self, $cmd, @args) = @_;
-        send($self->{op_p}, @args ? "$cmd\0".ipc_freeze(\@args) : $cmd, MSG_EOR)
+        send($self->{op_p}, @args ? "$cmd\0".ipc_freeze(\@args) : $cmd, 0)
 }
 
 sub event_step {
         my ($self) = @_;
         my $c = $self->{sock};
-        my $msg;
-        while (1) {
-                my $n = recv($c, $msg, 4096, 0);
-                unless (defined $n) {
-                        return if $! == EAGAIN;
-                        next if $! == EINTR;
-                        $self->close;
-                        die "recv: $!";
-                }
-                my ($cmd, @pargs);
-                if (index($msg, "\0") > 0) {
-                        ($cmd, my $pargs) = split(/\0/, $msg, 2);
-                        @pargs = @{ipc_thaw($pargs)};
+        my $n = recv($c, my $msg, 4096, 0);
+        unless (defined $n) {
+                return if $! == EAGAIN;
+                die "recv: $!" if $! != ECONNRESET; # we may be bidirectional
+        }
+        my ($cmd, @pargs);
+        if (index($msg, "\0") > 0) {
+                ($cmd, my $pargs) = split(/\0/, $msg, 2);
+                @pargs = @{ipc_thaw($pargs)};
+        } else {
+                # for compatibility with the script/lei in client mode,
+                # it doesn't load Sereal||Storable for startup speed
+                ($cmd, @pargs) = split(/ /, $msg);
+        }
+        my $op = $self->{ops}->{$cmd //= $msg};
+        if ($op) {
+                my ($obj, @args) = (@$op, @pargs);
+                if (blessed($args[0]) && $args[0]->can('do_env')) {
+                        my $lei = shift @args;
+                        $lei->do_env($obj, @args);
+                } elsif (blessed($obj)) {
+                        $obj->can('do_env') ? $obj->do_env($cmd, @args)
+                                                : $obj->$cmd(@args);
                 } else {
-                        # for compatibility with the script/lei in client mode,
-                        # it doesn't load Sereal||Storable for startup speed
-                        ($cmd, @pargs) = split(/ /, $msg);
-                }
-                my $op = $self->{ops}->{$cmd //= $msg};
-                if ($op) {
-                        my ($sub, @args) = @$op;
-                        $sub->(@args, @pargs);
-                } elsif ($msg ne '') {
-                        die "BUG: unknown message: `$cmd'";
+                        $obj->(@args);
                 }
-                return $self->close if $msg eq ''; # close on EOF
+        } elsif ($msg ne '') {
+                die "BUG: unknown message: `$cmd'";
         }
+        $self->close if $msg eq ''; # close on EOF
 }
 
 1;