# pronounced "exit": x_it(1 << 8) => exit(1); x_it(13) => SIGPIPE
sub x_it ($$) {
my ($self, $code) = @_;
- local $current_lei = $self;
# make sure client sees stdout before exit
$self->{1}->autoflush(1) if $self->{1};
stop_pager($self);
sub fail_handler ($;$$) {
my ($lei, $code, $io) = @_;
- local $current_lei = $lei;
close($io) if $io; # needed to avoid warnings on SIGPIPE
_drop_wq($lei);
x_it($lei, $code // (1 << 8));
$pkg->can($pfx.$ucmd) : undef;
}
+sub do_env {
+ my $lei = shift;
+ fchdir($lei);
+ my $cb = shift // return ($lei, %{$lei->{env}}) ;
+ local ($current_lei, %ENV) = ($lei, %{$lei->{env}});
+ $cb = $lei->can($cb) if !ref($cb); # $cb may be a scalar sub name
+ eval { $cb->($lei, @_) };
+ $lei->fail($@) if $@;
+}
+
sub dispatch {
my ($self, $cmd, @argv) = @_;
- fchdir($self);
- local %ENV = %{$self->{env}};
- local $current_lei = $self; # for __WARN__
+ local ($current_lei, %ENV) = do_env($self);
$self->{2}->autoflush(1); # keep stdout buffered until x_it|DESTROY
return _help($self, 'no command given') unless defined($cmd);
# do not support Getopt bundling for this
# called by top-level lei-daemon when first worker is done with auth
# passes updated net auth info to current workers
sub net_merge_continue {
- my ($wq, $lei, $net_new) = @_;
+ my ($lei, $wq, $net_new) = @_;
$wq->{-net_new} = $net_new; # for "lei up"
$wq->wq_broadcast('PublicInbox::LeiAuth::net_merge_all', $net_new);
$wq->net_merge_all_done($lei); # defined per-WQ
sub op_merge { # prepares PktOp->pair ops
my ($self, $ops, $wq, $lei) = @_;
- $ops->{net_merge_continue} = [ \&net_merge_continue, $wq, $lei ];
+ $ops->{net_merge_continue} = [ \&net_merge_continue, $lei, $wq ];
}
sub new { bless \(my $x), __PACKAGE__ }
# run in script/lei foreground
my ($op_c, $op_p) = PublicInbox::PktOp->pair;
# $op_p will EOF when $EDITOR is done
- $op_c->{ops} = { '' => [\&cfg_edit_done, $self] };
+ $op_c->{ops} = { '' => [\&cfg_edit_done, $lei, $self] };
$lei->send_exec_cmd([ @$lei{qw(0 1 2)}, $op_p->{op_p} ], $cmd, $env);
}
-sub cfg_edit_done { # PktOp
- my ($self) = @_;
- eval {
- open my $fh, '+>', undef or die "open($!)";
- my $cfg = do {
- local $self->{lei}->{2} = $fh;
- $self->{lei}->cfg_dump($self->{-f});
- } or do {
- seek($fh, 0, SEEK_SET);
- return cfg_do_edit($self, do { local $/; <$fh> });
- };
- $self->cfg_verify($cfg) if $self->can('cfg_verify');
+sub cfg_edit_done { # PktOp lei->do_env cb
+ my ($lei, $self) = @_;
+ open my $fh, '+>', undef or die "open($!)";
+ my $cfg = do {
+ local $lei->{2} = $fh;
+ $lei->cfg_dump($self->{-f});
+ } or do {
+ seek($fh, 0, SEEK_SET);
+ return cfg_do_edit($self, do { local $/; <$fh> });
};
- $self->{lei}->fail($@) if $@;
+ $self->cfg_verify($cfg) if $self->can('cfg_verify');
}
sub lei_config {
$self->wq_close;
}
+sub do_inspect { # lei->do_env cb
+ my ($lei) = @_;
+ my $str = delete $lei->{istr};
+ $str =~ s/\A[\r\n]*From [^\r\n]*\r?\n//s;
+ my $eml = PublicInbox::Eml->new(\$str);
+ inspect_start($lei, [ 'blob:'.$lei->git_oid($eml)->hexdigest,
+ map { "mid:$_" } @{mids($eml)} ]);
+}
+
sub ins_add { # InputPipe->consume callback
my ($lei) = @_; # $_[1] = $rbuf
- if (defined $_[1]) {
- $_[1] eq '' and return eval {
- $lei->fchdir;
- local %ENV = %{$lei->{env}};
- local $PublicInbox::LEI::current_lei = $lei;
- my $str = delete $lei->{istr};
- $str =~ s/\A[\r\n]*From [^\r\n]*\r?\n//s;
- my $eml = PublicInbox::Eml->new(\$str);
- inspect_start($lei, [
- 'blob:'.$lei->git_oid($eml)->hexdigest,
- map { "mid:$_" } @{mids($eml)} ]);
- };
- $lei->{istr} .= $_[1];
- } else {
- $lei->fail("error reading stdin: $!");
- }
+ $_[1] // return $lei->fail("error reading stdin: $!");
+ return $lei->{istr} .= $_[1] if $_[1] ne '';
+ $lei->do_env(\&do_inspect);
}
sub lei_inspect {
@q ? join(' OR ', @q) : $lei->fail("no Message-ID in: @argv");
}
+sub do_lcat { # lei->do_env cb
+ my ($lei) = @_;
+ my @argv = split(/\s+/, $lei->{mset_opt}->{qstr});
+ $lei->{mset_opt}->{qstr} = extract_all($lei, @argv) or return;
+ $lei->_start_query;
+}
+
sub _stdin { # PublicInbox::InputPipe::consume callback for --stdin
my ($lei) = @_; # $_[1] = $rbuf
$_[1] // return $lei->fail("error reading stdin: $!");
return $lei->{mset_opt}->{qstr} .= $_[1] if $_[1] ne '';
- eval {
- $lei->fchdir;
- local %ENV = %{$lei->{env}};
- local $PublicInbox::LEI::current_lei = $lei;
- my @argv = split(/\s+/, $lei->{mset_opt}->{qstr});
- $lei->{mset_opt}->{qstr} = extract_all($lei, @argv) or return;
- $lei->_start_query;
- };
- $lei->fail($@) if $@;
+ $lei->do_env(\&do_lcat);
}
sub lei_lcat {
$lxs->do_query($self);
}
+sub do_qry { # do_env cb
+ my ($lei) = @_;
+ $lei->{mset_opt}->{q_raw} = $lei->{mset_opt}->{qstr};
+ $lei->{lse}->query_approxidate($lei->{lse}->git,
+ $lei->{mset_opt}->{qstr});
+ _start_query($lei);
+}
+
sub qstr_add { # PublicInbox::InputPipe::consume callback for --stdin
my ($lei) = @_; # $_[1] = $rbuf
$_[1] // $lei->fail("error reading stdin: $!");
return $lei->{mset_opt}->{qstr} .= $_[1] if $_[1] ne '';
- eval {
- $lei->fchdir;
- local %ENV = %{$lei->{env}};
- local $PublicInbox::LEI::current_lei = $lei;
- $lei->{mset_opt}->{q_raw} = $lei->{mset_opt}->{qstr};
- $lei->{lse}->query_approxidate($lei->{lse}->git,
- $lei->{mset_opt}->{qstr});
- _start_query($lei);
- };
- $lei->fail($@) if $@;
+ $lei->do_env(\&do_qry);
}
# make the URI||PublicInbox::{Inbox,ExtSearch} a config-file friendly string
sub query_done { # EOF callback for main daemon
my ($lei) = @_;
- local $PublicInbox::LEI::current_lei = $lei;
- eval {
- my $l2m = delete $lei->{l2m};
- delete $lei->{lxs};
- ($lei->{opt}->{'mail-sync'} && !$lei->{sto}) and
- warn "BUG: {sto} missing with --mail-sync";
- $lei->sto_done_request if $lei->{sto};
- if (my $v2w = delete $lei->{v2w}) {
- my $wait = $v2w->wq_do('done'); # may die
- $v2w->wq_close;
- }
- $lei->{ovv}->ovv_end($lei);
- if ($l2m) { # close() calls LeiToMail reap_compress
- if (my $out = delete $lei->{old_1}) {
- if (my $mbout = $lei->{1}) {
- close($mbout) or die <<"";
+ my $l2m = delete $lei->{l2m};
+ delete $lei->{lxs};
+ ($lei->{opt}->{'mail-sync'} && !$lei->{sto}) and
+ warn "BUG: {sto} missing with --mail-sync";
+ $lei->sto_done_request if $lei->{sto};
+ if (my $v2w = delete $lei->{v2w}) {
+ my $wait = $v2w->wq_do('done'); # may die
+ $v2w->wq_close;
+ }
+ $lei->{ovv}->ovv_end($lei);
+ if ($l2m) { # close() calls LeiToMail reap_compress
+ if (my $out = delete $lei->{old_1}) {
+ if (my $mbout = $lei->{1}) {
+ close($mbout) or die <<"";
Error closing $lei->{ovv}->{dst}: \$!=$! \$?=$?
- }
- $lei->{1} = $out;
- }
- if ($l2m->lock_free) {
- $l2m->poke_dst;
- $lei->poke_mua;
- } else { # mbox users
- delete $l2m->{mbl}; # drop dotlock
}
+ $lei->{1} = $out;
}
- if ($lei->{-progress}) {
- my $tot = $lei->{-mset_total} // 0;
- my $nr_w = $lei->{-nr_write} // 0;
- my $d = ($lei->{-nr_seen} // 0) - $nr_w;
- my $x = "$tot matches";
- $x .= ", $d duplicates" if $d;
- if ($l2m) {
- my $m = "# $nr_w written to " .
- "$lei->{ovv}->{dst} ($x)";
- $nr_w ? $lei->qfin($m) : $lei->qerr($m);
- } else {
- $lei->qerr("# $x");
- }
+ if ($l2m->lock_free) {
+ $l2m->poke_dst;
+ $lei->poke_mua;
+ } else { # mbox users
+ delete $l2m->{mbl}; # drop dotlock
}
- $lei->start_mua if $l2m && !$l2m->lock_free;
- $lei->dclose;
- };
- $lei->fail($@) if $@;
+ }
+ if ($lei->{-progress}) {
+ my $tot = $lei->{-mset_total} // 0;
+ my $nr_w = $lei->{-nr_write} // 0;
+ my $d = ($lei->{-nr_seen} // 0) - $nr_w;
+ my $x = "$tot matches";
+ $x .= ", $d duplicates" if $d;
+ if ($l2m) {
+ my $m = "# $nr_w written to " .
+ "$lei->{ovv}->{dst} ($x)";
+ $nr_w ? $lei->qfin($m) : $lei->qerr($m);
+ } else {
+ $lei->qerr("# $x");
+ }
+ }
+ $lei->start_mua if $l2m && !$l2m->lock_free;
+ $lei->dclose;
}
sub do_post_augment {
my ($lei) = @_;
- local $PublicInbox::LEI::current_lei = $lei;
my $l2m = $lei->{l2m} or return; # client disconnected
- eval {
- $lei->fchdir;
- $l2m->post_augment($lei);
- };
+ eval { $l2m->post_augment($lei) };
my $err = $@;
if ($err) {
if (my $lxs = delete $lei->{lxs}) {
}
sub incr_start_query { # called whenever an l2m shard starts do_post_auth
- my ($self, $lei) = @_;
+ my ($lei, $self) = @_;
my $l2m = $lei->{l2m};
return if ++$self->{nr_start_query} != $l2m->{-wq_nr_workers};
start_query($self, $lei);
my ($self, $lei) = @_;
my $l2m = $lei->{l2m};
my $ops = {
- 'sigpipe_handler' => [ $lei ],
- 'fail_handler' => [ $lei ],
- 'do_post_augment' => [ \&do_post_augment, $lei ],
- 'incr_post_augment' => [ \&incr_post_augment, $lei ],
+ sigpipe_handler => [ $lei ],
+ fail_handler => [ $lei ],
+ do_post_augment => [ \&do_post_augment, $lei ],
+ incr_post_augment => [ \&incr_post_augment, $lei ],
'' => [ \&query_done, $lei ],
- 'mset_progress' => [ \&mset_progress, $lei ],
- 'l2m_progress' => [ \&l2m_progress, $lei ],
- 'x_it' => [ $lei ],
- 'child_error' => [ $lei ],
- 'incr_start_query' => [ $self, $lei ],
+ mset_progress => [ \&mset_progress, $lei ],
+ l2m_progress => [ \&l2m_progress, $lei ],
+ x_it => [ $lei ],
+ child_error => [ $lei ],
+ incr_start_query => [ \&incr_start_query, $lei, $self ],
};
$lei->{auth}->op_merge($ops, $l2m, $lei) if $l2m && $lei->{auth};
my $end = $lei->pkt_op_pair;
-# 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
# 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 ECONNRESET);
use PublicInbox::Syscall qw(EPOLLIN);
my $op = $self->{ops}->{$cmd //= $msg};
if ($op) {
my ($obj, @args) = (@$op, @pargs);
- blessed($obj) ? $obj->$cmd(@args) : $obj->(@args);
+ 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 {
+ $obj->(@args);
+ }
} elsif ($msg ne '') {
die "BUG: unknown message: `$cmd'";
}
projects / thirdparty / public-inbox.git / commitdiff
