@c_opt ],
'import' => [ 'LOCATION...|--stdin',
'one-time import/update from URL or filesystem',
- qw(stdin| offset=i recursive|r exclude=s include|I=s
+ qw(stdin| offset=i recursive|r exclude=s include|I=s jobs=s
lock=s@ in-format|F=s kw! verbose|v+ incremental! mail-sync!),
qw(no-torsocks torsocks=s), PublicInbox::LeiQuery::curl_opt(), @c_opt ],
'forget-mail-sync' => [ 'LOCATION...',
'leistore.dir' => 'top-level storage location',
);
-my @WQ_KEYS = qw(lxs l2m wq1 ikw); # internal workers
+my @WQ_KEYS = qw(lxs l2m ikw pmd wq1); # internal workers
sub _drop_wq {
my ($self) = @_;
sub fail ($$;$) {
my ($self, $buf, $exit_code) = @_;
+ $self->{failed}++;
err($self, $buf) if defined $buf;
# calls fail_handler:
$self->{pkt_op_p}->pkt_do('!') if $self->{pkt_op_p};
}
sub workers_start {
- my ($lei, $wq, $jobs, $ops) = @_;
+ my ($lei, $wq, $jobs, $ops, $flds) = @_;
$ops = {
'!' => [ \&fail_handler, $lei ],
'|' => [ \&sigpipe_handler, $lei ],
$ops->{''} //= [ $wq->can('_lei_wq_eof') || \&wq_eof, $lei ];
my $end = $lei->pkt_op_pair;
my $ident = $wq->{-wq_ident} // "lei-$lei->{cmd} worker";
- $wq->wq_workers_start($ident, $jobs, $lei->oldset, { lei => $lei });
+ $flds->{lei} = $lei;
+ $wq->wq_workers_start($ident, $jobs, $lei->oldset, $flds);
delete $lei->{pkt_op_p};
my $op_c = delete $lei->{pkt_op_c};
+ # {-lei_sock} persists script/lei process until ops->{''} EOF callback
+ $op_c->{-lei_sock} = $lei->{sock};
@$end = ();
$lei->event_step_init;
($op_c, $ops);
}
+# call this when we're ready to wait on events and yield to other clients
+sub wait_wq_events {
+ my ($lei, $op_c, $ops) = @_;
+ for my $wq (grep(defined, @$lei{qw(ikw pmd)})) { # auxiliary WQs
+ $wq->wq_close(1);
+ }
+ $op_c->{ops} = $ops;
+}
+
sub _help {
require PublicInbox::LeiHelp;
PublicInbox::LeiHelp::call($_[0], $_[1], \%CMD, \%OPTDESC);
sub dclose {
my ($self) = @_;
delete $self->{-progress};
- _drop_wq($self);
+ _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
}