'add-external' => [ 'LOCATION',
'add/set priority of a publicinbox|extindex for extra matches',
- qw(boost=i mirror=s inbox-version=i verbose|v+),
+ qw(boost=i mirror=s inbox-version=i epoch=s verbose|v+),
@c_opt, index_opt(), @net_opt ],
'ls-external' => [ '[FILTER]', 'list publicinbox|extindex locations',
qw(format|f=s z|0 globoff|g invert-match|v local remote), @c_opt ],
'remote' => 'limit operations to those requiring network access',
'remote!' => 'prevent operations requiring network access',
-'all:s up' => ['local|remote', 'update all remote or local saved searches' ],
+# up, refresh-mail-sync, export-kw
+'all:s' => ['TYPE|local|remote', 'all remote or local folders' ],
+
'remote-fudge-time=s' => [ 'INTERVAL',
'look for mail INTERVAL older than the last successful query' ],
my ($self) = @_;
delete $self->{-progress};
_drop_wq($self) if $self->{failed};
- close(delete $self->{1}) if $self->{1}; # may reap_compress
$self->close if $self->{-event_init_done}; # PublicInbox::DS::close
}
sub event_step {
my ($self) = @_;
local %ENV = %{$self->{env}};
- my $sock = $self->{sock};
local $current_lei = $self;
eval {
- while (my @fds = $recv_cmd->($sock, my $buf, 4096)) {
+ my $buf;
+ while (my @fds = $recv_cmd->($self->{sock}, $buf, 4096)) {
if (scalar(@fds) == 1 && !defined($fds[0])) {
return if $! == EAGAIN;
next if $! == EINTR;
last if $! == ECONNRESET;
die "recvmsg: $!";
}
- for my $fd (@fds) {
- open my $rfh, '+<&=', $fd;
+ for (@fds) { open my $rfh, '+<&=', $_ }
+ }
+ if ($buf eq '') {
+ _drop_wq($self); # EOF, client disconnected
+ dclose($self);
+ } elsif ($buf =~ /\A(STOP|CONT)\z/) {
+ for my $wq (grep(defined, @$self{@WQ_KEYS})) {
+ $wq->wq_kill($buf) or $wq->wq_kill_old($buf);
}
+ } else {
die "unrecognized client signal: $buf";
}
- _drop_wq($self); # EOF, client disconnected
- dclose($self);
};
if (my $err = $@) {
eval { $self->fail($err) };
my ($self) = @_;
my $sock = $self->{sock} or return;
$self->{-event_init_done} //= do { # persist til $ops done
+ $sock->blocking(0);
$self->SUPER::new($sock, EPOLLIN|EPOLLET);
$sock;
};
$errors_log = "$sock_dir/errors.log";
my $addr = pack_sockaddr_un($path);
my $lk = bless { lock_path => $errors_log }, 'PublicInbox::Lock';
+ umask(077) // die("umask(077): $!");
$lk->lock_acquire;
socket($listener, AF_UNIX, SOCK_SEQPACKET, 0) or die "socket: $!";
if ($errno == ECONNREFUSED || $errno == ENOENT) {
$! = $errno; # allow interpolation to stringify in die
die "connect($path): $!";
}
- umask(077) // die("umask(077): $!");
bind($listener, $addr) or die "bind($path): $!";
$lk->lock_release;
undef $lk;
sub wq_eof { # EOF callback for main daemon
my ($lei) = @_;
my $wq1 = delete $lei->{wq1} // return $lei->fail; # already failed
- $wq1->wq_wait_old(\&wq_done_wait, $lei);
+ $wq1->wq_wait_old($wq1->can('_wq_done_wait') // \&wq_done_wait, $lei);
}
sub watch_state_ok ($) {