lei_to_mail: start atomic and compressed mbox writing

We'll allow using multiple workers to write to a single
mbox (which could be compressed).  This is can be done
safely with O_APPEND + syswrite for uncompressed files,
and using a lock when piping to pigz/gzip/bzip2/xz.

1 files changed, 16 insertions, 5 deletions

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;