src/allmydata/introducer/server.py

   1
   2 import time, os.path
   3 from zope.interface import implements
   4 from twisted.application import service
   5 from foolscap import Referenceable
   6 import allmydata
   7 from allmydata import node
   8 from allmydata.util import log
   9 from allmydata.introducer.interfaces import \
  10      RIIntroducerPublisherAndSubscriberService
  11 from allmydata.introducer.common import make_index
  12
  13 class IntroducerNode(node.Node):
  14     PORTNUMFILE = "introducer.port"
  15     NODETYPE = "introducer"
  16
  17     def __init__(self, basedir="."):
  18         node.Node.__init__(self, basedir)
  19         self.read_config()
  20         self.init_introducer()
  21         webport = self.get_config("node", "web.port", None)
  22         if webport:
  23             self.init_web(webport) # strports string
  24
  25     def init_introducer(self):
  26         introducerservice = IntroducerService(self.basedir)
  27         self.add_service(introducerservice)
  28
  29         d = self.when_tub_ready()
  30         def _publish(res):
  31             self.introducer_url = self.tub.registerReference(introducerservice,
  32                                                              "introducer")
  33             self.log(" introducer is at %s" % self.introducer_url)
  34             self.write_config("introducer.furl", self.introducer_url + "\n")
  35         d.addCallback(_publish)
  36         d.addErrback(log.err, facility="tahoe.init",
  37                      level=log.BAD, umid="UaNs9A")
  38
  39     def init_web(self, webport):
  40         self.log("init_web(webport=%s)", args=(webport,))
  41
  42         from allmydata.webish import IntroducerWebishServer
  43         nodeurl_path = os.path.join(self.basedir, "node.url")
  44         ws = IntroducerWebishServer(webport, nodeurl_path)
  45         self.add_service(ws)
  46
  47 class IntroducerService(service.MultiService, Referenceable):
  48     implements(RIIntroducerPublisherAndSubscriberService)
  49     name = "introducer"
  50     VERSION = { "http://allmydata.org/tahoe/protocols/introducer/v1":
  51                  { },
  52                 "application-version": str(allmydata.__full_version__),
  53                 }
  54
  55     def __init__(self, basedir="."):
  56         service.MultiService.__init__(self)
  57         self.introducer_url = None
  58         # 'index' is (tubid, service_name)
  59         self._announcements = {} # dict of index -> (announcement, timestamp)
  60         self._subscribers = {} # dict of (rref->timestamp) dicts
  61
  62     def log(self, *args, **kwargs):
  63         if "facility" not in kwargs:
  64             kwargs["facility"] = "tahoe.introducer"
  65         return log.msg(*args, **kwargs)
  66
  67     def get_announcements(self):
  68         return self._announcements
  69     def get_subscribers(self):
  70         return self._subscribers
  71
  72     def remote_get_version(self):
  73         return self.VERSION
  74
  75     def remote_publish(self, announcement):
  76         self.log("introducer: announcement published: %s" % (announcement,) )
  77         index = make_index(announcement)
  78         if index in self._announcements:
  79             (old_announcement, timestamp) = self._announcements[index]
  80             if old_announcement == announcement:
  81                 self.log("but we already knew it, ignoring", level=log.NOISY)
  82                 return
  83             else:
  84                 self.log("old announcement being updated", level=log.NOISY)
  85         self._announcements[index] = (announcement, time.time())
  86         (furl, service_name, ri_name, nickname, ver, oldest) = announcement
  87         for s in self._subscribers.get(service_name, []):
  88             s.callRemote("announce", set([announcement]))
  89
  90     def remote_subscribe(self, subscriber, service_name):
  91         self.log("introducer: subscription[%s] request at %s" % (service_name,
  92                                                                  subscriber))
  93         if service_name not in self._subscribers:
  94             self._subscribers[service_name] = {}
  95         subscribers = self._subscribers[service_name]
  96         if subscriber in subscribers:
  97             self.log("but they're already subscribed, ignoring",
  98                      level=log.UNUSUAL)
  99             return
 100         subscribers[subscriber] = time.time()
 101         def _remove():
 102             self.log("introducer: unsubscribing[%s] %s" % (service_name,
 103                                                            subscriber))
 104             subscribers.pop(subscriber, None)
 105         subscriber.notifyOnDisconnect(_remove)
 106
 107         announcements = set( [ ann
 108                                for idx,(ann,when) in self._announcements.items()
 109                                if idx[1] == service_name] )
 110         d = subscriber.callRemote("announce", announcements)
 111         d.addErrback(log.err, facility="tahoe.introducer", level=log.UNUSUAL)
 112
 113
 114