]> Sergey Matveev's repositories - public-inbox.git/blobdiff - lib/PublicInbox/LeiToMail.pm
config: lazy-load coderepos, support extindex
[public-inbox.git] / lib / PublicInbox / LeiToMail.pm
index b3228a59e31f4219e0a5b8cf83f9ac8608bd91e5..27e1338f7f1bb57305dc43b67cc1fde5ede59b8b 100644 (file)
@@ -170,9 +170,9 @@ sub reap_compress { # dwaitpid callback
 # { foo => '' } means "--foo" is passed to the command-line,
 # otherwise { foo => '--bar' } passes "--bar"
 our %zsfx2cmd = (
-       gz => [ qw(GZIP pigz gzip), { rsyncable => '', threads => '-p' } ],
+       gz => [ qw(GZIP pigz gzip), { rsyncable => '' } ],
        bz2 => [ 'bzip2', {} ],
-       xz => [ 'xz', { threads => '-T' } ],
+       xz => [ 'xz', {} ],
        # XXX does anybody care for these?  I prefer zstd on entire FSes,
        # so it's probably not necessary on a per-file basis
        # zst => [ 'zstd', { -default => [ qw(-q) ], # it's noisy by default
@@ -203,7 +203,7 @@ sub zsfx2cmd ($$$) {
                my $switch = $cmd_opt->{rsyncable} // next;
                push @cmd, '--'.($switch || $bool);
        }
-       for my $key (qw(threads)) { # support compression level?
+       for my $key (qw(rsyncable)) { # support compression level?
                my $switch = $cmd_opt->{$key} // next;
                my $val = $lei->{opt}->{$key} // next;
                push @cmd, $switch, $val;
@@ -223,8 +223,6 @@ sub _post_augment_mbox { # open a compressor process
        $dup->{$_} = $lei->{$_} for qw(2 sock);
        tie *$pp, 'PublicInbox::ProcessPipe', $pid, $w, \&reap_compress, $dup;
        $lei->{1} = $pp;
-       die 'BUG: unexpected {ovv}->{lock_path}' if $lei->{ovv}->{lock_path};
-       $lei->{ovv}->ovv_out_lk_init;
 }
 
 sub decompress_src ($$$) {
@@ -246,6 +244,13 @@ sub _augment { # MboxReader eml_cb
        $lei->{dedupe}->is_dup($eml);
 }
 
+sub _mbox_augment_kw_maybe {
+       my ($eml, $lei, $lse, $augment) = @_;
+       my $kw = PublicInbox::MboxReader::mbox_keywords($eml);
+       update_kw_maybe($lei, $lse, $eml, $kw);
+       _augment($eml, $lei) if $augment;
+}
+
 sub _mbox_write_cb ($$) {
        my ($self, $lei) = @_;
        my $ovv = $lei->{ovv};
@@ -274,7 +279,7 @@ sub update_kw_maybe ($$$$) {
        if ($x) {
                $lei->{sto}->ipc_do('set_eml', $eml, @$kw);
        } elsif (!defined($x)) {
-               # TODO: xkw
+               $lei->{sto}->ipc_do('set_xkw', $eml, $kw);
        }
 }
 
@@ -391,7 +396,7 @@ sub new {
                                "$dst exists and is not a directory\n";
                $lei->{ovv}->{dst} = $dst .= '/' if substr($dst, -1) ne '/';
        } elsif (substr($fmt, 0, 4) eq 'mbox') {
-               require PublicInbox::MboxReader if $lei->{opt}->{augment};
+               require PublicInbox::MboxReader;
                (-d $dst || (-e _ && !-w _)) and die
                        "$dst exists and is not a writable file\n";
                $self->can("eml2$fmt") or die "bad mbox format: $fmt\n";
@@ -431,7 +436,7 @@ sub _pre_augment_maildir {
 sub _do_augment_maildir {
        my ($self, $lei) = @_;
        my $dst = $lei->{ovv}->{dst};
-       my $lse = $lei->{sto}->search if $lei->{opt}->{'import-augment'};
+       my $lse = $lei->{sto}->search if $lei->{opt}->{'import-before'};
        my ($mod, $shard) = @{$self->{shard_info} // []};
        if ($lei->{opt}->{augment}) {
                my $dedupe = $lei->{dedupe};
@@ -463,7 +468,7 @@ sub _imap_augment_or_delete { # PublicInbox::NetReader::imap_each cb
 sub _do_augment_imap {
        my ($self, $lei) = @_;
        my $net = $lei->{net};
-       my $lse = $lei->{sto}->search if $lei->{opt}->{'import-augment'};
+       my $lse = $lei->{sto}->search if $lei->{opt}->{'import-before'};
        if ($lei->{opt}->{augment}) {
                my $dedupe = $lei->{dedupe};
                if ($dedupe && $dedupe->prepare_dedupe) {
@@ -485,9 +490,10 @@ sub _do_augment_imap {
 sub _pre_augment_mbox {
        my ($self, $lei) = @_;
        my $dst = $lei->{ovv}->{dst};
+       my $out = $lei->{1};
        if ($dst ne '/dev/stdout') {
-               my $out;
                if (-p $dst) {
+                       $out = undef;
                        open $out, '>', $dst or die "open($dst): $!";
                } elsif (-f _ || !-e _) {
                        require PublicInbox::MboxLock;
@@ -495,36 +501,61 @@ sub _pre_augment_mbox {
                                        PublicInbox::MboxLock->defaults;
                        $self->{mbl} = PublicInbox::MboxLock->acq($dst, 1, $m);
                        $out = $self->{mbl}->{fh};
-                       if (!$lei->{opt}->{augment} and !truncate($out, 0)) {
-                               die "truncate($dst): $!";
-                       }
                }
                $lei->{old_1} = $lei->{1}; # keep for spawning MUA
-               $lei->{1} = $out;
        }
        # Perl does SEEK_END even with O_APPEND :<
-       $self->{seekable} = seek($lei->{1}, 0, SEEK_SET);
+       $self->{seekable} = seek($out, 0, SEEK_SET);
        if (!$self->{seekable} && $! != ESPIPE && $dst ne '/dev/stdout') {
                die "seek($dst): $!\n";
        }
+       if (!$self->{seekable}) {
+               my $ia = $lei->{opt}->{'import-before'};
+               die "--import-before specified but $dst is not seekable\n"
+                       if $ia && !ref($ia);
+               die "--augment specified but $dst is not seekable\n" if
+                       $lei->{opt}->{augment};
+       }
        state $zsfx_allow = join('|', keys %zsfx2cmd);
-       ($self->{zsfx}) = ($dst =~ /\.($zsfx_allow)\z/) or return;
-       pipe(my ($r, $w)) or die "pipe: $!";
-       $lei->{zpipe} = [ $r, $w ];
+       if (($self->{zsfx}) = ($dst =~ /\.($zsfx_allow)\z/)) {
+               pipe(my ($r, $w)) or die "pipe: $!";
+               $lei->{zpipe} = [ $r, $w ];
+               $lei->{ovv}->{lock_path} and
+                       die 'BUG: unexpected {ovv}->{lock_path}';
+               $lei->{ovv}->ovv_out_lk_init;
+       } elsif (!$self->{seekable} && !$lei->{ovv}->{lock_path}) {
+               $lei->{ovv}->ovv_out_lk_init;
+       }
+       $lei->{1} = $out;
+       undef;
 }
 
 sub _do_augment_mbox {
        my ($self, $lei) = @_;
-       return if !$lei->{opt}->{augment};
-       my $dedupe = $lei->{dedupe};
-       my $dst = $lei->{ovv}->{dst};
-       die "cannot augment $dst, not seekable\n" if !$self->{seekable};
+       return unless $self->{seekable};
+       my $opt = $lei->{opt};
        my $out = $lei->{1};
-       if (-s $out && $dedupe && $dedupe->prepare_dedupe) {
-               my $zsfx = $self->{zsfx};
-               my $rd = $zsfx ? decompress_src($out, $zsfx, $lei) :
-                               dup_src($out);
-               my $fmt = $lei->{ovv}->{fmt};
+       my ($fmt, $dst) = @{$lei->{ovv}}{qw(fmt dst)};
+       return unless -s $out;
+       unless ($opt->{augment} || $opt->{'import-before'}) {
+               truncate($out, 0) or die "truncate($dst): $!";
+               return;
+       }
+       my $zsfx = $self->{zsfx};
+       my $rd = $zsfx ? decompress_src($out, $zsfx, $lei) : dup_src($out);
+       my $dedupe;
+       if ($opt->{augment}) {
+               $dedupe = $lei->{dedupe};
+               $dedupe->prepare_dedupe if $dedupe;
+       }
+       if ($opt->{'import-before'}) { # the default
+               my $lse = $lei->{sto}->search;
+               PublicInbox::MboxReader->$fmt($rd, \&_mbox_augment_kw_maybe,
+                                               $lei, $lse, $opt->{augment});
+               if (!$opt->{augment} and !truncate($out, 0)) {
+                       die "truncate($dst): $!";
+               }
+       } else { # --augment --no-import-before
                PublicInbox::MboxReader->$fmt($rd, \&_augment, $lei);
        }
        # maybe some systems don't honor O_APPEND, Perl does this:
@@ -549,7 +580,7 @@ sub do_augment { # slow, runs in wq worker
 # fast (spawn compressor or mkdir), runs in same process as pre_augment
 sub post_augment {
        my ($self, $lei, @args) = @_;
-       my $wait = $lei->{opt}->{'import-augment'} ?
+       my $wait = $lei->{opt}->{'import-before'} ?
                        $lei->{sto}->ipc_do('checkpoint', 1) : 0;
        # _post_augment_mbox
        my $m = $self->can("_post_augment_$self->{base_type}") or return;