src/allmydata/logpublisher.py

   1
   2 import os.path
   3 from zope.interface import implements
   4 from twisted.application import service
   5 from twisted.python import log
   6 from foolscap import Referenceable, RemoteInterface
   7 from foolscap.schema import DictOf, Any
   8 from allmydata import get_package_versions
   9
  10 class RILogObserver(RemoteInterface):
  11     def msg(logmsg=DictOf(str, Any())):
  12         return None
  13 class RISubscription(RemoteInterface):
  14     pass
  15
  16 class RILogPublisher(RemoteInterface):
  17     def get_versions():
  18         return DictOf(str, str)
  19     def subscribe_to_all(observer=RILogObserver):
  20         return RISubscription
  21     def unsubscribe(subscription=Any()):
  22         # I don't know how to get the constraint right: unsubscribe() should
  23         # accept return value of subscribe_to_all()
  24         return None
  25
  26 class RILogGatherer(RemoteInterface):
  27     def logport(nodeid=str, logport=RILogPublisher):
  28         return None
  29
  30 class Subscription(Referenceable):
  31     implements(RISubscription)
  32
  33 class LogPublisher(Referenceable, service.MultiService):
  34     implements(RILogPublisher)
  35     name = "log_publisher"
  36
  37     def __init__(self):
  38         service.MultiService.__init__(self)
  39         self._subscribers = {}
  40         self._notifyOnDisconnectors = {}
  41
  42     def startService(self):
  43         service.MultiService.startService(self)
  44         furlfile = os.path.join(self.parent.basedir, "logport.furl")
  45         self.parent.tub.registerReference(self, furlFile=furlfile)
  46         os.chmod(furlfile, 0600)
  47
  48         log.addObserver(self._twisted_log_observer)
  49
  50     def stopService(self):
  51         log.removeObserver(self._twisted_log_observer)
  52         return service.MultiService.stopService(self)
  53
  54     def _twisted_log_observer(self, d):
  55         # Twisted will remove this for us if it fails.
  56
  57         # keys:
  58         #  ['message']: *args
  59         #  ['time']: float
  60         #  ['isError']: bool, usually False
  61         #  ['system']: string
  62
  63         for o in self._subscribers.values():
  64             o.callRemoteOnly("msg", d)
  65
  66         #f = open("/tmp/f.out", "a")
  67         #print >>f, d['message']
  68         #f.close()
  69
  70     def remote_get_versions(self):
  71         # Convert all the version instances to strings.
  72         return dict([(k,str(v))
  73                      for k,v in get_package_versions().iteritems()])
  74
  75     def remote_subscribe_to_all(self, observer):
  76         s = Subscription()
  77         self._subscribers[s] = observer
  78         c = observer.notifyOnDisconnect(self.remote_unsubscribe, s)
  79         self._notifyOnDisconnectors[s] = c
  80         return s
  81
  82     def remote_unsubscribe(self, s):
  83         observer = self._subscribers.pop(s)
  84         c = self._notifyOnDisconnectors.pop(s)
  85         observer.dontNotifyOnDisconnect(c)
  86