foreach my $pfx (qw(publicinboxwatch publicinboxlearn)) {
my $k = "$pfx.watchspam";
defined(my $dirs = $config->{$k}) or next;
- $dirs = [ $dirs ] if !ref($dirs);
+ $dirs = PublicInbox::Config::_array($dirs);
for my $dir (@$dirs) {
if (is_maildir($dir)) {
# skip "new", no MUA has seen it, yet.
# need to make all inboxes writable for spam removal:
my $ibx = $_[0] = PublicInbox::InboxWritable->new($_[0]);
- my $watch = $ibx->{watch} or return;
- if (is_maildir($watch)) {
- compile_watchheaders($ibx);
- my ($new, $cur) = ("$watch/new", "$watch/cur");
- return if is_watchspam($cur, $mdmap{$cur}, $ibx);
- push @mdir, $new unless $uniq{$new}++;
- push @mdir, $cur unless $uniq{$cur}++;
- push @{$mdmap{$new} ||= []}, $ibx;
- push @{$mdmap{$cur} ||= []}, $ibx;
- } elsif (my $url = imap_url($watch)) {
- return if is_watchspam($url, $imap{$url}, $ibx);
- compile_watchheaders($ibx);
- push @{$imap{$url} ||= []}, $ibx;
- } else {
- warn "watch unsupported: $k=$watch\n";
+ my $watches = $ibx->{watch} or return;
+ $watches = PublicInbox::Config::_array($watches);
+ for my $watch (@$watches) {
+ if (is_maildir($watch)) {
+ compile_watchheaders($ibx);
+ my ($new, $cur) = ("$watch/new", "$watch/cur");
+ my $cur_dst = $mdmap{$cur} //= [];
+ return if is_watchspam($cur, $cur_dst, $ibx);
+ push @mdir, $new unless $uniq{$new}++;
+ push @mdir, $cur unless $uniq{$cur}++;
+ push @{$mdmap{$new} //= []}, $ibx;
+ push @$cur_dst, $ibx;
+ } elsif (my $url = imap_url($watch)) {
+ return if is_watchspam($url, $imap{$url}, $ibx);
+ compile_watchheaders($ibx);
+ push @{$imap{$url} ||= []}, $ibx;
+ } else {
+ warn "watch unsupported: $k=$watch\n";
+ }
}
});
return unless scalar(@mdir) || scalar(keys %imap);
}
if (my $idle_mic = $self->{idle_mic}) {
eval { $idle_mic->done };
- warn "IDLE DONE error: $@\n" if $@;
- eval { $idle_mic->disconnect };
- warn "IDLE LOGOUT error: $@\n" if $@;
+ if ($@) {
+ warn "IDLE DONE error: $@\n";
+ eval { $idle_mic->disconnect };
+ warn "IDLE LOGOUT error: $@\n" if $@;
+ }
}
}
$uri->scheme . '://' . $uri->authority;
}
-sub cfg_intvl ($$) {
- my ($cfg, $key) = @_;
- defined(my $v = $cfg->{lc($key)}) or return;
+sub cfg_intvl ($$$$$) {
+ my ($cfg, $cfg_section, $cfg_key, $imap_section, $url) = @_;
+ my $key = "$cfg_section.$imap_section.$cfg_key";
+ my $v = $cfg->{lc($key)} //
+ $cfg->urlmatch("$cfg_section.$cfg_key", $url) // return;
$v =~ /\A[0-9]+(?:\.[0-9]+)?\z/s and return $v + 0;
if (ref($v) eq 'ARRAY') {
$v = join(', ', @$v);
my $sec = imap_section($uri);
for my $k (qw(Starttls Debug Compress)) {
my $key = lc("imap.$sec.$k");
- defined(my $orig = $cfg->{$key}) or next;
+ my $orig = $cfg->{$key} //
+ $cfg->urlmatch("imap.$k", $url) // next;
my $v = PublicInbox::Config::_git_config_bool($orig);
if (defined($v)) {
$mic_args->{$sec}->{$k} = $v;
warn "W: $key=$orig is not boolean\n";
}
}
- my $to = cfg_intvl($cfg, "imap.$sec.Timeout");
+ my $to = cfg_intvl($cfg, 'imap', 'Timeout', $sec, $url);
$mic_args->{$sec}->{Timeout} = $to if $to;
- $to = cfg_intvl($cfg, "imap.$sec.PollInterval");
+ $to = cfg_intvl($cfg, 'imap', 'PollInterval', $sec, $url);
$self->{imap_opt}->{$sec}->{poll_intvl} = $to if $to;
- $to = cfg_intvl($cfg, "imap.$sec.IdleInterval");
+ $to = cfg_intvl($cfg, 'imap', 'IdleInterval', $sec, $url);
$self->{imap_opt}->{$sec}->{idle_intvl} = $to if $to;
+
+ my $key = lc("imap.$sec.fetchBatchSize");
+ my $bs = $cfg->{lc($key)} //
+ $cfg->urlmatch('imap.fetchBatchSize', $url) // next;
+ if ($bs =~ /\A([0-9]+)\z/) {
+ $self->{imap_opt}->{$sec}->{batch_size} = $bs;
+ } else {
+ warn "W: $key=$bs is not an integer\n";
+ }
}
$mic_args;
}
$mic;
}
+sub imap_import_msg ($$$$) {
+ my ($self, $url, $uid, $raw) = @_;
+ # our target audience expects LF-only, save storage
+ $$raw =~ s/\r\n/\n/sg;
+
+ my $inboxes = $self->{imap}->{$url};
+ if (ref($inboxes)) {
+ for my $ibx (@$inboxes) {
+ my $eml = PublicInbox::Eml->new($$raw);
+ my $x = import_eml($self, $ibx, $eml);
+ }
+ } elsif ($inboxes eq 'watchspam') {
+ my $eml = PublicInbox::Eml->new($raw);
+ my $arg = [ $self, $eml, "$url UID:$uid" ];
+ $self->{config}->each_inbox(\&remove_eml_i, $arg);
+ } else {
+ die "BUG: destination unknown $inboxes";
+ }
+}
+
sub imap_fetch_all ($$$) {
my ($self, $mic, $uri) = @_;
my $sec = imap_section($uri);
return "E: $url cannot get UIDVALIDITY";
$r_uidnext //= $mic->uidnext($mbx) //
return "E: $url cannot get UIDNEXT";
- my $itrk = PublicInbox::IMAPTracker->new;
- my ($l_uidval, $l_uid) = $itrk->get_last($url);
+ my $itrk = PublicInbox::IMAPTracker->new($url);
+ my ($l_uidval, $l_uid) = $itrk->get_last;
$l_uidval //= $r_uidval; # first time
$l_uid //= 1;
if ($l_uidval != $r_uidval) {
}
return if $l_uid >= $r_uid; # nothing to do
+ warn "I: $url fetching UID $l_uid:$r_uid\n";
$mic->Uid(1); # the default, we hope
+ my $bs = $self->{imap_opt}->{$sec}->{batch_size} // 1;
my $req = $mic->imap4rev1 ? 'BODY.PEEK[]' : 'RFC822.PEEK';
+
+ # TODO: FLAGS may be useful for personal use
my $key = $req;
$key =~ s/\.PEEK//;
- my $inboxes = $self->{imap}->{$url};
- warn "I: $url fetching $l_uid..$r_uid\n";
- my $uid = -1;
+ my ($uids, $batch);
my $warn_cb = $SIG{__WARN__} || sub { print STDERR @_ };
local $SIG{__WARN__} = sub {
- $warn_cb->("$url UID:$uid\n");
+ $batch //= '?';
+ $warn_cb->("$url UID:$batch\n");
$warn_cb->(@_);
};
my $err;
- $itrk->{dbh}->begin_work;
- for my $u ($l_uid..$r_uid) {
- $uid = $u;
- local $0 = "UID:$uid $mbx $sec";
- my $r = $mic->fetch_hash($uid, $req);
- unless ($r) { # network error?
- $err = "E: $url UID FETCH $uid error: $!\n";
- last;
- }
-
- # messages get deleted, so holes appear
- defined(my $raw = delete $r->{$uid}->{$key}) or next;
-
- # our target audience expects LF-only, save storage
- $raw =~ s/\r\n/\n/sg;
-
- if (ref($inboxes)) {
- for my $ibx (@$inboxes) {
- my $eml = PublicInbox::Eml->new($raw);
- my $x = import_eml($self, $ibx, $eml);
+ do {
+ # I wish "UID FETCH $START:*" could work, but:
+ # 1) servers do not need to return results in any order
+ # 2) Mail::IMAPClient doesn't offer a streaming API
+ $uids = $mic->search("UID $l_uid:*") or
+ return "E: $url UID SEARCH $l_uid:* error: $!";
+ return if scalar(@$uids) == 0;
+
+ # RFC 3501 doesn't seem to indicate order of UID SEARCH
+ # responses, so sort it ourselves. Order matters so
+ # IMAPTracker can store the newest UID.
+ @$uids = sort { $a <=> $b } @$uids;
+
+ # Did we actually get new messages?
+ return if $uids->[0] < $l_uid;
+
+ $l_uid = $uids->[-1] + 1; # for next search
+ my $last_uid;
+
+ while (scalar @$uids) {
+ my @batch = splice(@$uids, 0, $bs);
+ $batch = join(',', @batch);
+ local $0 = "UID:$batch $mbx $sec";
+ my $r = $mic->fetch_hash($batch, $req);
+ unless ($r) { # network error?
+ $err = "E: $url UID FETCH $batch error: $!";
+ last;
}
- } elsif ($inboxes eq 'watchspam') {
- my $eml = PublicInbox::Eml->new($raw);
- my $arg = [ $self, $eml, "$uri UID:$uid" ];
- $self->{config}->each_inbox(\&remove_eml_i, $arg);
- } else {
- die "BUG: destination unknown $inboxes";
+ for my $uid (@batch) {
+ # messages get deleted, so holes appear
+ my $per_uid = delete $r->{$uid} // next;
+ my $raw = delete($per_uid->{$key}) // next;
+ imap_import_msg($self, $url, $uid, \$raw);
+ $last_uid = $uid;
+ last if $self->{quit};
+ }
+ last if $self->{quit};
}
- $itrk->update_last($url, $r_uidval, $uid);
- last if $self->{quit};
- }
- _done_for_now($self);
- $itrk->{dbh}->commit;
+ _done_for_now($self);
+ $itrk->update_last($r_uidval, $last_uid) if defined $last_uid;
+ } until ($err || $self->{quit});
$err;
}
my ($self) = @_;
delete $self->{idle_pids};
delete $self->{poll_pids};
+ delete $self->{opendirs};
PublicInbox::DS->Reset;
PublicInbox::Sigfd::sig_setmask($self->{oldset});
%SIG = (%SIG, %{$self->{sig}});
}
+sub watch_atfork_parent ($) {
+ my ($self) = @_;
+ _done_for_now($self);
+ $self->{mics} = {}; # going to be forking, so disconnect
+}
+
sub imap_idle_reap { # PublicInbox::DS::dwaitpid callback
my ($self, $pid) = @_;
my $uri_intvl = delete $self->{idle_pids}->{$pid} or
return if $self->{quit};
my $idle_todo = $self->{idle_todo};
if ($idle_todo && @$idle_todo) {
- $self->{mics} = {}; # going to be forking, so disconnect
+ watch_atfork_parent($self);
while (my $uri_intvl = shift(@$idle_todo)) {
imap_idle_fork($self, $uri_intvl);
}
}
}
-sub imap_fetch_fork ($$$) {
- my ($self, $intvl, $uris) = @_;
+sub imap_fetch_fork ($) { # DS::add_timer callback
+ my ($self, $intvl, $uris) = @{$_[0]};
return if $self->{quit};
- $self->{mics} = {}; # going to be forking, so disconnect
+ watch_atfork_parent($self);
defined(my $pid = fork) or die "fork: $!";
if ($pid == 0) {
watch_atfork_child($self);
PublicInbox::DS::dwaitpid($pid, \&imap_fetch_reap, $self);
}
-sub imap_fetch_cb ($$$) {
- my ($self, $intvl, $uris) = @_;
- sub { imap_fetch_fork($self, $intvl, $uris) };
-}
-
sub imap_fetch_reap { # PublicInbox::DS::dwaitpid callback
my ($self, $pid) = @_;
my $intvl_uris = delete $self->{poll_pids}->{$pid} or
map { $_->as_string."\n" } @$uris;
}
warn('I: will check ', $_->as_string, " in ${intvl}s\n") for @$uris;
- PublicInbox::DS::add_timer($intvl, imap_fetch_cb($self, $intvl, $uris));
+ PublicInbox::DS::add_timer($intvl, \&imap_fetch_fork,
+ [$self, $intvl, $uris]);
}
sub watch_imap_init ($) {
# poll all URIs for a given interval sequentially
while (my ($intvl, $uris) = each %$poll) {
- PublicInbox::DS::requeue(imap_fetch_cb($self, $intvl, $uris));
+ PublicInbox::DS::add_timer(0, \&imap_fetch_fork,
+ [$self, $intvl, $uris]);
}
}