diff options
| -rw-r--r-- | lib/PublicInbox/LeiToMail.pm | 99 | ||||
| -rw-r--r-- | lib/PublicInbox/ProcessPipe.pm | 21 | ||||
| -rw-r--r-- | t/lei_to_mail.t | 47 |
3 files changed, 158 insertions, 9 deletions
diff --git a/lib/PublicInbox/LeiToMail.pm b/lib/PublicInbox/LeiToMail.pm index b0d4b664..ebb50c50 100644 --- a/lib/PublicInbox/LeiToMail.pm +++ b/lib/PublicInbox/LeiToMail.pm @@ -6,6 +6,12 @@ package PublicInbox::LeiToMail; use strict; use v5.10.1; use PublicInbox::Eml; +use PublicInbox::Lock; +use PublicInbox::ProcessPipe; +use PublicInbox::Spawn qw(which spawn); +use Symbol qw(gensym); +use File::Temp (); +use IO::Handle; # ->autoflush my %kw2char = ( # Maildir characters draft => 'D', @@ -45,10 +51,14 @@ sub _mbox_hdr_buf ($$$) { $buf; } -sub write_in_full_atomic ($$) { - my ($fh, $buf) = @_; - defined(my $w = syswrite($fh, $$buf)) or die "write: $!"; - $w == length($$buf) or die "short write: $w != ".length($$buf); +sub write_in_full ($$$) { + my ($fh, $buf, $atomic) = @_; + if ($atomic) { + defined(my $w = syswrite($fh, $$buf)) or die "write: $!"; + $w == length($$buf) or die "short write: $w != ".length($$buf); + } else { + print $fh $$buf or die "print: $!"; + } } sub eml2mboxrd ($;$) { @@ -106,4 +116,85 @@ sub eml2mboxcl2 { $buf; } +sub mkmaildir ($) { + my ($maildir) = @_; + for (qw(new tmp cur)) { + my $d = "$maildir/$_"; + next if -d $d; + require File::Path; + if (!File::Path::mkpath($d) && !-d $d) { + die "failed to mkpath($d): $!\n"; + } + } +} + +sub git_to_mail { # git->cat_async callback + my ($bref, $oid, $type, $size, $arg) = @_; + if ($type ne 'blob') { + if ($type eq 'missing') { + warn "missing $oid\n"; + } else { + warn "unexpected type=$type for $oid\n"; + } + } + if ($size > 0) { + my ($write_cb, $kw) = @$arg; + $write_cb->($bref, $oid, $kw); + } +} + +sub reap_compress { # dwaitpid callback + my ($lei, $pid) = @_; + my $cmd = delete $lei->{"pid.$pid"}; + return if $? == 0; + $lei->fail("@$cmd failed", $? >> 8); +} + +sub compress_dst { + my ($out, $sfx, $lei) = @_; + my $cmd = []; + if ($sfx eq 'gz') { + $cmd->[0] = which($lei->{env}->{GZIP} // 'pigz') // + which('gzip') // + die "pigz or gzip missing for $sfx\n"; + # TODO: use IO::Compress::Gzip + push @$cmd, '-c'; # stdout + push @$cmd, '--rsyncable' if $lei->{opt}->{rsyncable}; + } else { + die "TODO $sfx" + } + pipe(my ($r, $w)) or die "pipe: $!"; + my $rdr = { 0 => $r, 1 => $out, 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 $tmp = File::Temp->new("$sfx.lock-XXXXXX", TMPDIR => 1); + my $pipe_lk = ($lei->{opt}->{jobs} // 0) > 1 ? bless({ + lock_path => $tmp->filename, + tmp => $tmp + }, 'PublicInbox::Lock') : undef; + ($pp, $pipe_lk); +} + +sub write_cb { + my ($cls, $dst, $lei) = @_; + if ($dst =~ s!\A(mbox(?:rd|cl|cl2|o))?:!!) { + my $m = "eml2$1"; + my $eml2mbox = $cls->can($m) or die "$cls->$m missing"; + my ($out, $pipe_lk); + open $out, '>>', $dst or die "open $dst: $!"; + my $atomic = !!(($lei->{opt}->{jobs} // 0) > 1); + if ($dst =~ /\.(gz|bz2|xz)\z/) { + ($out, $pipe_lk) = compress_dst($out, $1, $lei); + } + sub { + my ($buf, $oid, $kw) = @_; + $buf = $eml2mbox->(PublicInbox::Eml->new($buf), $kw); + my $lock = $pipe_lk->lock_for_scope if $pipe_lk; + write_in_full($out, $buf, $atomic); + } + } +} + 1; diff --git a/lib/PublicInbox/ProcessPipe.pm b/lib/PublicInbox/ProcessPipe.pm index 2ce7eb8f..c9234f42 100644 --- a/lib/PublicInbox/ProcessPipe.pm +++ b/lib/PublicInbox/ProcessPipe.pm @@ -4,28 +4,39 @@ # a tied handle for auto reaping of children tied to a pipe, see perltie(1) package PublicInbox::ProcessPipe; use strict; -use warnings; +use v5.10.1; sub TIEHANDLE { - my ($class, $pid, $fh) = @_; - bless { pid => $pid, fh => $fh }, $class; + my ($class, $pid, $fh, $cb, $arg) = @_; + bless { pid => $pid, fh => $fh, cb => $cb, arg => $arg }, $class; } sub READ { read($_[0]->{fh}, $_[1], $_[2], $_[3] || 0) } sub READLINE { readline($_[0]->{fh}) } +sub WRITE { + use bytes qw(length); + syswrite($_[0]->{fh}, $_[1], $_[2] // length($_[1]), $_[3] // 0); +} + +sub PRINT { + my $self = shift; + print { $self->{fh} } @_; +} + sub CLOSE { my $fh = delete($_[0]->{fh}); my $ret = defined $fh ? close($fh) : ''; - my $pid = delete $_[0]->{pid}; + my ($pid, $cb, $arg) = delete @{$_[0]}{qw(pid cb arg)}; if (defined $pid) { # PublicInbox::DS may not be loaded - eval { PublicInbox::DS::dwaitpid($pid, undef, undef) }; + eval { PublicInbox::DS::dwaitpid($pid, $cb, $arg) }; if ($@) { # ok, not in the event loop, work synchronously waitpid($pid, 0); $ret = '' if $?; + $cb->($arg, $pid) if $cb; } } $ret; diff --git a/t/lei_to_mail.t b/t/lei_to_mail.t index 089a422e..231cf543 100644 --- a/t/lei_to_mail.t +++ b/t/lei_to_mail.t @@ -62,4 +62,51 @@ for my $mbox (qw(mboxrd mboxo mboxcl mboxcl2)) { } } +my ($tmpdir, $for_destroy) = tmpdir(); +local $ENV{TMPDIR} = $tmpdir; +open my $err, '>>', "$tmpdir/lei.err" or BAIL_OUT $!; +my $lei = { 2 => $err }; +my $buf = <<'EOM'; +From: x@example.com +Subject: x + +blah +EOM +my $fn = "$tmpdir/x.mbox"; +my $orig = do { + my $wcb = PublicInbox::LeiToMail->write_cb("mboxcl2:$fn", $lei); + is(ref $wcb, 'CODE', 'write_cb returned callback'); + ok(-f $fn && !-s _, 'empty file created'); + $wcb->(\(my $dup = $buf), 'deadbeef', [ qw(seen) ]); + undef $wcb; + open my $fh, '<', $fn or BAIL_OUT $!; + my $raw = do { local $/; <$fh> }; + like($raw, qr/^blah\n/sm, 'wrote content'); + unlink $fn or BAIL_OUT $!; + + local $lei->{opt} = { jobs => 2 }; + $wcb = PublicInbox::LeiToMail->write_cb("mboxcl2:$fn", $lei); + $wcb->(\($dup = $buf), 'deadbeef', [ qw(seen) ]); + undef $wcb; + open $fh, '<', $fn or BAIL_OUT $!; + is($raw, do { local $/; <$fh> }, 'jobs > 1'); + $raw; +}; +SKIP: { + use PublicInbox::Spawn qw(which); + my $gzip = which('gzip') or skip 'gzip not found', 1; + my $wcb = PublicInbox::LeiToMail->write_cb("mboxcl2:$fn.gz", $lei); + $wcb->(\(my $dup = $buf), 'deadbeef', [ qw(seen) ]); + undef $wcb; + my $uncompressed = xqx([$gzip, '-dc', "$fn.gz"]); + is($uncompressed, $orig, 'gzip works'); + + local $lei->{opt} = { jobs => 2 }; + unlink "$fn.gz" or die "unlink $!"; + $wcb = PublicInbox::LeiToMail->write_cb("mboxcl2:$fn.gz", $lei); + $wcb->(\(my $dupe = $buf), 'deadbeef', [ qw(seen) ]); + undef $wcb; + is(xqx([$gzip, '-dc', "$fn.gz"]), $orig); +} + done_testing; |