lei: simplify common LeiInput users with ->wq1_start

This method replaces a common pattern of starting workers,
preparing internal auth ops, and asynchronous waiting of
command completion.

It also adds missing LeiAuth support to rediff and rm
which rarely need auth.

12 files changed, 25 insertions, 61 deletions

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index 3e1706a0..887025de 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -631,7 +631,10 @@ sub pkt_ops {
 
 sub workers_start {
         my ($lei, $wq, $jobs, $ops, $flds) = @_;
-        $ops = pkt_ops($lei, { ($ops ? %$ops : ()) });
+        $ops //= {};
+        ($wq->can('net_merge_all_done') && $lei->{auth}) and
+                $lei->{auth}->op_merge($ops, $wq, $lei);
+        pkt_ops($lei, $ops);
         $ops->{''} //= [ $wq->can('_lei_wq_eof') || \&wq_eof, $lei ];
         my $end = $lei->pkt_op_pair;
         my $ident = $wq->{-wq_ident} // "lei-$lei->{cmd} worker";
@@ -648,12 +651,22 @@ sub workers_start {
 # call this when we're ready to wait on events and yield to other clients
 sub wait_wq_events {
         my ($lei, $op_c, $ops) = @_;
+        my $wq1 = $lei->{wq1};
+        ($wq1 && $wq1->can('net_merge_all_done') && !$lei->{auth}) and
+                $wq1->net_merge_all_done;
         for my $wq (grep(defined, @$lei{qw(ikw pmd)})) { # auxiliary WQs
                 $wq->wq_close;
         }
         $op_c->{ops} = $ops;
 }
 
+sub wq1_start {
+        my ($lei, $wq, $jobs) = @_;
+        my ($op_c, $ops) = workers_start($lei, $wq, $jobs // 1);
+        $lei->{wq1} = $wq;
+        wait_wq_events($lei, $op_c, $ops); # net_merge_all_done if !{auth}
+}
+
 sub _help {
         require PublicInbox::LeiHelp;
         PublicInbox::LeiHelp::call($_[0], $_[1], \%CMD, \%OPTDESC);
diff --git a/lib/PublicInbox/LeiExportKw.pm b/lib/PublicInbox/LeiExportKw.pm
index 0ecfb782..d2396fa7 100644
--- a/lib/PublicInbox/LeiExportKw.pm
+++ b/lib/PublicInbox/LeiExportKw.pm
@@ -124,13 +124,8 @@ EOM
                                         'imap_add_kw' : 'imap_set_kw');
                 $self->{nwr}->{-skip_creat} = 1;
         }
-        my $ops = {};
-        $lei->{auth}->op_merge($ops, $self, $lei) if $lei->{auth};
-        (my $op_c, $ops) = $lei->workers_start($self, 1, $ops);
-        $lei->{wq1} = $self;
         $lei->{-err_type} = 'non-fatal';
-        net_merge_all_done($self) unless $lei->{auth};
-        $lei->wait_wq_events($op_c, $ops); # net_merge_all_done if !{auth}
+        $lei->wq1_start($self);
 }
 
 sub _complete_export_kw {
diff --git a/lib/PublicInbox/LeiForgetSearch.pm b/lib/PublicInbox/LeiForgetSearch.pm
index dfeb0293..dd358ae1 100644
--- a/lib/PublicInbox/LeiForgetSearch.pm
+++ b/lib/PublicInbox/LeiForgetSearch.pm
@@ -45,12 +45,7 @@ sub lei_forget_search {
         if ($self->{o_remote}) { # setup lei->{auth}
                 $self->prepare_inputs($lei, $self->{o_remote}) or return;
         }
-        my $ops = {};
-        $lei->{auth}->op_merge($ops, $self, $lei) if $lei->{auth};
-        (my $op_c, $ops) = $lei->workers_start($self, 1, $ops);
-        $lei->{wq1} = $self;
-        net_merge_all_done($self) unless $lei->{auth};
-        $lei->wait_wq_events($op_c, $ops);
+        $lei->wq1_start($self);
 }
 
 sub do_prune {
diff --git a/lib/PublicInbox/LeiImport.pm b/lib/PublicInbox/LeiImport.pm
index d8f39fdf..bbc0634e 100644
--- a/lib/PublicInbox/LeiImport.pm
+++ b/lib/PublicInbox/LeiImport.pm
@@ -102,14 +102,9 @@ sub do_import_index ($$@) {
         }
         ($lei->{opt}->{'new-only'} && (!$net || !$net->{imap_order})) and
                 warn "# --new-only is only for IMAP\n";
-        my $ops = {};
-        $lei->{auth}->op_merge($ops, $self, $lei) if $lei->{auth};
         $lei->{-eml_noisy} = 1;
-        (my $op_c, $ops) = $lei->workers_start($self, $j, $ops);
-        $lei->{wq1} = $self;
         $lei->{-err_type} = 'non-fatal';
-        net_merge_all_done($self) unless $lei->{auth};
-        $lei->wait_wq_events($op_c, $ops);
+        $lei->wq1_start($self, $j);
 }
 
 sub lei_import { # the main "lei import" method
diff --git a/lib/PublicInbox/LeiLsMailSource.pm b/lib/PublicInbox/LeiLsMailSource.pm
index 5eb7032d..50799270 100644
--- a/lib/PublicInbox/LeiLsMailSource.pm
+++ b/lib/PublicInbox/LeiLsMailSource.pm
@@ -95,13 +95,8 @@ sub lei_ls_mail_source {
                 $json->pretty(1)->indent(2) if $isatty || $lei->{opt}->{pretty};
         }
         $lei->start_pager if $isatty;
-        my $ops = {};
-        $lei->{auth}->op_merge($ops, $self, $lei);
-        (my $op_c, $ops) = $lei->workers_start($self, 1, $ops);
-        $lei->{wq1} = $self;
         $lei->{-err_type} = 'non-fatal';
-        net_merge_all_done($self) unless $lei->{auth};
-        $lei->wait_wq_events($op_c, $ops); # net_merge_all_done if !{auth}
+        $lei->wq1_start($self);
 }
 
 sub _complete_ls_mail_source {
diff --git a/lib/PublicInbox/LeiMailDiff.pm b/lib/PublicInbox/LeiMailDiff.pm
index 48ba74cf..2b4cfd9e 100644
--- a/lib/PublicInbox/LeiMailDiff.pm
+++ b/lib/PublicInbox/LeiMailDiff.pm
@@ -81,13 +81,8 @@ sub lei_mail_diff {
         my $isatty = -t $lei->{1};
         $lei->{opt}->{color} //= $isatty;
         $lei->start_pager if $isatty;
-        my $ops = {};
-        $lei->{auth}->op_merge($ops, $self, $lei) if $lei->{auth};
-        (my $op_c, $ops) = $lei->workers_start($self, 1, $ops);
-        $lei->{wq1} = $self;
         $lei->{-err_type} = 'non-fatal';
-        net_merge_all_done($self) unless $lei->{auth};
-        $lei->wait_wq_events($op_c, $ops);
+        $lei->wq1_start($self);
 }
 
 no warnings 'once';
diff --git a/lib/PublicInbox/LeiP2q.pm b/lib/PublicInbox/LeiP2q.pm
index 09ec0a07..610adb78 100644
--- a/lib/PublicInbox/LeiP2q.pm
+++ b/lib/PublicInbox/LeiP2q.pm
@@ -183,12 +183,7 @@ sub lei_p2q { # the "lei patch-to-query" entry point
         $lei->{opt}->{'in-format'} //= 'eml' if $lei->{opt}->{stdin};
         my $self = bless { missing_ok => 1 }, __PACKAGE__;
         $self->prepare_inputs($lei, \@inputs) or return;
-        my $ops = {};
-        $lei->{auth}->op_merge($ops, $self, $lei) if $lei->{auth};
-        (my $op_c, $ops) = $lei->workers_start($self, 1, $ops);
-        $lei->{wq1} = $self;
-        net_merge_all_done($self) unless $lei->{auth};
-        $lei->wait_wq_events($op_c, $ops);
+        $lei->wq1_start($self);
 }
 
 sub ipc_atfork_child {
diff --git a/lib/PublicInbox/LeiRediff.pm b/lib/PublicInbox/LeiRediff.pm
index 56c457fc..f0521bcc 100644
--- a/lib/PublicInbox/LeiRediff.pm
+++ b/lib/PublicInbox/LeiRediff.pm
@@ -279,10 +279,7 @@ sub lei_rediff {
         my $isatty = -t $lei->{1};
         $lei->{opt}->{color} //= $isatty;
         $lei->start_pager if $isatty;
-        my ($op_c, $ops) = $lei->workers_start($self, 1);
-        $lei->{wq1} = $self;
-        net_merge_all_done($self) unless $lei->{auth};
-        $lei->wait_wq_events($op_c, $ops);
+        $lei->wq1_start($self);
 }
 
 sub ipc_atfork_child {
diff --git a/lib/PublicInbox/LeiRefreshMailSync.pm b/lib/PublicInbox/LeiRefreshMailSync.pm
index f516f572..7821008f 100644
--- a/lib/PublicInbox/LeiRefreshMailSync.pm
+++ b/lib/PublicInbox/LeiRefreshMailSync.pm
@@ -81,13 +81,8 @@ EOM
         my $self = bless { missing_ok => 1, lms => $lms }, __PACKAGE__;
         $lei->{opt}->{'mail-sync'} = 1; # for prepare_inputs
         $self->prepare_inputs($lei, \@folders) or return;
-        my $ops = {};
-        $lei->{auth}->op_merge($ops, $self, $lei) if $lei->{auth};
-        (my $op_c, $ops) = $lei->workers_start($self, 1, $ops);
-        $lei->{wq1} = $self;
         $lei->{-err_type} = 'non-fatal';
-        net_merge_all_done($self) unless $lei->{auth};
-        $lei->wait_wq_events($op_c, $ops); # net_merge_all_done if !{auth}
+        $lei->wq1_start($self);
 }
 
 sub ipc_atfork_child { # needed for PublicInbox::LeiPmdir
diff --git a/lib/PublicInbox/LeiRm.pm b/lib/PublicInbox/LeiRm.pm
index cc1abbff..62423ac9 100644
--- a/lib/PublicInbox/LeiRm.pm
+++ b/lib/PublicInbox/LeiRm.pm
@@ -19,11 +19,8 @@ sub lei_rm {
         $lei->{opt}->{'in-format'} //= 'eml';
         my $self = bless {}, __PACKAGE__;
         $self->prepare_inputs($lei, \@inputs) or return;
-        my ($op_c, $ops) = $lei->workers_start($self, 1);
-        $lei->{wq1} = $self;
         $lei->{-err_type} = 'non-fatal';
-        net_merge_all_done($self) unless $lei->{auth};
-        $lei->wait_wq_events($op_c, $ops);
+        $lei->wq1_start($self);
 }
 
 no warnings 'once';
diff --git a/lib/PublicInbox/LeiTag.pm b/lib/PublicInbox/LeiTag.pm
index d64a9f86..2dc59f70 100644
--- a/lib/PublicInbox/LeiTag.pm
+++ b/lib/PublicInbox/LeiTag.pm
@@ -37,13 +37,8 @@ sub lei_tag { # the "lei tag" method
         $self->prepare_inputs($lei, \@argv) or return;
         grep(defined, @$vmd_mod{qw(+kw +L -L -kw)}) or
                 return $lei->fail('no keywords or labels specified');
-        my $ops = {};
-        $lei->{auth}->op_merge($ops, $self, $lei) if $lei->{auth};
-        (my $op_c, $ops) = $lei->workers_start($self, 1, $ops);
-        $lei->{wq1} = $self;
         $lei->{-err_type} = 'non-fatal';
-        net_merge_all_done($self) unless $lei->{auth};
-        $lei->wait_wq_events($op_c, $ops);
+        $lei->wq1_start($self);
 }
 
 sub note_unimported {
diff --git a/lib/PublicInbox/LeiUp.pm b/lib/PublicInbox/LeiUp.pm
index 79639d5e..66d950b2 100644
--- a/lib/PublicInbox/LeiUp.pm
+++ b/lib/PublicInbox/LeiUp.pm
@@ -136,10 +136,7 @@ EOM
         if ($lei->{auth}) { # start auth worker
                 require PublicInbox::NetWriter;
                 bless $lei->{net}, 'PublicInbox::NetWriter';
-                $lei->{auth}->op_merge(my $ops = {}, $self, $lei);
-                (my $op_c, $ops) = $lei->workers_start($self, 1, $ops);
-                $lei->{wq1} = $self;
-                $lei->wait_wq_events($op_c, $ops);
+                $lei->wq1_start($self);
                 # net_merge_all_done will fire when auth is done
         } else {
                 redispatch_all($self, $lei); # see below