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;
}
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);
}
}
}
+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);
}
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";
}
if ($lei->{opt}->{augment}) {
my $dedupe = $lei->{dedupe};
if ($dedupe && $dedupe->prepare_dedupe) {
- $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
}
}
-sub _post_augment_maildir {} # noop
+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 $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});
+ }
+}
sub _pre_augment_mbox {
my ($self, $lei) = @_;
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);
}
# 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];