tord/pyro.py

   1 import Pyro.core
   2 from socket import gethostname, gethostbyname
   3 from threading import Thread
   4 from common import log
   5
   6 def ping(uri):
   7   try: Pyro.core.getProxyForURI(uri)
   8   except: return False
   9   return True
  10
  11 def shutdown(obj, daemon, client = False):
  12   obj.looping = False
  13   obj.shutdown()
  14   daemon.shutdown()
  15
  16 def run(obj, name, client = False):
  17   Pyro.core.initServer()
  18   if client: Pyro.core.initClient()
  19
  20   daemon = Pyro.core.Daemon()
  21   uri = daemon.connect(obj, name)
  22   log.info(uri)
  23
  24   try:
  25     addr = gethostbyname(gethostname())
  26     if client: obj.register(addr, obj.get_core())
  27     obj.start()
  28     daemon.requestLoop(lambda: obj.looping, 1)
  29   except: log.debug("Error starting: "+name)
  30   finally: shutdown(obj, daemon, client)
  31
  32 class PyroBase(Thread, Pyro.core.ObjBase):
  33     def __init__(self):
  34       Pyro.core.ObjBase.__init__(self)
  35       Thread.__init__(self)
  36       self.setDaemon(True)
  37       self.looping = True
  38
  39     def run(self):
  40       self.go()
  41       self.looping = False
  42
  43 def PyroClient(Client, server=None):
  44   class PyroClient(PyroBase, Client):
  45     def __init__(self, uri):
  46       PyroBase.__init__(self)
  47       Client.__init__(self)
  48       self.uri = uri
  49
  50     def get_core(self):
  51       return Pyro.core.getProxyForURI(self.uri)
  52
  53     def register(self, name, core):
  54       core.join(name, self.getProxy())
  55
  56   if not server: server = gethostname()
  57   uri = "PYROLOC://"+server+":7766/tord"
  58   if ping(uri):
  59     run(PyroClient(uri), "torc", True)
  60
  61 def PyroServer(Server, name):
  62   class PyroServer(PyroBase, Server):
  63     def __init__(self):
  64       PyroBase.__init__(self)
  65       Server.__init__(self)
  66
  67       self.publishMutex = Pyro.util.getLockObject()
  68       self.clients = []
  69
  70     def join(self, name, callback):
  71       log.debug(("joined:", name, callback))
  72       self.clients.append((name, callback))
  73
  74     def publish(self, msg):
  75       for name, cb in self.clients:
  76         try:
  77           self.publishMutex.acquire()
  78           try:
  79             cb._transferThread()
  80             cb.message(msg)
  81           finally:
  82             self.publishMutex.release()
  83         except:
  84           log.debug(("removed:", name, cb))
  85           if (name, cb) in self.clients:
  86             self.clients.remove((name, cb))
  87
  88   run(PyroServer(), "tord")