# { 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 => '', threads => '-p' } ],
bz2 => [ 'bzip2', {} ],
xz => [ 'xz', { threads => '-T' } ],
# XXX does anybody care for these? I prefer zstd on entire FSes,
}
sub compress_dst {
- my ($out, $zsfx, $lei) = @_;
+ my ($self, $zsfx, $lei) = @_;
my $cmd = zsfx2cmd($zsfx, undef, $lei);
pipe(my ($r, $w)) or die "pipe: $!";
- my $rdr = { 0 => $r, 1 => $out, 2 => $lei->{2} };
+ my $rdr = { 0 => $r, 1 => $lei->{1}, 2 => $lei->{2} };
my $pid = spawn($cmd, $lei->{env}, $rdr);
$lei->{"pid.$pid"} = $cmd;
my $pp = gensym;
tie *$pp, 'PublicInbox::ProcessPipe', $pid, $w, \&reap_compress, $lei;
- my $pipe_lk = ($lei->{opt}->{jobs} // 0) > 1 ?
- PublicInbox::Lock->new_tmp($zsfx) : undef;
- ($pp, $pipe_lk);
+ $lei->{1} = $pp;
+ die 'BUG: unexpected {ovv}->{lock_path}' if $lei->{ovv}->{lock_path};
+ $lei->{ovv}->ovv_out_lk_init if ($lei->{opt}->{jobs} // 2) > 1;
}
sub decompress_src ($$$) {
my ($in, $zsfx, $lei) = @_;
my $cmd = zsfx2cmd($zsfx, 1, $lei);
- my $rdr = { 0 => $in, 2 => $lei->{2} };
- popen_rd($cmd, $lei->{env}, $rdr);
+ popen_rd($cmd, $lei->{env}, { 0 => $in, 2 => $lei->{2} });
}
sub dup_src ($) {
$lei->{dedupe}->is_dup($eml);
}
-sub _mbox_write_cb ($$$$) {
- my ($cls, $mbox, $dst, $lei) = @_;
- my $m = "eml2$mbox";
- my $eml2mbox = $cls->can($m) or die "$cls->$m missing";
- my ($out, $pipe_lk, $seekable);
- # XXX should we support /dev/stdout.gz ?
- if ($dst eq '/dev/stdout') {
- $out = $lei->{1};
- } else { # TODO: mbox locking (but mairix doesn't...)
- my $mode = -p $dst ? '>' : '+>>';
- if (-f _ && !$lei->{opt}->{augment} and !unlink($dst)) {
- die "unlink $dst: $!" if $! != ENOENT;
- }
- open $out, $mode, $dst or die "open $dst: $!";
- # Perl does SEEK_END even with O_APPEND :<
- $seekable = seek($out, 0, SEEK_SET);
- die "seek $dst: $!\n" if !$seekable && $! != ESPIPE;
- }
- my $jobs = $lei->{opt}->{jobs} // 0;
- state $zsfx_allow = join('|', keys %zsfx2cmd);
- my ($zsfx) = ($dst =~ /\.($zsfx_allow)\z/);
- my $write = $jobs > 1 && !$zsfx ? \&atomic_append : \&_print_full;
- my $dedupe = $lei->{dedupe} = PublicInbox::LeiDedupe->new($lei);
- if ($lei->{opt}->{augment}) {
- die "cannot augment $dst, not seekable\n" if !$seekable;
- if (-s $out && $dedupe->prepare_dedupe) {
- my $rd = $zsfx ? decompress_src($out, $zsfx, $lei) :
- dup_src($out);
- PublicInbox::MboxReader->$mbox($rd, \&_augment, $lei);
- }
- # maybe some systems don't honor O_APPEND, Perl does this:
- seek($out, 0, SEEK_END) or die "seek $dst: $!";
- $dedupe->pause_dedupe if $jobs; # are we forking?
- }
- $dedupe->prepare_dedupe if !$jobs;
- ($out, $pipe_lk) = compress_dst($out, $zsfx, $lei) if $zsfx;
+sub _mbox_write_cb ($$) {
+ my ($self, $lei) = @_;
+ my $ovv = $lei->{ovv};
+ my $m = 'eml2'.$ovv->{fmt};
+ my $eml2mbox = $self->can($m) or die "$self->$m missing";
+ my $out = $lei->{1} // die "no stdout ($m, $ovv->{dst})"; # redirected earlier
+ $out->autoflush(1);
+ my $write = $ovv->{lock_path} ? \&_print_full : \&atomic_append;
+ my $dedupe = $lei->{dedupe};
+ $dedupe->prepare_dedupe;
sub { # for git_to_mail
my ($buf, $oid, $kw) = @_;
my $eml = PublicInbox::Eml->new($buf);
if (!$dedupe->is_dup($eml, $oid)) {
$buf = $eml2mbox->($eml, $kw);
- my $lock = $pipe_lk->lock_for_scope if $pipe_lk;
+ my $lk = $ovv->lock_for_scope;
$write->($out, $buf);
}
}
}
}
-
sub _maildir_write_cb ($$) {
- my ($dst, $lei) = @_;
- $dst .= '/' unless substr($dst, -1) eq '/';
- my $dedupe = $lei->{dedupe} = PublicInbox::LeiDedupe->new($lei, $dst);
- my $jobs = $lei->{opt}->{jobs} // 0;
+ my ($self, $lei) = @_;
+ my $dedupe = $lei->{dedupe};
+ $dedupe->prepare_dedupe;
+ my $dst = $lei->{ovv}->{dst};
+ sub { # for git_to_mail
+ my ($buf, $oid, $kw) = @_;
+ return _buf2maildir($dst, $buf, $oid, $kw) if !$dedupe;
+ my $eml = PublicInbox::Eml->new($$buf); # copy buf
+ return if $dedupe->is_dup($eml, $oid);
+ undef $eml;
+ _buf2maildir($dst, $buf, $oid, $kw);
+ }
+}
+
+sub write_cb { # returns a callback for git_to_mail
+ my ($self, $lei) = @_;
+ # _mbox_write_cb or _maildir_write_cb
+ my $m = "_$self->{base_type}_write_cb";
+ $self->$m($lei);
+}
+
+sub new {
+ my ($cls, $lei) = @_;
+ my $fmt = $lei->{ovv}->{fmt};
+ my $dst = $lei->{ovv}->{dst};
+ my $self = bless {}, $cls;
+ if ($fmt eq 'maildir') {
+ $self->{base_type} = 'maildir';
+ $lei->{ovv}->{dst} = $dst .= '/' if substr($dst, -1) ne '/';
+ } elsif (substr($fmt, 0, 4) eq 'mbox') {
+ $self->can("eml2$fmt") or die "bad mbox --format=$fmt\n";
+ $self->{base_type} = 'mbox';
+ } else {
+ die "bad mail --format=$fmt\n";
+ }
+ my $dedupe = $lei->{dedupe} //= PublicInbox::LeiDedupe->new($lei, $dst);
+ $self;
+}
+
+sub _prepare_maildir {
+ my ($self, $lei) = @_;
+ my $dst = $lei->{ovv}->{dst};
if ($lei->{opt}->{augment}) {
+ my $dedupe = $lei->{dedupe};
if ($dedupe && $dedupe->prepare_dedupe) {
require PublicInbox::InboxWritable; # eml_from_path
_maildir_each_file($dst, \&_augment_file, $lei);
- $dedupe->pause_dedupe if $jobs; # are we forking?
+ $dedupe->pause_dedupe;
}
} else { # clobber existing Maildir
_maildir_each_file($dst, \&_unlink);
my $d = $dst.$x;
next if -d $d;
require File::Path;
- if (!File::Path::mkpath($d) && !-d $d) {
- die "failed to mkpath($d): $!\n";
- }
- }
- $dedupe->prepare_dedupe if $dedupe && !$jobs;
- sub { # for git_to_mail
- my ($buf, $oid, $kw) = @_;
- return _buf2maildir($dst, $buf, $oid, $kw) if !$dedupe;
- my $eml = PublicInbox::Eml->new($$buf); # copy buf
- return if $dedupe->is_dup($eml, $oid);
- undef $eml;
- _buf2maildir($dst, $buf, $oid, $kw);
+ File::Path::mkpath($d) or die "mkpath($d): $!";
+ -d $d or die "$d is not a directory";
}
}
-sub write_cb { # returns a callback for git_to_mail
- my ($cls, $dst, $lei) = @_;
- require PublicInbox::LeiDedupe;
- if ($dst =~ s!\A(mbox(?:rd|cl|cl2|o))?:!!) {
- _mbox_write_cb($cls, $1, $dst, $lei);
- } elsif ($dst =~ s!\A[Mm]aildir:!!) { # typically capitalized
- _maildir_write_cb($dst, $lei);
+sub _prepare_mbox {
+ my ($self, $lei) = @_;
+ my $dst = $lei->{ovv}->{dst};
+ my ($out, $seekable);
+ if ($dst eq '/dev/stdout') {
+ $out = $lei->{1};
} else {
- undef;
+ my $mode = -p $dst ? '>' : '+>>';
+ if (-f _ && !$lei->{opt}->{augment} and !unlink($dst)) {
+ $! == ENOENT or die "unlink($dst): $!";
+ }
+ open $out, $mode, $dst or die "open($dst): $!";
+ # Perl does SEEK_END even with O_APPEND :<
+ $seekable = seek($out, 0, SEEK_SET);
+ die "seek($dst): $!\n" if !$seekable && $! != ESPIPE;
+ $lei->{1} = $out;
+ }
+ state $zsfx_allow = join('|', keys %zsfx2cmd);
+ my ($zsfx) = ($dst =~ /\.($zsfx_allow)\z/);
+ my $dedupe = $lei->{dedupe};
+ if ($lei->{opt}->{augment}) {
+ die "cannot augment $dst, not seekable\n" if !$seekable;
+ if (-s $out && $dedupe && $dedupe->prepare_dedupe) {
+ my $rd = $zsfx ? decompress_src($out, $zsfx, $lei) :
+ dup_src($out);
+ my $fmt = $lei->{ovv}->{fmt};
+ require PublicInbox::MboxReader;
+ PublicInbox::MboxReader->$fmt($rd, \&_augment, $lei);
+ }
+ # maybe some systems don't honor O_APPEND, Perl does this:
+ seek($out, 0, SEEK_END) or die "seek $dst: $!";
+ $dedupe->pause_dedupe if $dedupe;
}
- # TODO: Maildir, MH, IMAP, JMAP ...
+ compress_dst($self, $zsfx, $lei) if $zsfx;
+}
+
+sub do_prepare {
+ my ($self, $lei) = @_;
+ my $m = "_prepare_$self->{base_type}";
+ $self->$m($lei);
}
1;
use List::Util qw(shuffle);
require_mods(qw(DBD::SQLite));
require PublicInbox::MboxReader;
+require PublicInbox::LeiOverview;
use_ok 'PublicInbox::LeiToMail';
my $from = "Content-Length: 10\nSubject: x\n\nFrom hell\n";
my $noeol = "Subject: x\n\nFrom hell";
EOM
my $fn = "$tmpdir/x.mbox";
my ($mbox) = shuffle(@MBOX); # pick one, shouldn't matter
+my $wcb_get = sub {
+ my ($fmt, $dst) = @_;
+ delete $lei->{dedupe};
+ $lei->{ovv} = bless {
+ fmt => $fmt,
+ dst => $dst
+ }, 'PublicInbox::LeiOverview';
+ my $l2m = PublicInbox::LeiToMail->new($lei);
+ SKIP: {
+ require_mods('Storable', 1);
+ my $dup = Storable::thaw(Storable::freeze($l2m));
+ is_deeply($dup, $l2m, "$fmt round-trips through storable");
+ }
+ $l2m->do_prepare($lei);
+ my $cb = $l2m->write_cb($lei);
+ delete $lei->{1};
+ $cb;
+};
+
my $orig = do {
- my $wcb = PublicInbox::LeiToMail->write_cb("$mbox:$fn", $lei);
+ my $wcb = $wcb_get->($mbox, $fn);
is(ref $wcb, 'CODE', 'write_cb returned callback');
ok(-f $fn && !-s _, 'empty file created');
$wcb->(\(my $dup = $buf), 'deadbeef', [ qw(seen) ]);
unlink $fn or BAIL_OUT $!;
local $lei->{opt} = { jobs => 2 };
- $wcb = PublicInbox::LeiToMail->write_cb("$mbox:$fn", $lei);
+ $wcb = $wcb_get->($mbox, $fn);
ok(-f $fn && !-s _, 'truncated mbox destination');
- $lei->{dedupe}->prepare_dedupe;
$wcb->(\($dup = $buf), 'deadbeef', [ qw(seen) ]);
undef $wcb;
open $fh, '<', $fn or BAIL_OUT $!;
- is($raw, do { local $/; <$fh> }, 'jobs > 1');
+ is(do { local $/; <$fh> }, $raw, 'jobs > 1');
$raw;
};
for my $zsfx (qw(gz bz2 xz)) { # XXX should we support zst, zz, lzo, lzma?
my $dc_cmd = eval { $zsfx2cmd->($zsfx, 1, $lei) };
ok($dc_cmd, "decompressor for .$zsfx");
my $f = "$fn.$zsfx";
- my $dst = "$mbox:$f";
- my $wcb = PublicInbox::LeiToMail->write_cb($dst, $lei);
+ my $wcb = $wcb_get->($mbox, $f);
$wcb->(\(my $dup = $buf), 'deadbeef', [ qw(seen) ]);
undef $wcb;
my $uncompressed = xqx([@$dc_cmd, $f]);
local $lei->{opt} = { jobs => 2 }; # for atomic writes
unlink $f or BAIL_OUT "unlink $!";
- $wcb = PublicInbox::LeiToMail->write_cb($dst, $lei);
- $lei->{dedupe}->prepare_dedupe;
+ $wcb = $wcb_get->($mbox, $f);
$wcb->(\($dup = $buf), 'deadbeef', [ qw(seen) ]);
undef $wcb;
is(xqx([@$dc_cmd, $f]), $orig, "$zsfx matches with lock");
local $lei->{opt} = { augment => 1 };
- $wcb = PublicInbox::LeiToMail->write_cb($dst, $lei);
- $lei->{dedupe}->prepare_dedupe;
+ $wcb = $wcb_get->($mbox, $f);
$wcb->(\($dup = $buf . "\nx\n"), 'deadbeef', [ qw(seen) ]);
undef $wcb; # commit
like($raw[0], qr/\nblah\n\z/s, "original preserved $zsfx");
local $lei->{opt} = { augment => 1, jobs => 2 };
- $wcb = PublicInbox::LeiToMail->write_cb($dst, $lei);
- $lei->{dedupe}->prepare_dedupe;
+ $wcb = $wcb_get->($mbox, $f);
$wcb->(\($dup = $buf . "\ny\n"), 'deadbeef', [ qw(seen) ]);
undef $wcb; # commit
unlink $fn or BAIL_OUT $!;
if ('default deduplication uses content_hash') {
- my $wcb = PublicInbox::LeiToMail->write_cb("mboxo:$fn", $lei);
+ my $wcb = $wcb_get->('mboxo', $fn);
$wcb->(\(my $x = $buf), 'deadbeef', []) for (1..2);
undef $wcb; # undef to commit changes
my $cmp = '';
is($cmp, $buf, 'only one message written');
local $lei->{opt} = { augment => 1 };
- $wcb = PublicInbox::LeiToMail->write_cb("mboxo:$fn", $lei);
+ $wcb = $wcb_get->('mboxo', $fn);
$wcb->(\($x = $buf . "\nx\n"), 'deadbeef', []) for (1..2);
undef $wcb; # undef to commit changes
open $fh, '<', $fn or BAIL_OUT $!;
{ # stdout support
open my $tmp, '+>', undef or BAIL_OUT $!;
local $lei->{1} = $tmp;
- my $wcb = PublicInbox::LeiToMail->write_cb("mboxrd:/dev/stdout", $lei);
+ my $wcb = $wcb_get->('mboxrd', '/dev/stdout');
$wcb->(\(my $x = $buf), 'deadbeef', []);
undef $wcb; # commit
seek($tmp, 0, SEEK_SET) or BAIL_OUT $!;
my $fn = "$tmpdir/fifo";
mkfifo($fn, 0600) or skip("mkfifo not supported: $!", 1);
my $cat = popen_rd([which('cat'), $fn]);
- my $wcb = PublicInbox::LeiToMail->write_cb("mboxo:$fn", $lei);
+ my $wcb = $wcb_get->('mboxo', $fn);
$wcb->(\(my $x = $buf), 'deadbeef', []);
undef $wcb; # commit
my $cmp = '';
{ # Maildir support
my $md = "$tmpdir/maildir/";
- my $wcb = PublicInbox::LeiToMail->write_cb("Maildir:$md", $lei);
- ok($wcb, 'got Maildir callback');
- $lei->{dedupe}->prepare_dedupe;
+ my $wcb = $wcb_get->('maildir', $md);
+ is(ref($wcb), 'CODE', 'got Maildir callback');
$wcb->(\(my $x = $buf), 'badc0ffee', []);
- undef $wcb; # commit
my @f;
PublicInbox::LeiToMail::_maildir_each_file($md, sub { push @f, shift });
- is(scalar(@f), 1, 'wrote one file');
open my $fh, $f[0] or BAIL_OUT $!;
is(do { local $/; <$fh> }, $buf, 'wrote to Maildir');
- $wcb = PublicInbox::LeiToMail->write_cb("maildir:$md", $lei);
- $lei->{dedupe}->prepare_dedupe;
+ $wcb = $wcb_get->('maildir', $md);
$wcb->(\($x = $buf."\nx\n"), 'deadcafe', []);
- undef $wcb; # commit
my @x = ();
PublicInbox::LeiToMail::_maildir_each_file($md, sub { push @x, shift });
is(do { local $/; <$fh> }, $buf."\nx\n", 'wrote new file to Maildir');
local $lei->{opt}->{augment} = 1;
- $wcb = PublicInbox::LeiToMail->write_cb("maildir:$md", $lei);
- $lei->{dedupe}->prepare_dedupe;
+ $wcb = $wcb_get->('maildir', $md);
$wcb->(\($x = $buf."\ny\n"), 'deadcafe', []);
$wcb->(\($x = $buf."\ny\n"), 'b4dc0ffee', []); # skipped by dedupe
- undef $wcb; # commit
@f = ();
PublicInbox::LeiToMail::_maildir_each_file($md, sub { push @f, shift });
is(scalar grep(/\A\Q$x[0]\E\z/, @f), 1, 'old file still there');