use PublicInbox::OnDestroy;
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);
+my ($maildir_each_file);
# struggles with short-lived repos, Gcf2Client makes little sense with lei;
# but we may use in-process libgit2 in the future.
}
}
-sub maildir_each_file ($$;@) {
- my ($dir, $cb, @arg) = @_;
- $dir .= '/' unless substr($dir, -1) eq '/';
- for my $d (qw(new/ cur/)) {
- my $pfx = $dir.$d;
- opendir my $dh, $pfx or next;
- while (defined(my $fn = readdir($dh))) {
- $cb->($pfx.$fn, @arg) if $fn =~ /:2,[A-Za-z]*\z/;
- }
- }
-}
-
sub _augment_file { # maildir_each_file cb
- my ($f, $lei) = @_;
+ my ($f, $lei, $mod, $shard) = @_;
+ 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;
+ }
my $eml = PublicInbox::InboxWritable::eml_from_path($f) or return;
_augment($eml, $lei);
}
$rand = '';
do {
$final = $dst.$rand.$common.':2,'.$sfx;
- } while (!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";
} else {
my $err = "Error writing $smsg->{blob} to $dst: $!\n";
}
}
+sub _imap_write_cb ($$) {
+ my ($self, $lei) = @_;
+ my $dedupe = $lei->{dedupe};
+ $dedupe->prepare_dedupe if $dedupe;
+ my $imap_append = $lei->{nwr}->can('imap_append');
+ my $mic = $lei->{nwr}->mic_get($self->{uri});
+ my $folder = $self->{uri}->mailbox;
+ sub { # for git_to_mail
+ my ($bref, $smsg, $eml) = @_;
+ $mic // return $lei->fail; # dst may be undef-ed in last run
+ if ($dedupe) {
+ $eml //= PublicInbox::Eml->new($$bref); # copy bref
+ return if $dedupe->is_dup($eml, $smsg->{blob});
+ }
+ eval { $imap_append->($mic, $folder, $bref, $smsg, $eml) };
+ if (my $err = $@) {
+ undef $mic;
+ die $err;
+ }
+ }
+}
+
sub write_cb { # returns a callback for git_to_mail
my ($self, $lei) = @_;
- # _mbox_write_cb or _maildir_write_cb
+ # _mbox_write_cb, _maildir_write_cb or _imap_write_cb
my $m = "_$self->{base_type}_write_cb";
$self->$m($lei);
}
my $dst = $lei->{ovv}->{dst};
my $self = bless {}, $cls;
if ($fmt eq 'maildir') {
+ $maildir_each_file //= do {
+ require PublicInbox::MdirReader;
+ PublicInbox::MdirReader->can('maildir_each_file');
+ };
+ $lei->{opt}->{augment} and
+ require PublicInbox::InboxWritable; # eml_from_path
$self->{base_type} = 'maildir';
-e $dst && !-d _ and die
"$dst exists and is not a directory\n";
$lei->{ovv}->{dst} = $dst .= '/' if substr($dst, -1) ne '/';
} elsif (substr($fmt, 0, 4) eq 'mbox') {
+ require PublicInbox::MboxReader if $lei->{opt}->{augment};
(-d $dst || (-e _ && !-w _)) and die
"$dst exists and is not a writable file\n";
- $self->can("eml2$fmt") or die "bad mbox --format=$fmt\n";
+ $self->can("eml2$fmt") or die "bad mbox format: $fmt\n";
$self->{base_type} = 'mbox';
+ } elsif ($fmt =~ /\Aimaps?\z/) { # TODO .onion support
+ require PublicInbox::NetWriter;
+ my $nwr = PublicInbox::NetWriter->new;
+ $nwr->add_url($dst);
+ $nwr->{quiet} = $lei->{opt}->{quiet};
+ my $err = $nwr->errors($dst);
+ return $lei->fail($err) if $err;
+ require PublicInbox::URIimap; # TODO: URI cast early
+ $self->{uri} = PublicInbox::URIimap->new($dst);
+ $self->{uri}->mailbox or die "No mailbox: $dst";
+ $lei->{nwr} = $nwr;
+ $self->{base_type} = 'imap';
} else {
die "bad mail --format=$fmt\n";
}
$self;
}
-sub _pre_augment_maildir {} # noop
+sub _pre_augment_maildir {
+ my ($self, $lei) = @_;
+ my $dst = $lei->{ovv}->{dst};
+ for my $x (qw(tmp new cur)) {
+ my $d = $dst.$x;
+ next if -d $d;
+ require File::Path;
+ File::Path::mkpath($d);
+ -d $d or die "$d is not a directory";
+ }
+}
sub _do_augment_maildir {
my ($self, $lei) = @_;
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);
+ my ($mod, $shard) = @{$self->{shard_info} // []};
+ $maildir_each_file->($dst, \&_augment_file,
+ $lei, $mod, $shard);
$dedupe->pause_dedupe;
}
} else { # clobber existing Maildir
- maildir_each_file($dst, \&_unlink);
+ $maildir_each_file->($dst, \&_unlink);
}
}
-sub _post_augment_maildir {
+sub _augment_imap { # PublicInbox::NetReader::imap_each cb
+ my ($url, $uid, $kw, $eml, $lei) = @_;
+ _augment($eml, $lei);
+}
+
+sub _do_augment_imap {
my ($self, $lei) = @_;
- my $dst = $lei->{ovv}->{dst};
- for my $x (qw(tmp new cur)) {
- my $d = $dst.$x;
- next if -d $d;
- require File::Path;
- File::Path::mkpath($d);
- -d $d or die "$d is not a directory";
+ my $nwr = $lei->{nwr};
+ if ($lei->{opt}->{augment}) {
+ my $dedupe = $lei->{dedupe};
+ if ($dedupe && $dedupe->prepare_dedupe) {
+ $nwr->imap_each($self->{uri}, \&_augment_imap, $lei);
+ $dedupe->pause_dedupe;
+ }
+ } else { # clobber existing IMAP folder
+ $nwr->imap_delete_all($self->{uri});
}
}
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:
$dedupe->pause_dedupe if $dedupe;
}
-sub pre_augment { # fast (1 disk seek), runs in main daemon
+sub pre_augment { # fast (1 disk seek), runs in same process as post_augment
my ($self, $lei) = @_;
# _pre_augment_maildir, _pre_augment_mbox
- my $m = "_pre_augment_$self->{base_type}";
- $self->$m($lei);
+ my $m = $self->can("_pre_augment_$self->{base_type}") or return;
+ $m->($self, $lei);
}
sub do_augment { # slow, runs in wq worker
my ($self, $lei) = @_;
- # _do_augment_maildir, _do_augment_mbox
+ # _do_augment_maildir, _do_augment_mbox, or _do_augment_imap
my $m = "_do_augment_$self->{base_type}";
$self->$m($lei);
}
-sub post_augment { # fast (spawn compressor or mkdir), runs in main daemon
+# fast (spawn compressor or mkdir), runs in same process as pre_augment
+sub post_augment {
my ($self, $lei, @args) = @_;
- # _post_augment_maildir, _post_augment_mbox
- my $m = "_post_augment_$self->{base_type}";
- $self->$m($lei, @args);
+ # _post_augment_mbox
+ my $m = $self->can("_post_augment_$self->{base_type}") or return;
+ $m->($self, $lei, @args);
}
sub ipc_atfork_child {
my ($self) = @_;
my $lei = delete $self->{lei};
$lei->lei_atfork_child;
+ my $aug;
+ if (lock_free($self)) {
+ my $mod = $self->{-wq_nr_workers};
+ my $shard = $self->{-wq_worker_nr};
+ if (my $nwr = $lei->{nwr}) {
+ $nwr->{shard_info} = [ $mod, $shard ];
+ } else { # Maildir (MH?)
+ $self->{shard_info} = [ $mod, $shard ];
+ }
+ $aug = '+'; # incr_post_augment
+ } elsif ($self->{-wq_worker_nr} == 0) {
+ $aug = '.'; # do_post_augment
+ }
+ if ($aug) {
+ local $0 = 'do_augment';
+ eval { do_augment($self, $lei) };
+ $lei->fail($@) if $@;
+ pkt_do($lei->{pkt_op_p}, $aug) == 1 or
+ die "do_post_augment trigger: $!";
+ }
if (my $zpipe = delete $lei->{zpipe}) {
$lei->{1} = $zpipe->[1];
close $zpipe->[0];