]> Sergey Matveev's repositories - public-inbox.git/commitdiff
lei: start implementing inotify Maildir support
authorEric Wong <e@80x24.org>
Mon, 19 Jul 2021 08:59:35 +0000 (08:59 +0000)
committerEric Wong <e@80x24.org>
Thu, 22 Jul 2021 02:29:00 +0000 (02:29 +0000)
This allows lei to automatically note keyword (message flag)
changes made to a Maildir and propagate it into lei/store:

lei add-watch --state=tag-ro /path/to/Maildir

This doesn't persist across restarts, yet.  In the future,
it will be applied automatically to "lei q" output Maildirs
by default (with an option to disable it).

State values of tag-rw, index-<ro|rw>, import-<ro|rw> will all
be supported for Maildir.

This represents a fairly major internal change that's fairly
intrusive, but the whole daemon-oriented design was to
facilitate being able to automatically monitor (and propagate)
Maildir/IMAP flag changes.

MANIFEST
lib/PublicInbox/DirIdle.pm
lib/PublicInbox/LEI.pm
lib/PublicInbox/LeiAddWatch.pm [new file with mode: 0644]
lib/PublicInbox/LeiInput.pm
lib/PublicInbox/LeiLsWatch.pm [new file with mode: 0644]
lib/PublicInbox/LeiNoteEvent.pm [new file with mode: 0644]
lib/PublicInbox/LeiSearch.pm
lib/PublicInbox/LeiSelfSocket.pm [new file with mode: 0644]
lib/PublicInbox/LeiWatch.pm [new file with mode: 0644]
t/lei-watch.t [new file with mode: 0644]

index 146a32abed239653ba7bf0613a8fa493ae565e9a..1d79b7c9e6dea566facbe0f69fe2af6a8987f2e7 100644 (file)
--- a/MANIFEST
+++ b/MANIFEST
@@ -198,6 +198,7 @@ lib/PublicInbox/Isearch.pm
 lib/PublicInbox/KQNotify.pm
 lib/PublicInbox/LEI.pm
 lib/PublicInbox/LeiALE.pm
+lib/PublicInbox/LeiAddWatch.pm
 lib/PublicInbox/LeiAuth.pm
 lib/PublicInbox/LeiBlob.pm
 lib/PublicInbox/LeiConvert.pm
@@ -220,8 +221,10 @@ lib/PublicInbox/LeiLsLabel.pm
 lib/PublicInbox/LeiLsMailSource.pm
 lib/PublicInbox/LeiLsMailSync.pm
 lib/PublicInbox/LeiLsSearch.pm
+lib/PublicInbox/LeiLsWatch.pm
 lib/PublicInbox/LeiMailSync.pm
 lib/PublicInbox/LeiMirror.pm
+lib/PublicInbox/LeiNoteEvent.pm
 lib/PublicInbox/LeiOverview.pm
 lib/PublicInbox/LeiP2q.pm
 lib/PublicInbox/LeiPmdir.pm
@@ -232,6 +235,7 @@ lib/PublicInbox/LeiRemote.pm
 lib/PublicInbox/LeiRm.pm
 lib/PublicInbox/LeiSavedSearch.pm
 lib/PublicInbox/LeiSearch.pm
+lib/PublicInbox/LeiSelfSocket.pm
 lib/PublicInbox/LeiStore.pm
 lib/PublicInbox/LeiStoreErr.pm
 lib/PublicInbox/LeiSucks.pm
@@ -239,6 +243,7 @@ lib/PublicInbox/LeiTag.pm
 lib/PublicInbox/LeiToMail.pm
 lib/PublicInbox/LeiUp.pm
 lib/PublicInbox/LeiViewText.pm
+lib/PublicInbox/LeiWatch.pm
 lib/PublicInbox/LeiXSearch.pm
 lib/PublicInbox/Linkify.pm
 lib/PublicInbox/Listener.pm
@@ -434,6 +439,7 @@ t/lei-q-save.t
 t/lei-q-thread.t
 t/lei-sigpipe.t
 t/lei-tag.t
+t/lei-watch.t
 t/lei.t
 t/lei_dedupe.t
 t/lei_external.t
index 5142d005e4aa326543fe8a99dca0c4e491eb42c5..7031e5fd5615f17aa1df4467c6aebdcf36d7c31a 100644 (file)
@@ -53,6 +53,23 @@ sub new {
        $self;
 }
 
+sub add_watches {
+       my ($self, $dirs, $gone) = @_;
+       my $fl = $MAIL_IN | ($gone ? $MAIL_GONE : 0);
+       for my $d (@$dirs) {
+               $self->{inot}->watch($d, $fl);
+       }
+       PublicInbox::FakeInotify::poll_once($self) if !$ino_cls;
+}
+
+sub rm_watches {
+       my ($self, $dir) = @_;
+       my $inot = $self->{inot};
+       if (my $cb = $inot->can('rm_watches')) { # TODO for fake watchers
+               $cb->($inot, $dir);
+       }
+}
+
 sub event_step {
        my ($self) = @_;
        my $cb = $self->{cb};
index a9f5edaebe4efad8cac852874a7adaba31208c66..b92d75125df4c9578c5475083a0c79e348b8d3cb 100644 (file)
@@ -28,14 +28,15 @@ use Time::HiRes qw(stat); # ctime comparisons for config cache
 use File::Path qw(mkpath);
 use File::Spec;
 our $quit = \&CORE::exit;
-our ($current_lei, $errors_log, $listener, $oldset, $dir_idle);
-my ($recv_cmd, $send_cmd);
+our ($current_lei, $errors_log, $listener, $oldset, $dir_idle,
+       $recv_cmd, $send_cmd);
 my $GLP = Getopt::Long::Parser->new;
 $GLP->configure(qw(gnu_getopt no_ignore_case auto_abbrev));
 my $GLP_PASS = Getopt::Long::Parser->new;
 $GLP_PASS->configure(qw(gnu_getopt no_ignore_case auto_abbrev pass_through));
 
 our %PATH2CFG; # persistent for socket daemon
+our $MDIR2CFGPATH; # /path/to/maildir => { /path/to/config => undef }
 
 # TBD: this is a documentation mechanism to show a subcommand
 # (may) pass options through to another command:
@@ -232,10 +233,9 @@ our %CMD = ( # sorted in order of importance/use:
        qw(exact! all jobs:i indexed), @c_opt ],
 
 'add-watch' => [ 'LOCATION', 'watch for new messages and flag changes',
-       qw(import! kw! interval=s recursive|r
-       exclude=s include=s), @c_opt ],
+       qw(poll-interval=s state=s recursive|r), @c_opt ],
 'ls-watch' => [ '[FILTER...]', 'list active watches with numbers and status',
-               qw(format|f=s z), @c_opt ],
+               qw(l z|0), @c_opt ],
 'pause-watch' => [ '[WATCH_NUMBER_OR_FILTER]', qw(all local remote), @c_opt ],
 'resume-watch' => [ '[WATCH_NUMBER_OR_FILTER]', qw(all local remote), @c_opt ],
 'forget-watch' => [ '{WATCH_NUMBER|--prune}', 'stop and forget a watch',
@@ -391,6 +391,7 @@ my %OPTDESC = (
 'format|f=s    ls-search' => ['OUT|json|jsonl|concatjson',
                        'listing output format' ],
 'l     ls-search' => 'long listing format',
+'l     ls-watch' => 'long listing format',
 'l     ls-mail-source' => 'long listing format',
 'url   ls-mail-source' => 'show full URL of newsgroup or IMAP folder',
 'format|f=s    ls-external' => $ls_format,
@@ -435,7 +436,7 @@ my %CONFIG_KEYS = (
        'leistore.dir' => 'top-level storage location',
 );
 
-my @WQ_KEYS = qw(lxs l2m ikw pmd wq1); # internal workers
+my @WQ_KEYS = qw(lxs l2m ikw pmd wq1 lne); # internal workers
 
 sub _drop_wq {
        my ($self) = @_;
@@ -538,7 +539,7 @@ sub _lei_atfork_child {
                chdir '/' or die "chdir(/): $!";
                close($_) for (grep(defined, delete @$self{qw(0 1 2 sock)}));
                if (my $cfg = $self->{cfg}) {
-                       delete $cfg->{-lei_store};
+                       delete @$cfg{qw(-lei_store -watches -lei_note_event)};
                }
        } else { # worker, Net::NNTP (Net::Cmd) uses STDERR directly
                open STDERR, '+>&='.fileno($self->{2}) or warn "open $!";
@@ -555,6 +556,8 @@ sub _lei_atfork_child {
        undef $listener;
        undef $dir_idle;
        %PATH2CFG = ();
+       $MDIR2CFGPATH = {};
+       eval 'no warnings; undef $PublicInbox::LeiNoteEvent::to_flush';
        undef $errors_log;
        $quit = \&CORE::exit;
        $self->{-eml_noisy} or # only "lei import" sets this atm
@@ -781,10 +784,12 @@ sub _lei_cfg ($;$) {
        my $f = _config_path($self);
        my @st = stat($f);
        my $cur_st = @st ? pack('dd', $st[10], $st[7]) : ''; # 10:ctime, 7:size
-       my ($sto, $sto_dir);
+       my ($sto, $sto_dir, $watches, $lne);
        if (my $cfg = $PATH2CFG{$f}) { # reuse existing object in common case
                return ($self->{cfg} = $cfg) if $cur_st eq $cfg->{-st};
-               ($sto, $sto_dir) = @$cfg{qw(-lei_store leistore.dir)};
+               ($sto, $sto_dir, $watches, $lne) =
+                               @$cfg{qw(-lei_store leistore.dir -watches
+                                       -lei_note_event)};
        }
        if (!@st) {
                unless ($creat) {
@@ -805,6 +810,8 @@ sub _lei_cfg ($;$) {
                        eq File::Spec->canonpath($cfg->{'leistore.dir'} //
                                                store_path($self))) {
                $cfg->{-lei_store} = $sto;
+               $cfg->{-lei_note_event} = $lne;
+               $cfg->{-watches} = $watches if $watches;
        }
        if (scalar(keys %PATH2CFG) > 5) {
                # FIXME: use inotify/EVFILT_VNODE to detect unlinked configs
@@ -817,7 +824,7 @@ sub _lei_cfg ($;$) {
 
 sub _lei_store ($;$) {
        my ($self, $creat) = @_;
-       my $cfg = _lei_cfg($self, $creat);
+       my $cfg = _lei_cfg($self, $creat) // return;
        $cfg->{-lei_store} //= do {
                require PublicInbox::LeiStore;
                my $dir = $cfg->{'leistore.dir'} // store_path($self);
@@ -1126,6 +1133,53 @@ sub dump_and_clear_log {
        }
 }
 
+sub cfg2lei ($) {
+       my ($cfg) = @_;
+       my $lei = bless { env => { %{$cfg->{-env}} } }, __PACKAGE__;
+       open($lei->{0}, '<&', \*STDIN) or die "dup 0: $!";
+       open($lei->{1}, '>>&', \*STDOUT) or die "dup 1: $!";
+       open($lei->{2}, '>>&', \*STDERR) or die "dup 2: $!";
+       open($lei->{3}, '/') or die "open /: $!";
+       chdir($lei->{3}) or die "chdir /': $!";
+       my ($x, $y);
+       socketpair($x, $y, AF_UNIX, SOCK_SEQPACKET, 0) or die "socketpair: $!";
+       $lei->{sock} = $x;
+       require PublicInbox::LeiSelfSocket;
+       PublicInbox::LeiSelfSocket->new($y); # adds to event loop
+       $lei;
+}
+
+sub dir_idle_handler ($) { # PublicInbox::DirIdle callback
+       my ($ev) = @_; # Linux::Inotify2::Event or duck type
+       my $fn = $ev->fullname;
+       if ($fn =~ m!\A(.+)/(new|cur)/([^/]+)\z!) { # Maildir file
+               my ($mdir, $nc, $bn) = ($1, $2, $3);
+               $nc = '' if $ev->IN_DELETE;
+               for my $f (keys %{$MDIR2CFGPATH->{$mdir} // {}}) {
+                       my $cfg = $PATH2CFG{$f} // next;
+                       eval {
+                               local %ENV = %{$cfg->{-env}};
+                               my $lei = cfg2lei($cfg);
+                               $lei->dispatch('note-event',
+                                               "maildir:$mdir", $nc, $bn, $fn);
+                       };
+                       warn "E note-event $f: $@\n" if $@;
+               }
+       }
+       if ($ev->can('cancel') && ($ev->IN_IGNORE || $ev->IN_UNMOUNT)) {
+               $ev->cancel;
+       }
+       if ($fn =~ m!\A(.+)/(?:new|cur)\z! && !-e $fn) {
+               delete $MDIR2CFGPATH->{$1};
+       }
+       if (!-e $fn) { # config file or Maildir gone
+               for my $cfgpaths (values %$MDIR2CFGPATH) {
+                       delete $cfgpaths->{$fn};
+               }
+               delete $PATH2CFG{$fn};
+       }
+}
+
 # lei(1) calls this when it can't connect
 sub lazy_start {
        my ($path, $errno, $narg) = @_;
@@ -1175,6 +1229,7 @@ sub lazy_start {
        return if $pid;
        $0 = "lei-daemon $path";
        local %PATH2CFG;
+       local $MDIR2CFGPATH;
        $listener->blocking(0);
        my $exit_code;
        my $pil = PublicInbox::Listener->new($listener, \&accept_dispatch);
@@ -1204,8 +1259,8 @@ sub lazy_start {
        local $SIG{PIPE} = 'IGNORE';
        require PublicInbox::DirIdle;
        local $dir_idle = PublicInbox::DirIdle->new([$sock_dir], sub {
-               # just rely on wakeup ot hit PostLoopCallback set below
-               _dir_idle_handler(@_) if $_[0]->fullname ne $path;
+               # just rely on wakeup to hit PostLoopCallback set below
+               dir_idle_handler($_[0]) if $_[0]->fullname ne $path;
        }, 1);
        if ($sigfd) {
                undef $sigfd; # unref, already in DS::DescriptorMap
@@ -1293,4 +1348,41 @@ sub wq_eof { # EOF callback for main daemon
        $wq1->wq_wait_old(\&wq_done_wait, $lei);
 }
 
+sub watch_state_ok ($) {
+       my ($state) = $_[-1]; # $_[0] may be $self
+       $state =~ /\Apause|(?:import|index|tag)-(?:ro|rw)\z/;
+}
+
+sub refresh_watches {
+       my ($lei) = @_;
+       my $cfg = _lei_cfg($lei) or return;
+       $cfg->{-env} //= { %{$lei->{env}}, PWD => '/' }; # for cfg2lei
+       my $watches = $cfg->{-watches} //= {};
+       require PublicInbox::LeiWatch;
+       for my $w (grep(/\Awatch\..+\.state\z/, keys %$cfg)) {
+               my $url = substr($w, length('watch.'), -length('.state'));
+               my $lw = $watches->{$w} //= PublicInbox::LeiWatch->new($url);
+               my $state = $cfg->get_1("watch.$url", 'state');
+               if (!watch_state_ok($state)) {
+                       $lei->err("watch.$url.state=$state not supported");
+                       next;
+               }
+               my $f = $cfg->{'-f'};
+               if ($url =~ /\Amaildir:(.+)/i) {
+                       my $d = File::Spec->canonpath($1);
+                       if ($state eq 'pause') {
+                               delete $MDIR2CFGPATH->{$d}->{$f};
+                               scalar(keys %{$MDIR2CFGPATH->{$d}}) or
+                                       delete $MDIR2CFGPATH->{$d};
+                       } elsif (!exists($MDIR2CFGPATH->{$d}->{$f})) {
+                               $dir_idle->add_watches(["$d/cur", "$d/new"], 1);
+                               $MDIR2CFGPATH->{$d}->{$f} = undef;
+                       }
+               } else { # TODO: imap/nntp/jmap
+                       $lei->child_error(1,
+                               "E: watch $url not supported, yet");
+               }
+       }
+}
+
 1;
diff --git a/lib/PublicInbox/LeiAddWatch.pm b/lib/PublicInbox/LeiAddWatch.pm
new file mode 100644 (file)
index 0000000..671d54f
--- /dev/null
@@ -0,0 +1,41 @@
+# Copyright all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# "lei add-watch" command
+package PublicInbox::LeiAddWatch;
+use strict;
+use v5.10.1;
+use parent qw(PublicInbox::LeiInput);
+
+sub lei_add_watch {
+       my ($lei, @argv) = @_;
+       my $cfg = $lei->_lei_cfg(1);
+       my $self = bless {}, __PACKAGE__;
+       $lei->{opt}->{'mail-sync'} = 1; # for prepare_inputs
+       my $state = $lei->{opt}->{'state'} // 'import-rw';
+       $lei->watch_state_ok($state) or
+               return $lei->fail("invalid state: $state");
+       my $vmd_mod = $self->vmd_mod_extract(\@argv);
+       return $lei->fail(join("\n", @{$vmd_mod->{err}})) if $vmd_mod->{err};
+       $self->prepare_inputs($lei, \@argv) or return;
+       my @vmd;
+       while (my ($type, $vals) = each %$vmd_mod) {
+               push @vmd, "$type:$_" for @$vals;
+       }
+       my $vmd0 = shift @vmd;
+       for my $w (@{$self->{inputs}}) {
+               # clobber existing, allow multiple
+               if (defined($vmd0)) {
+                       $lei->_config("watch.$w.vmd", '--replace-all', $vmd0);
+                       for my $v (@vmd) {
+                               $lei->_config("watch.$w.vmd", $v);
+                       }
+               }
+               next if defined $cfg->{"watch.$w.state"};
+               $lei->_config("watch.$w.state", $state);
+       }
+       delete $lei->{cfg}; # force reload
+       $lei->refresh_watches;
+}
+
+1;
index de2a8ff1c31944639031f862b2a44c1dc93e925f..fa330df5fbdb86887faefd686c690a56925de616 100644 (file)
@@ -372,6 +372,7 @@ sub input_only_net_merge_all_done {
 
 # like Getopt::Long, but for +kw:FOO and -kw:FOO to prepare
 # for update_xvmd -> update_vmd
+# returns something like { "+L" => [ @Labels ], ... }
 sub vmd_mod_extract {
        my $argv = $_[-1];
        my $vmd_mod = {};
diff --git a/lib/PublicInbox/LeiLsWatch.pm b/lib/PublicInbox/LeiLsWatch.pm
new file mode 100644 (file)
index 0000000..f96dc4e
--- /dev/null
@@ -0,0 +1,15 @@
+# Copyright all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+package PublicInbox::LeiLsWatch;
+use strict;
+use v5.10.1;
+
+sub lei_ls_watch {
+       my ($lei) = @_;
+       my $cfg = $lei->_lei_cfg or return;
+       my @w = (join("\n", keys %$cfg) =~ m/^watch\.(.+?)\.state$/sgm);
+       $lei->puts(join("\n", @w)) if @w;
+}
+
+1;
diff --git a/lib/PublicInbox/LeiNoteEvent.pm b/lib/PublicInbox/LeiNoteEvent.pm
new file mode 100644 (file)
index 0000000..bf15cd2
--- /dev/null
@@ -0,0 +1,124 @@
+# Copyright (C) all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# internal command for dealing with inotify, kqueue vnodes, etc
+package PublicInbox::LeiNoteEvent;
+use strict;
+use v5.10.1;
+use parent qw(PublicInbox::IPC);
+
+my $flush_timer;
+our $to_flush; # { cfgpath => $lei }
+
+sub flush_lei ($) {
+       my ($lei) = @_;
+       if (my $lne = delete $lei->{cfg}->{-lei_note_event}) {
+               $lne->wq_close(1, undef, $lei); # runs _lei_wq_eof;
+       } else { # lms_clear_src calls only:
+               my $wait = $lei->{sto}->ipc_do('done');
+       }
+}
+
+# we batch up writes and flush every 5s (matching Linux default
+# writeback behavior) since MUAs can trigger a storm of inotify events
+sub flush_task { # PublicInbox::DS timer callback
+       undef $flush_timer;
+       my $todo = $to_flush // return;
+       $to_flush = undef;
+       for my $lei (values %$todo) { flush_lei($lei) }
+}
+
+# sets a timer to flush
+sub note_event_arm_done ($) {
+       my ($lei) = @_;
+       $flush_timer //= PublicInbox::DS::add_timer(5, \&flush_task);
+       $to_flush->{$lei->{cfg}->{'-f'}} //= $lei;
+}
+
+sub eml_event ($$$$) {
+       my ($self, $eml, $kw, $state) = @_;
+       my $sto = $self->{lei}->{sto};
+       my $lse = $self->{lse} //= $sto->search;
+       my $vmd = { kw => $kw };
+       if ($state =~ /\Aimport-(?:rw|ro)\z/) {
+               $sto->ipc_do('set_eml', $eml, $vmd);
+       } elsif ($state =~ /\Aindex-(?:rw|ro)\z/) {
+               my $xoids = $self->{lei}->ale->xoids_for($eml);
+               $sto->ipc_do('index_eml_only', $eml, $vmd, $xoids);
+       } elsif ($state =~ /\Atag-(?:rw|ro)\z/) {
+               my $c = $lse->kw_changed($eml, $kw, my $docids = []);
+               if (scalar @$docids) { # already in lei/store
+                       $sto->ipc_do('set_eml_vmd', undef, $vmd, $docids) if $c;
+               } elsif (my $xoids = $self->{lei}->ale->xoids_for($eml)) {
+                       # it's in an external, only set kw, here
+                       $sto->ipc_do('set_xvmd', $xoids, $eml, $vmd);
+               } # else { totally unknown
+       } else {
+               warn "unknown state: $state (in $self->{lei}->{cfg}->{'-f'})\n";
+       }
+}
+
+sub maildir_event { # via wq_io_do
+       my ($self, $fn, $kw, $state) = @_;
+       my $eml = PublicInbox::InboxWritable::eml_from_path($fn) // return;
+       eml_event($self, $eml, $kw, $state);
+}
+
+sub lei_note_event {
+       my ($lei, $folder, $new_cur, $bn, $fn, @rest) = @_;
+       die "BUG: unexpected: @rest" if @rest;
+       my $cfg = $lei->_lei_cfg or return; # gone (race)
+       my $sto = $lei->_lei_store or return; # gone
+       return flush_lei($lei) if $folder eq 'done'; # special case
+       my $lms = $sto->search->lms or return;
+       my $err = $lms->arg2folder($lei, [ $folder ]);
+       return if $err->{fail};
+       undef $lms;
+       my $state = $cfg->get_1("watch.$folder", 'state') // 'pause';
+       return if $state eq 'pause';
+       $lei->ale; # prepare
+       $sto->write_prepare($lei);
+       if ($new_cur eq '') {
+               $sto->ipc_do('lms_clear_src', $folder, \$bn);
+               return note_event_arm_done($lei);
+       }
+       require PublicInbox::MdirReader;
+       my $self = $cfg->{-lei_note_event} //= do {
+               my $wq = bless {}, __PACKAGE__;
+               # MUAs such as mutt can trigger massive rename() storms so
+               # use all CPU power available:
+               my $jobs = $wq->detect_nproc // 1;
+               my ($op_c, $ops) = $lei->workers_start($wq, $jobs);
+               $lei->wait_wq_events($op_c, $ops);
+               note_event_arm_done($lei);
+               $lei->{lne} = $wq;
+       };
+       if ($folder =~ /\Amaildir:/i) {
+               my $fl = PublicInbox::MdirReader::maildir_basename_flags($bn)
+                       // return;
+               return if index($fl, 'T') >= 0;
+               my $kw = PublicInbox::MdirReader::flags2kw($fl);
+               $self->wq_io_do('maildir_event', [], $fn, $kw, $state);
+       } # else: TODO: imap
+}
+
+sub ipc_atfork_child {
+       my ($self) = @_;
+       $self->{lei}->_lei_atfork_child(1); # persistent, for a while
+       $self->SUPER::ipc_atfork_child;
+}
+
+sub lne_done_wait {
+       my ($arg, $pid) = @_;
+       my ($self, $lei) = @$arg;
+       $lei->can('wq_done_wait')->($arg, $pid);
+}
+
+sub _lei_wq_eof { # EOF callback for main lei daemon
+       my ($lei) = @_;
+       my $lne = delete $lei->{lne} or return $lei->fail;
+       my $wait = $lei->{sto}->ipc_do('done');
+       $lne->wq_wait_old(\&lne_done_wait, $lei);
+}
+
+1;
index 06ea62995a2b089f1bc10922fa40a8d1be640fbc..37bfc65e27cc7515b0b7d285b84e81120dbd3d5f 100644 (file)
@@ -5,7 +5,7 @@
 package PublicInbox::LeiSearch;
 use strict;
 use v5.10.1;
-use parent qw(PublicInbox::ExtSearch);
+use parent qw(PublicInbox::ExtSearch); # PublicInbox::Search->reopen
 use PublicInbox::Search qw(xap_terms);
 use PublicInbox::ContentHash qw(content_digest content_hash);
 use PublicInbox::MID qw(mids mids_for_index);
diff --git a/lib/PublicInbox/LeiSelfSocket.pm b/lib/PublicInbox/LeiSelfSocket.pm
new file mode 100644 (file)
index 0000000..3d84764
--- /dev/null
@@ -0,0 +1,45 @@
+# Copyright all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# dummy placeholder socket for internal lei commands.
+# This receives what script/lei receives, but isn't connected
+# to an interactive terminal so I'm not sure what to do with it...
+package PublicInbox::LeiSelfSocket;
+use strict;
+use v5.10.1;
+use parent qw(PublicInbox::DS);
+use Data::Dumper;
+$Data::Dumper::Useqq = 1; # should've been the Perl default :P
+use PublicInbox::Syscall qw(EPOLLIN EPOLLET);
+use PublicInbox::Spawn;
+my $recv_cmd;
+
+sub new {
+       my ($cls, $r) = @_;
+       my $self = bless { sock => $r }, $cls;
+       $r->blocking(0);
+       no warnings 'once';
+       $recv_cmd = $PublicInbox::LEI::recv_cmd;
+       $self->SUPER::new($r, EPOLLIN|EPOLLET);
+}
+
+sub event_step {
+       my ($self) = @_;
+       while (1) {
+               my (@fds) = $recv_cmd->($self->{sock}, my $buf, 4096 * 33);
+               if (scalar(@fds) == 1 && !defined($fds[0])) {
+                       return if $!{EAGAIN};
+                       next if $!{EINTR};
+                       die "recvmsg: $!";
+               }
+               # open so perl can auto-close them:
+               for my $fd (@fds) {
+                       open(my $newfh, '+<&=', $fd) or die "open +<&=$fd: $!";
+               }
+               return $self->close if $buf eq '';
+               warn Dumper({ 'unexpected self msg' => $buf, fds => \@fds });
+               # TODO: figure out what to do with these messages...
+       }
+}
+
+1;
diff --git a/lib/PublicInbox/LeiWatch.pm b/lib/PublicInbox/LeiWatch.pm
new file mode 100644 (file)
index 0000000..35267b5
--- /dev/null
@@ -0,0 +1,13 @@
+# Copyright all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# represents a Maildir or IMAP "watch" item
+package PublicInbox::LeiWatch;
+use strict;
+use v5.10.1;
+use parent qw(PublicInbox::IPC);
+
+# "url" may be something like "maildir:/path/to/dir"
+sub new { bless { url => $_[1] }, $_[0] }
+
+1;
diff --git a/t/lei-watch.t b/t/lei-watch.t
new file mode 100644 (file)
index 0000000..3a2f9e6
--- /dev/null
@@ -0,0 +1,49 @@
+#!perl -w
+# Copyright all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+use strict; use v5.10.1; use PublicInbox::TestCommon;
+use File::Path qw(make_path);
+require_mods('lei');
+my $have_fast_inotify = eval { require Linux::Inotify2 } ||
+       eval { require IO::KQueue };
+
+$have_fast_inotify or
+       diag("$0 IO::KQueue or Linux::Inotify2 missing, test will be slow");
+
+my ($ro_home, $cfg_path) = setup_public_inboxes;
+test_lei(sub {
+       my $md = "$ENV{HOME}/md";
+       my $md2 = $md.'2';
+       lei_ok 'ls-watch';
+       is($lei_out, '', 'nothing in ls-watch, yet');
+       if (0) { # TODO
+               my $url = 'imaps://example.com/foo.bar.0';
+               lei_ok([qw(add-watch --state=pause), $url], undef, {});
+               lei_ok 'ls-watch';
+               is($lei_out, "$url\n", 'ls-watch shows added watch');
+               ok(!lei(qw(add-watch --state=pause), 'bogus'.$url),
+                       'bogus URL rejected');
+       }
+
+       # first, make sure tag-ro works
+       make_path("$md/new", "$md/cur", "$md/tmp");
+       lei_ok qw(add-watch --state=tag-ro), $md;
+       lei_ok 'ls-watch';
+       like($lei_out, qr/^\Qmaildir:$md\E$/sm, 'maildir shown');
+       lei_ok qw(q mid:testmessage@example.com -o), $md, '-I', "$ro_home/t1";
+       my @f = glob("$md/cur/*:2,");
+       is(scalar(@f), 1, 'got populated maildir with one result');
+       rename($f[0], "$f[0]S") or xbail "rename $!"; # set (S)een
+       $have_fast_inotify or tick(2);
+       lei_ok qw(note-event done); # flushes immediately (instead of 5s)
+
+       lei_ok qw(q mid:testmessage@example.com -o), $md2, '-I', "$ro_home/t1";
+       my @f2 = glob("$md2/*/*");
+       is(scalar(@f2), 1, 'got one result');
+       like($f2[0], qr/S\z/, 'seen set from rename');
+       my $e2 = eml_load($f2[0]);
+       my $e1 = eml_load("$f[0]S");
+       is_deeply($e2, $e1, 'results match');
+});
+
+done_testing;