--- AnyEvent-MP/MP/LogCatcher.pm	2009/09/06 00:13:21	1.1
+++ AnyEvent-MP/MP/LogCatcher.pm	2012/03/22 22:57:50	1.7
@@ -8,11 +8,12 @@
 
 =head1 DESCRIPTION
 
-This relatively simple module overrides C<$AnyEvent::MP::Kernel::WARN> on
-every node and sends all log messages to the node running this service.
+This relatively simple module attaches itself to the
+C<$AnyEvent::Log::COLLECT> context on every node and sends all log
+messages to the node showing interest via the C<catch> function.
 
-No attempt to buffer log messages on connection loss, or retransmit lost
-messages, is done.
+No attempt to buffer log messages on connection loss, or to retransmit
+lost messages, is done.
 
 =head1 GLOBALS AND FUNCTIONS
 
@@ -27,6 +28,7 @@
 use POSIX ();
 
 use AnyEvent ();
+use AnyEvent::Log ();
 use AnyEvent::Util ();
 
 use AnyEvent::MP;
@@ -34,72 +36,118 @@
 
 use base "Exporter";
 
-$AnyEvent::MP::Kernel::WARN->(7, "starting log catcher service.");
+AE::log 7 => "starting log catcher service.";
 
 our $LOGLEVEL;
-our %lport; # local logging ports
+our $MON;
+our $PROPAGATE = 1; # set to one when messages ought to be send to remote nodes
+our %LPORT; # local logging ports
 
 # other nodes connect via this
 sub connect {
    my ($version, $rport, $loglevel) = @_;
 
-   my $cb = sub {
-      snd $rport, @_
-         if $_[0] <= $loglevel;
-   };
-
-   rcv $SELF, sub { };#d#
+   # context to catch log messages
+   my $ctx = new AnyEvent::Log::Ctx
+      title  => "AnyEvent::MP::LogCatcher",
+      level  => $loglevel,
+      log_cb => sub {
+         snd $rport, @{ $_[0] }
+            if $PROPAGATE;
+      },
+      fmt_cb => sub {
+         [$_[0], $_[1]->title, $_[2], $_[3]]
+      },
+   ;
 
-   push @AnyEvent::MP::Kernel::WARN, $cb;
+   $AnyEvent::Log::COLLECT->attach ($ctx);
 
-   # monitor them, silently die
+   # monitor them, silently die if they die
    mon $rport, sub {
-      @AnyEvent::MP::Kernel::WARN = grep $_ != $cb, @AnyEvent::MP::Kernel::WARN;
+      $AnyEvent::Log::COLLECT->detach ($ctx);
    };
+
+   AE::log 8 => "starting to propagate log messages to $rport";
 }
 
 sub mon_node {
-   my ($node, $is_up) = @_;
+   my ($node) = @_;
 
-   return unless $is_up;
+   # don't log messages from ourselves
+   return if $node eq $NODE;
 
-   my $lport = $lport{$node} = port {
-      my ($level, $msg) = @_;
+   $LPORT{$node} ||= do {
+      my $lport = port {
+         my ($time, $ctx, $level, $msg) = @_;
 
-      $msg =~ s/\n$//;
+         $level = 2 if $level < 2; # do not exit just because others do so
 
-      printf STDERR "%s [%s] <%d> %s\n",
-             (POSIX::strftime "%Y-%m-%d %H:%M:%S", localtime time),
-             $node,
-             $level,
-             $msg;
-   };
+         my $diff = AE::now - $time;
+         $diff = (abs $diff) < 1e-3 ? "" : sprintf "%+.3fs", $diff;
 
-   # establish connection
-   my $rport = spawn $node, "AnyEvent::MP::LogCatcher::connect", 0, $lport, $LOGLEVEL;
+         local $PROPAGATE; # do not propagate to other nodes
+         (AnyEvent::Log::ctx $ctx)->log ($level, "[$node$diff] $msg");
+      };
 
-   mon $rport, $lport;
-   mon $lport, sub {
-      warn "$lport $rport <@_>\n";#d#
-   };
+      mon $lport, sub {
+         delete $LPORT{$node}
+            or return; # do not monitor if node is not there
+         AE::log error => "@_"
+            if @_; # log error if there really was one
+         mon_node ($node); # try to reocnnect
+      };
+
+      # establish connection
+      AnyEvent::MP::Kernel::snd_to_func $node, "AnyEvent::MP::LogCatcher::connect", 0, $lport, $LOGLEVEL;
+
+      mon $node, $lport;
+
+      $lport
+   }
 }
 
+=item AnyEvent::MP::LogCatcher::catch [$level]
+
+Starts catching all log messages from all nodes with level C<$level> or
+lower. If the C<$level> is C<undef>, then stop catching all messages
+again.
+
+Example: start a node that catches all messages (you might have to specify
+a suitable profile name).
+
+   AE_VERBOSE=9 aemp run profilename services '[["AnyEvent::MP::LogCatcher::catch",9]]'
+
+=cut
+
 sub catch {
    $LOGLEVEL = $_[0];
-   kil $_, "restart" for values %lport;
-   %lport = ();
+   kil $_ for values %LPORT;
+   %LPORT = ();
 
    return unless defined $LOGLEVEL;
 
-   mon_node $_, 1
-      for up_nodes;
+   $MON = db_mon "'l" => sub {
+      my ($family, $add, $chg, $del) = @_;
+
+      kil delete $LPORT{$_}
+         for @$del;
+
+      mon_node $_
+         for @$add;
+   };
 
-   mon_nodes \&mon_node;
    ()
 }
 
 =back
 
+=head1 LOGGING
+
+AnyEvent::MP::LogCatcher logs messages from remote nodes. It logs them
+into the original logging context and prepends the origin node name
+and, if the time difference is larger than 1e-4 seconds, also the time
+difference between local time and origin time.
+
 =head1 SEE ALSO
 
 L<AnyEvent::MP>.