use v5.10.1;
use parent qw(PublicInbox::IPC);
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;
use PublicInbox::PktOp qw(pkt_do);
use Symbol qw(gensym);
use IO::Handle; # ->autoflush
use Fcntl qw(SEEK_SET SEEK_END O_CREAT O_EXCL O_WRONLY);
-use Errno qw(EEXIST ESPIPE ENOENT EPIPE);
-use Digest::SHA qw(sha256_hex);
-
-# struggles with short-lived repos, Gcf2Client makes little sense with lei;
-# but we may use in-process libgit2 in the future.
-$PublicInbox::GitAsyncCat::GCF2C = 0;
my %kw2char = ( # Maildir characters
draft => 'D',
flagged => 'F',
+ forwarded => 'P', # passed
answered => 'R',
- seen => 'S'
+ seen => 'S',
);
my %kw2status = (
if (defined(my $w = syswrite($lei->{1} // return, $$buf))) {
return if $w == length($$buf);
$buf = "short atomic write: $w != ".length($$buf);
- } elsif ($! == EPIPE) {
+ } elsif ($!{EPIPE}) {
return $lei->note_sigpipe(1);
} else {
$buf = "atomic write: $!";
$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:
- my $hex = $f =~ m!\b([a-f0-9]{40,})[^/]*\z! ?
- $1 : sha256_hex($f);
- my $recno = hex(substr($hex, 0, 8));
- return if ($recno % $mod) != $shard;
- update_kw_maybe($lei, $lse, $eml, $kw);
- }
+sub _md_update { # maildir_each_eml cb
+ my ($f, $kw, $eml, $lei, $lse, $unlink) = @_;
+ update_kw_maybe($lei, $lse, $eml, $kw);
$unlink ? unlink($f) : _augment($eml, $lei);
}
do {
$tmp = $dst.'tmp/'.$rand.$common;
} while (!($ok = sysopen($fh, $tmp, O_CREAT|O_EXCL|O_WRONLY)) &&
- $! == EEXIST && ($rand = _rand.','));
+ $!{EEXIST} && ($rand = _rand.','));
if ($ok && print $fh $$buf and close($fh)) {
# ignore new/ and write only to cur/, otherwise MUAs
# with R/W access to the Maildir will end up doing
$rand = '';
do {
$final = $dst.$rand.$common.':2,'.$sfx;
- } while (!($ok = link($tmp, $final)) && $! == EEXIST &&
+ } while (!($ok = link($tmp, $final)) && $!{EEXIST} &&
($rand = _rand.','));
die "link($tmp, $final): $!" unless $ok;
unlink($tmp) or warn "W: failed to unlink $tmp: $!\n";
my ($self, $lei) = @_;
my $dst = $lei->{ovv}->{dst};
my $lse = $lei->{opt}->{'import-before'} ? $lei->{lse} : undef;
- my ($mod, $shard) = @{$self->{shard_info} // []};
+ my $mdr = PublicInbox::MdirReader->new;
if ($lei->{opt}->{augment}) {
my $dedupe = $lei->{dedupe};
if ($dedupe && $dedupe->prepare_dedupe) {
- PublicInbox::MdirReader::maildir_each_eml($dst,
- \&_augment_or_unlink,
- $lei, $lse, $mod, $shard);
+ $mdr->{shard_info} = $self->{shard_info};
+ $mdr->maildir_each_eml($dst, \&_md_update, $lei, $lse);
$dedupe->pause_dedupe;
}
} elsif ($lse) {
- PublicInbox::MdirReader::maildir_each_eml($dst,
- \&_augment_or_unlink,
- $lei, $lse, $mod, $shard, 1);
+ $mdr->{shard_info} = $self->{shard_info};
+ $mdr->maildir_each_eml($dst, \&_md_update, $lei, $lse, 1);
} else {# clobber existing Maildir
- PublicInbox::MdirReader::maildir_each_file($dst, \&_unlink);
+ $mdr->maildir_each_file($dst, \&_unlink);
}
}
}
# Perl does SEEK_END even with O_APPEND :<
$self->{seekable} = seek($out, 0, SEEK_SET);
- if (!$self->{seekable} && $! != ESPIPE && !defined($devfd)) {
+ 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 {
sub write_mail { # via ->wq_io_do
my ($self, $smsg) = @_;
- git_async_cat($self->{lei}->{ale}->git, $smsg->{blob}, \&git_to_mail,
+ $self->{lei}->{ale}->git->cat_async($smsg->{blob}, \&git_to_mail,
[$self->{wcb}, $smsg]);
}