use PublicInbox::Eml;
use PublicInbox::Lock;
use PublicInbox::ProcessPipe;
-use PublicInbox::Spawn qw(which spawn popen_rd);
+use PublicInbox::Spawn qw(spawn);
use PublicInbox::LeiDedupe;
use PublicInbox::Git;
use PublicInbox::GitAsyncCat;
$lei->fail("@$cmd failed", $? >> 8);
}
-# all of these support -c for stdout and -d for decompression,
-# mutt is commonly distributed with hooks for gz, bz2 and xz, at least
-# { foo => '' } means "--foo" is passed to the command-line,
-# otherwise { foo => '--bar' } passes "--bar"
-our %zsfx2cmd = (
- gz => [ qw(GZIP pigz gzip), { rsyncable => '' } ],
- bz2 => [ 'bzip2', {} ],
- 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
- # rsyncable => '', threads => '-T' } ],
- # zz => [ 'pigz', { -default => [ '--zlib' ],
- # rsyncable => '', threads => '-p' }],
- # lzo => [ 'lzop', {} ],
- # lzma => [ 'lzma', {} ],
-);
-
-sub zsfx2cmd ($$$) {
- my ($zsfx, $decompress, $lei) = @_;
- my $x = $zsfx2cmd{$zsfx} // die "no support for suffix=.$zsfx";
- my @info = @$x;
- my $cmd_opt = pop @info;
- my @cmd = (undef, $decompress ? qw(-dc) : qw(-c));
- for my $exe (@info) {
- # I think respecting client's ENV{GZIP} is OK, not sure
- # about ENV overrides for other, less-common compressors
- if ($exe eq uc($exe)) {
- $exe = $lei->{env}->{$exe} or next;
- }
- $cmd[0] = which($exe) and last;
- }
- $cmd[0] // die join(' or ', @info)." missing for .$zsfx";
- # push @cmd, @{$cmd_opt->{-default}} if $cmd_opt->{-default};
- for my $bool (qw(rsyncable)) {
- my $switch = $cmd_opt->{rsyncable} // next;
- push @cmd, '--'.($switch || $bool);
- }
- for my $key (qw(rsyncable)) { # support compression level?
- my $switch = $cmd_opt->{$key} // next;
- my $val = $lei->{opt}->{$key} // next;
- push @cmd, $switch, $val;
- }
- \@cmd;
-}
-
sub _post_augment_mbox { # open a compressor process
my ($self, $lei) = @_;
my $zsfx = $self->{zsfx} or return;
- my $cmd = zsfx2cmd($zsfx, undef, $lei);
+ my $cmd = PublicInbox::MboxReader::zsfx2cmd($zsfx, undef, $lei);
my ($r, $w) = @{delete $lei->{zpipe}};
my $rdr = { 0 => $r, 1 => $lei->{1}, 2 => $lei->{2} };
my $pid = spawn($cmd, undef, $rdr);
$lei->{1} = $pp;
}
-sub decompress_src ($$$) {
- my ($in, $zsfx, $lei) = @_;
- my $cmd = zsfx2cmd($zsfx, 1, $lei);
- popen_rd($cmd, undef, { 0 => $in, 2 => $lei->{2} });
-}
-
sub dup_src ($) {
my ($in) = @_;
open my $dup, '+>>&', $in or die "dup: $!";
sub update_kw_maybe ($$$$) {
my ($lei, $lse, $eml, $kw) = @_;
return unless $lse;
- my $x = $lse->kw_changed($eml, $kw);
+ my $c = $lse->kw_changed($eml, $kw, my $docids = []);
my $vmd = { kw => $kw };
- if ($x) {
+ if (scalar @$docids) { # already in lei/store
+ $lei->{sto}->ipc_do('set_eml_vmd', undef, $vmd, $docids) if $c;
+ } elsif (my $xoids = $lei->{ale}->xoids_for($eml)) {
+ # it's in an external, only set kw, here
+ $lei->{sto}->ipc_do('set_xvmd', $xoids, $eml, $vmd);
+ } else { # never-before-seen, import the whole thing
+ # XXX this is critical in protecting against accidental
+ # data loss without --augment
$lei->{sto}->ipc_do('set_eml', $eml, $vmd);
- } elsif (!defined($x)) {
- if (my $xoids = $lei->{ale}->xoids_for($eml)) {
- $lei->{sto}->ipc_do('set_xvmd', $xoids, $eml, $vmd);
- } else {
- $lei->{sto}->ipc_do('set_eml', $eml, $vmd);
- }
}
}
sub _augment_or_unlink { # maildir_each_eml cb
my ($f, $kw, $eml, $lei, $lse, $mod, $shard, $unlink) = @_;
if ($mod) {
- # can't get dirent.d_ino w/ pure Perl, so we extract the OID
- # if it looks like one:
+ # can't get dirent.d_ino w/ pure Perl readdir, so we extract
+ # the OID if it looks like one instead of doing stat(2)
my $hex = $f =~ m!\b([a-f0-9]{40,})[^/]*\z! ?
$1 : sha256_hex($f);
my $recno = hex(substr($hex, 0, 8));
sub _pre_augment_mbox {
my ($self, $lei) = @_;
my $dst = $lei->{ovv}->{dst};
- my $out = $lei->{1};
- if ($dst ne '/dev/stdout') {
+ my $out;
+ my $devfd = $lei->path_to_fd($dst) // die "bad $dst";
+ if ($devfd >= 0) {
+ $out = $lei->{$devfd};
+ } else { # normal-looking path
if (-p $dst) {
- $out = undef;
open $out, '>', $dst or die "open($dst): $!";
} elsif (-f _ || !-e _) {
require PublicInbox::MboxLock;
PublicInbox::MboxLock->defaults;
$self->{mbl} = PublicInbox::MboxLock->acq($dst, 1, $m);
$out = $self->{mbl}->{fh};
+ } else {
+ die "$dst is not a file or FIFO\n";
}
$lei->{old_1} = $lei->{1}; # keep for spawning MUA
}
# Perl does SEEK_END even with O_APPEND :<
$self->{seekable} = seek($out, 0, SEEK_SET);
- if (!$self->{seekable} && $! != ESPIPE && $dst ne '/dev/stdout') {
+ if (!$self->{seekable} && $! != ESPIPE && !defined($devfd)) {
die "seek($dst): $!\n";
}
if (!$self->{seekable}) {
die "--augment specified but $dst is not seekable\n" if
$lei->{opt}->{augment};
}
- state $zsfx_allow = join('|', keys %zsfx2cmd);
- if (($self->{zsfx}) = ($dst =~ /\.($zsfx_allow)\z/)) {
+ if ($self->{zsfx} = PublicInbox::MboxReader::zsfx($dst)) {
pipe(my ($r, $w)) or die "pipe: $!";
$lei->{zpipe} = [ $r, $w ];
$lei->{ovv}->{lock_path} and
return;
}
my $zsfx = $self->{zsfx};
- my $rd = $zsfx ? decompress_src($out, $zsfx, $lei) : dup_src($out);
+ my $rd = $zsfx ? PublicInbox::MboxReader::zsfxcat($out, $zsfx, $lei)
+ : dup_src($out);
my $dedupe;
if ($opt->{augment}) {
$dedupe = $lei->{dedupe};
my $shard = $self->{-wq_worker_nr};
if (my $net = $lei->{net}) {
$net->{shard_info} = [ $mod, $shard ];
- } else { # Maildir (MH?)
+ } else { # Maildir
$self->{shard_info} = [ $mod, $shard ];
}
$aug = '+'; # incr_post_augment
}
sub lock_free {
- $_[0]->{base_type} =~ /\A(?:maildir|mh|imap|jmap)\z/ ? 1 : 0;
+ $_[0]->{base_type} =~ /\A(?:maildir|imap|jmap)\z/ ? 1 : 0;
}
sub poke_dst {