sub new {
my ($class, $config) = @_;
- my (%mdmap, @mdir, $spamc);
- my %uniq; # directory => count
+ my (%mdmap, $spamc);
my %imap; # url => [inbox objects] or 'watchspam'
# "publicinboxwatch" is the documented namespace
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.
- my $cur = "$dir/cur";
- push @mdir, $cur;
- $uniq{$cur}++;
- $mdmap{$cur} = 'watchspam';
+ $mdmap{"$dir/cur"} = 'watchspam';
} elsif (my $url = imap_url($dir)) {
$imap{$url} = 'watchspam';
} else {
# 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 @{$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);
my $mdre;
- if (@mdir) {
- $mdre = join('|', map { quotemeta($_) } @mdir);
+ if (scalar keys %mdmap) {
+ $mdre = join('|', map { quotemeta($_) } keys %mdmap);
$mdre = qr!\A($mdre)/!;
}
+ return unless $mdre || scalar(keys %imap);
bless {
spamcheck => $spamcheck,
mdmap => \%mdmap,
- mdir => \@mdir,
mdre => $mdre,
config => $config,
imap => scalar keys %imap ? \%imap : undef,
if (my $imap_pid = $self->{-imap_pid}) {
kill('QUIT', $imap_pid);
}
- if (my $idle_pids = $self->{idle_pids}) {
- kill('QUIT', $_) for (keys %$idle_pids);
+ for (qw(idle_pids poll_pids)) {
+ my $pids = $self->{$_} or next;
+ kill('QUIT', $_) for (keys %$pids);
}
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 $@;
+ }
}
}
$self->{done_timer} //= PublicInbox::DS::requeue($done);
};
require PublicInbox::DirIdle;
- PublicInbox::DirIdle->new($self->{mdir}, $cb); # EPOLL_CTL_ADD
+ # inotify_create + EPOLL_CTL_ADD
+ PublicInbox::DirIdle->new([keys %{$self->{mdmap}}], $cb);
}
# returns the git config section name, e.g [imap "imaps://user@example.com"]
$uri->scheme . '://' . $uri->authority;
}
-sub cfg_intvl ($$) {
- my ($cfg, $key) = @_;
- defined(my $v = $cfg->{lc($key)}) or return;
- $v =~ /\A[0-9]+\z/s and return $v + 0;
+sub cfg_intvl ($$$) {
+ my ($cfg, $key, $url) = @_;
+ my $v = $cfg->urlmatch($key, $url) // return;
+ $v =~ /\A[0-9]+(?:\.[0-9]+)?\z/s and return $v + 0;
if (ref($v) eq 'ARRAY') {
$v = join(', ', @$v);
warn "W: $key has multiple values: $v\nW: $key ignored\n";
} else {
- warn "W: $key=$v is not an integer value in seconds\n";
+ warn "W: $key=$v is not a numeric value in seconds\n";
}
}
for my $url (sort keys %{$self->{imap}}) {
my $uri = PublicInbox::URIimap->new($url);
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;
+ for my $f (qw(Starttls Debug Compress)) {
+ my $k = "imap.$f";
+ my $orig = $cfg->urlmatch($k, $url) // next;
my $v = PublicInbox::Config::_git_config_bool($orig);
if (defined($v)) {
- $mic_args->{$sec}->{$k} = $v;
+ $mic_args->{$sec}->{$f} = $v;
} else {
- warn "W: $key=$orig is not boolean\n";
+ warn "W: $k=$orig for $url is not boolean\n";
}
}
- my $to = cfg_intvl($cfg, "imap.$sec.Timeout");
+ my $to = cfg_intvl($cfg, 'imap.timeout', $url);
$mic_args->{$sec}->{Timeout} = $to if $to;
- $to = cfg_intvl($cfg, "imap.$sec.PollInterval");
+ $to = cfg_intvl($cfg, 'imap.pollInterval', $url);
$self->{imap_opt}->{$sec}->{poll_intvl} = $to if $to;
- $to = cfg_intvl($cfg, "imap.$sec.IdleInterval");
+ $to = cfg_intvl($cfg, 'imap.IdleInterval', $url);
$self->{imap_opt}->{$sec}->{idle_intvl} = $to if $to;
+
+ my $k = 'imap.fetchBatchSize';
+ my $bs = $cfg->urlmatch($k, $url) // next;
+ if ($bs =~ /\A([0-9]+)\z/) {
+ $self->{imap_opt}->{$sec}->{batch_size} = $bs;
+ } else {
+ warn "$k=$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;
}
sub watch_atfork_child ($) {
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);
}
goto(&fs_scan_step) if $self->{mdre};
}
+sub watch_imap_fetch_all ($$) {
+ my ($self, $uris) = @_;
+ for my $uri (@$uris) {
+ my $sec = imap_section($uri);
+ my $mic_arg = $self->{mic_arg}->{$sec} or
+ die "BUG: no Mail::IMAPClient->new arg for $sec";
+ my $mic = PublicInbox::IMAPClient->new(%$mic_arg) or next;
+ my $err = imap_fetch_all($self, $mic, $uri);
+ last if $self->{quit};
+ warn $err, "\n" if $err;
+ }
+}
+
+sub imap_fetch_fork ($) { # DS::add_timer callback
+ my ($self, $intvl, $uris) = @{$_[0]};
+ return if $self->{quit};
+ watch_atfork_parent($self);
+ defined(my $pid = fork) or die "fork: $!";
+ if ($pid == 0) {
+ watch_atfork_child($self);
+ watch_imap_fetch_all($self, $uris);
+ _exit(0);
+ }
+ $self->{poll_pids}->{$pid} = [ $intvl, $uris ];
+ PublicInbox::DS::dwaitpid($pid, \&imap_fetch_reap, $self);
+}
+
+sub imap_fetch_reap { # PublicInbox::DS::dwaitpid callback
+ my ($self, $pid) = @_;
+ my $intvl_uris = delete $self->{poll_pids}->{$pid} or
+ die "BUG: PID=$pid (unknown) reaped: \$?=$?\n";
+ return if $self->{quit};
+ my ($intvl, $uris) = @$intvl_uris;
+ if ($?) {
+ warn "W: PID=$pid died: \$?=$?\n",
+ map { $_->as_string."\n" } @$uris;
+ }
+ warn('I: will check ', $_->as_string, " in ${intvl}s\n") for @$uris;
+ PublicInbox::DS::add_timer($intvl, \&imap_fetch_fork,
+ [$self, $intvl, $uris]);
+}
+
sub watch_imap_init ($) {
my ($self) = @_;
eval { require PublicInbox::IMAPClient } or
$self->{idle_todo} = $idle;
PublicInbox::DS::requeue($self); # ->event_step to fork
}
- # TODO: polling
+ return unless scalar keys %$poll;
+ $self->{poll_pids} = {};
+
+ # poll all URIs for a given interval sequentially
+ while (my ($intvl, $uris) = each %$poll) {
+ PublicInbox::DS::add_timer(0, \&imap_fetch_fork,
+ [$self, $intvl, $uris]);
+ }
}
sub watch {
$opendirs->{$dir} = $dh if $n < 0;
}
if ($op && $op eq 'full') {
- foreach my $dir (@{$self->{mdir}}) {
+ foreach my $dir (keys %{$self->{mdmap}}) {
next if $opendirs->{$dir}; # already in progress
my $ok = opendir(my $dh, $dir);
unless ($ok) {