def start_connecting(tub, trigger_cb):
pass
def get_rref():
- pass
+ """Once a server is connected, I return a RemoteReference.
+ Before a server is connected for the first time, I return None.
+
+ Note that the rref I return will start producing DeadReferenceErrors
+ once the connection is lost.
+ """
class IMutableSlotWriter(Interface):
def test_add_rref(self, serverid, rref, ann):
s = NativeStorageServer(serverid, ann.copy())
s.rref = rref
+ s._is_connected = True
self.servers[serverid] = s
def test_add_server(self, serverid, s):
return frozenset(self.servers.keys())
def get_connected_servers(self):
- return frozenset([s for s in self.servers.values() if s.get_rref()])
+ return frozenset([s for s in self.servers.values() if s.is_connected()])
def get_known_servers(self):
return frozenset(self.servers.values())
self.last_loss_time = None
self.remote_host = None
self.rref = None
+ self._is_connected = False
self._reconnector = None
self._trigger_cb = None
return self.announcement
def get_remote_host(self):
return self.remote_host
+ def is_connected(self):
+ return self._is_connected
def get_last_connect_time(self):
return self.last_connect_time
def get_last_loss_time(self):
self.last_connect_time = time.time()
self.remote_host = rref.getPeer()
self.rref = rref
+ self._is_connected = True
rref.notifyOnDisconnect(self._lost)
def get_rref(self):
log.msg(format="lost connection to %(name)s", name=self.get_name(),
facility="tahoe.storage_broker", umid="zbRllw")
self.last_loss_time = time.time()
- self.rref = None
+ # self.rref is now stale: all callRemote()s will get a
+ # DeadReferenceError. We leave the stale reference in place so that
+ # uploader/downloader code (which received this IServer through
+ # get_connected_servers() or get_servers_for_psi()) can continue to
+ # use s.get_rref().callRemote() and not worry about it being None.
+ self._is_connected = False
self.remote_host = None
def stop_connecting(self):
return d
d.addCallback(_got_lit_filenode)
return d
+
+class Connections(SystemTestMixin, unittest.TestCase):
+ def test_rref(self):
+ self.basedir = "system/Connections/rref"
+ d = self.set_up_nodes(2)
+ def _start(ign):
+ self.c0 = self.clients[0]
+ for s in self.c0.storage_broker.get_connected_servers():
+ if "pub-"+s.get_longname() != self.c0.node_key_s:
+ break
+ self.s1 = s # s1 is the server, not c0
+ self.s1_rref = s.get_rref()
+ self.failIfEqual(self.s1_rref, None)
+ self.failUnless(self.s1.is_connected())
+ d.addCallback(_start)
+
+ # now shut down the server
+ d.addCallback(lambda ign: self.clients[1].disownServiceParent())
+ # and wait for the client to notice
+ def _poll():
+ return len(self.c0.storage_broker.get_connected_servers()) < 2
+ d.addCallback(lambda ign: self.poll(_poll))
+
+ def _down(ign):
+ self.failIf(self.s1.is_connected())
+ rref = self.s1.get_rref()
+ self.failUnless(rref)
+ self.failUnlessIdentical(rref, self.s1_rref)
+ d.addCallback(_down)
+ return d