src/allmydata/introducer.py

   1
   2 import re
   3 from zope.interface import implements
   4 from twisted.application import service
   5 from twisted.python import log
   6 from foolscap import Referenceable
   7 from allmydata.interfaces import RIIntroducer, RIIntroducerClient
   8 from allmydata.util import idlib, observer
   9
  10 class Introducer(service.MultiService, Referenceable):
  11     implements(RIIntroducer)
  12
  13     def __init__(self):
  14         service.MultiService.__init__(self)
  15         self.nodes = set()
  16         self.furls = set()
  17
  18     def remote_hello(self, node, furl):
  19         log.msg("introducer: new contact at %s, node is %s" % (furl, node))
  20         def _remove():
  21             log.msg(" introducer: removing %s %s" % (node, furl))
  22             self.nodes.remove(node)
  23             self.furls.remove(furl)
  24         node.notifyOnDisconnect(_remove)
  25         self.furls.add(furl)
  26         node.callRemote("new_peers", self.furls)
  27         for othernode in self.nodes:
  28             othernode.callRemote("new_peers", set([furl]))
  29         self.nodes.add(node)
  30
  31
  32 class IntroducerClient(service.Service, Referenceable):
  33     implements(RIIntroducerClient)
  34
  35     def __init__(self, tub, introducer_furl, my_furl):
  36         self.tub = tub
  37         self.introducer_furl = introducer_furl
  38         self.my_furl = my_furl
  39
  40         self.connections = {} # k: nodeid, v: ref
  41         self.reconnectors = {} # k: FURL, v: reconnector
  42
  43         self.connection_observers = observer.ObserverList()
  44
  45     def startService(self):
  46         self.introducer_reconnector = self.tub.connectTo(self.introducer_furl,
  47                                                          self._got_introducer)
  48
  49     def log(self, msg):
  50         self.parent.log(msg)
  51
  52     def remote_new_peers(self, furls):
  53         for furl in furls:
  54             self._new_peer(furl)
  55
  56     def stopService(self):
  57         service.Service.stopService(self)
  58         self.introducer_reconnector.stopConnecting()
  59         for reconnector in self.reconnectors.itervalues():
  60             reconnector.stopConnecting()
  61
  62     def _new_peer(self, furl):
  63         if furl in self.reconnectors:
  64             return
  65         # TODO: rather than using the TubID as a nodeid, we should use
  66         # something else. The thing that requires the least additional
  67         # mappings is to use the foolscap "identifier" (the last component of
  68         # the furl), since these are unguessable. Before we can do that,
  69         # though, we need a way to conveniently make these identifiers
  70         # persist from one run of the client program to the next. Also, using
  71         # the foolscap identifier would mean that anyone who knows the name
  72         # of the node also has all the secrets they need to contact and use
  73         # them, which may or may not be what we want.
  74         m = re.match(r'pb://(\w+)@', furl)
  75         assert m
  76         nodeid = idlib.a2b(m.group(1))
  77         def _got_peer(rref):
  78             self.log(" connected to(%s)" % idlib.b2a(nodeid))
  79             self.connection_observers.notify(nodeid, rref)
  80             self.connections[nodeid] = rref
  81             def _lost():
  82                 # TODO: notifyOnDisconnect uses eventually(), but connects do
  83                 # not. Could this cause a problem?
  84                 del self.connections[nodeid]
  85             rref.notifyOnDisconnect(_lost)
  86         self.log(" connecting to(%s)" % furl)
  87         self.reconnectors[furl] = self.tub.connectTo(furl, _got_peer)
  88
  89     def _got_introducer(self, introducer):
  90         self.log(" introducing ourselves: %s, %s" % (self, self.my_furl))
  91         d = introducer.callRemote("hello",
  92                              node=self,
  93                              furl=self.my_furl)
  94
  95     def notify_on_new_connection(self, cb):
  96         """Register a callback that will be fired (with nodeid, rref) when
  97         a new connection is established."""
  98         self.connection_observers.subscribe(cb)
  99