r46817 - lots more edge cases, stop testing internal state

Author: glyph
Date: Thu Feb 18 01:19:13 2016
New Revision: 46817

Modified:
   branches/persistent-client-service-4735-5/twisted/application/internet.py
   branches/persistent-client-service-4735-5/twisted/application/test/test_internet.py

Log:
lots more edge cases, stop testing internal state

Modified: branches/persistent-client-service-4735-5/twisted/application/internet.py
==============================================================================
--- branches/persistent-client-service-4735-5/twisted/application/internet.py	(original)
+++ branches/persistent-client-service-4735-5/twisted/application/internet.py	Thu Feb 18 01:19:13 2016
@@ -525,7 +525,8 @@
 class ClientService(service.Service, object):
     """
     A L{ClientService} maintains a single outgoing connection to a client
-    endpoint, with configurable timeout policies.
+    endpoint, reconnecting after a configurable timeout when a connection
+    fails, either before or after connecting.
     """
 
     _log = Logger()
@@ -564,11 +565,34 @@
         self._connectionInProgress = succeed(None)
         self._loseConnection = lambda: None
 
+        self._currentConnection = None
+        self._awaitingConnected = []
+
+
+    def whenConnected(self):
+        """
+        Retrieve the currently-connected L{Protocol}, or the next one to
+        connect.
+
+        @return: a Deferred that fires with a protocol produced by the factory
+            passed to C{__init__}
+        @rtype: L{Deferred} firing with L{IProtocol} or failing with
+            L{CancelledError} the service is stopped.
+        """
+        if self._currentConnection is not None:
+            return succeed(self._currentConnection)
+        else:
+            # XXX WROOONG
+            return Deferred()
+
 
     def startService(self):
         """
         Start this L{ClientService}, initiating the connection retry loop.
         """
+        if self.running:
+            self._log.warn("Duplicate ClientService.startService {log_source}")
+            return
         super(ClientService, self).startService()
         self._failedAttempts = 0
 
@@ -576,11 +600,16 @@
             self._failedAttempts = 0
             self._loseConnection = protocol.transport.loseConnection
             self._lostDeferred = Deferred()
+            self._currentConnection = protocol._protocol
+            self._awaitingConnected, waiting = [], self._awaitingConnected
+            for w in waiting:
+                w.callback(self._currentConnection)
 
         def clientDisconnect(reason):
+            self._currentConnection = None
             self._loseConnection = lambda: None
             self._lostDeferred.callback(None)
-            # XXX SHOULD BE A retry() HERE
+            retry(reason)
 
         factoryProxy = _DisconnectFactory(self._factory, clientDisconnect)
 
@@ -590,7 +619,9 @@
                                           .addCallback(clientConnect)
                                           .addErrback(retry))
 
-        def retry(error=None):
+        def retry(failure):
+            if not self.running:
+                return
             self._failedAttempts += 1
             delay = self._timeoutForAttempt(self._failedAttempts)
             self._log.info("Scheduling retry {attempt} to connect {endpoint} "

Modified: branches/persistent-client-service-4735-5/twisted/application/test/test_internet.py
==============================================================================
--- branches/persistent-client-service-4735-5/twisted/application/test/test_internet.py	(original)
+++ branches/persistent-client-service-4735-5/twisted/application/test/test_internet.py	Thu Feb 18 01:19:13 2016
@@ -432,6 +432,7 @@
     def __init__(self):
         self.connectQueue = []
         self.constructedProtocols = []
+        self.applicationProtocols = []
 
 
 
@@ -499,12 +500,21 @@
         nkw.update(clock=Clock())
         nkw.update(kw)
         cq, endpoint = endpointForTesting(fireImmediately=fireImmediately)
-        factory = Factory.forProtocol(Protocol)
+        class RememberingFactory(Factory, object):
+            protocol = Protocol
+            def buildProtocol(self, addr):
+                result = super(RememberingFactory, self).buildProtocol(addr)
+                cq.applicationProtocols.append(result)
+                return result
+        factory = RememberingFactory()
         service = ClientService(endpoint, factory, **nkw)
         def stop():
             service._protocol = None
             if service.running:
                 service.stopService()
+            # Ensure that we don't leave any state in the reactor after
+            # stopService.
+            self.assertEqual(service._clock.getDelayedCalls(), [])
         self.addCleanup(stop)
         if startService:
             service.startService()
@@ -568,8 +578,10 @@
         """
         clock = Clock()
         cq, service = self.makeReconnector(clock=clock)
+        awaitingProtocol = service.whenConnected()
         self.assertEqual(clock.getDelayedCalls(), [])
-        self.assertIdentical(service._protocol, cq.constructedProtocols[0])
+        self.assertIdentical(self.successResultOf(awaitingProtocol),
+                             cq.applicationProtocols[0])
 
 
     def test_clientConnectionFailed(self):
@@ -582,7 +594,7 @@
                                            clock=clock)
         self.assertEqual(len(cq.connectQueue), 1)
         cq.connectQueue[0].errback(Failure(Exception()))
-        self.assertIdentical(service._protocol, None)
+        self.assertNoResult(service.whenConnected())
         clock.advance(100.)
         self.assertEqual(len(cq.connectQueue), 2)
 
@@ -592,10 +604,19 @@
         When a client connection is lost, the service removes its reference
         to the protocol and calls retry.
         """
-        cq, service = self.makeReconnector()
-        service.startService()
+        clock = Clock()
+        cq, service = self.makeReconnector(clock=clock, fireImmediately=False)
+        self.assertEquals(len(cq.connectQueue), 1)
+        cq.connectQueue[0].callback(None)
+        self.assertEquals(len(cq.connectQueue), 1)
+        self.assertIdentical(self.successResultOf(service.whenConnected()),
+                             cq.applicationProtocols[0])
         cq.constructedProtocols[0].connectionLost(Failure(Exception()))
-        self.assertIdentical(service._protocol, None)
+        clock.advance(100.)
+        self.assertEquals(len(cq.connectQueue), 2)
+        cq.connectQueue[1].callback(None)
+        self.assertIdentical(self.successResultOf(service.whenConnected()),
+                             cq.applicationProtocols[1])
 
 
     def test_clientConnectionLostWhileStopping(self):
@@ -606,6 +627,23 @@
         """
         cq, service = self.makeReconnector()
         d = service.stopService()
-        cq.constructedProtocols[0].connectionLost(Failure(Exception()))
-        self.assertIdentical(service._protocol, None)
+        cq.constructedProtocols[0].connectionLost(Failure(IndentationError()))
+        self.assertFailure(service.whenConnected(), CancelledError)
         self.assertTrue(d.called)
+
+
+    def test_startTwice(self):
+        """
+        If L{ClientService} is started when it's already started, it will log a
+        complaint and do nothing else (in particular it will not make
+        additional connections).
+        """
+        cq, service = self.makeReconnector(fireImmediately=False,
+                                           startService=False)
+        self.assertEqual(len(cq.connectQueue), 0)
+        service.startService()
+        self.assertEqual(len(cq.connectQueue), 1)
+        messages = catchLogs(self)
+        service.startService()
+        self.assertEqual(len(cq.connectQueue), 1)
+        self.assertIn("Duplicate ClientService.startService", messages()[0])