Merge lp:~free.ekanayaka/landscape-client/amp-cleanup-2 into lp:~landscape/landscape-client/trunk

[Free Ekanayaka]

This branch is a step toward the goal of fully-synchronous tests in the client. In detail:

- Drop the retry_on_reconnect parameter of the RemoteBroker constructor, and
  stick it to MethodCallClientFactory instead. The idea is to make the factory
  a sort of controller of RemoteBroker. Since user code is supposed to deal
  with factories directly, the can control the RemoteBroker's behavior by
  modifying the factory.

- Move the MethodCallFactory class after the RemoteBroker class, since we now need
  to set the latter as 'remote' class attribute of the former.

- Rename MethodCallFactory to MethodCallClientFactory, since this is really
  a protocol factory for clients, not for servers.

- Implement a new interface for MethodCallClientFactory, which should be more convenient
  for user-code. This includes adding a getRemoteObject method and replacing the
  notify_success and notify_failure methods with a single notifyOnConnect (camel case
  for consistency with Twisted).

This branch is a step toward the goal of fully-synchronous tests in the client. In detail:

- Drop the retry_on_reconnect parameter of the RemoteBroker constructor, and
  stick it to MethodCallClientFactory instead. The idea is to make the factory
  a sort of controller of RemoteBroker. Since user code is supposed to deal
  with factories directly, the can control the RemoteBroker's behavior by
  modifying the factory.

- Move the MethodCallFactory class after the RemoteBroker class, since we now need
  to set the latter as 'remote' class attribute of the former.

- Rename MethodCallFactory to MethodCallClientFactory, since this is really
  a protocol factory for clients, not for servers.

- Implement a new interface for MethodCallClientFactory, which should be more convenient
  for user-code. This includes adding a getRemoteObject method and replacing the
  notify_success and notify_failure methods with a single notifyOnConnect (camel case
  for consistency with Twisted).

[Geoff Teale (tealeg)]

+1 Sweet

[1] this text is a little hard to follow:

248 + @ivar retryOnReconnect: If C{True}, the remote object returned by the
249 + C{getRemoteObject} method will retry to perform again requests that
250 + failed due to a lost connection, as soon as a new connection is
251 + available.

I'd rewrite it as:

 @ivar retryOnReconnect: If C{True}, the remote object returned
 by the C{getRemoteObject} method will retry requests that
 failed, as a result of a lost connection, as soon as a new
 connection is available.

[2] "a the" should be just "a" or just "the". Either works, I'd go for "a".

517 + """
518 + The C{notifyOnConnect} method supports specifying a callback that
519 + will be invoked when a the connection has been established.
520 + """

[3] Again a little hard to follow:

528 + """
529 + The C{notifyOnConnect} fires the callbacks also then a connection is
530 + re-established after it was lost.
531 + """

Something like this would be better:

 The C{notifyOnConnect} will also callback when a connection is
 re-established after it was lost.

review: Approve

[lp:~free.ekanayaka/landscape-client/amp-cleanup-2]

663. By Free Ekanayaka on 2013-04-23

Fix wording and typos

[Free Ekanayaka (free.ekanayaka)]

[1] to [3] all fixed. Thanks!

[Alberto Donato (ack)]

Looks good! +1

#1:
        def synchronous_send_method_call(method, args=[], kwargs={}):
            # Transparently flush the connection after a send_method_call
            # invokation

Typo, not from this branch: s/invokation/invocation/.

review: Approve

[lp:~free.ekanayaka/landscape-client/amp-cleanup-2]

664. By Free Ekanayaka on 2013-04-23

Fix typo

[Free Ekanayaka (free.ekanayaka)]

#1 Fixed.

=== modified file 'landscape/amp.py'
--- landscape/amp.py	2013-04-12 07:49:34 +0000
+++ landscape/amp.py	2013-04-23 13:44:28 +0000
@@ -51,12 +51,13 @@
 
     factory = ComponentProtocolFactory
 
-    def __init__(self, reactor, config, *args, **kwargs):
+    def __init__(self, reactor, config, retry_on_reconnect=False):
         self._twisted_reactor = reactor
+        self._retry_on_reconnect = retry_on_reconnect
         socket = os.path.join(config.sockets_path,
                               self.component.name + ".sock")
         super(RemoteComponentConnector, self).__init__(
-            self._twisted_reactor._reactor, socket, *args, **kwargs)
+            self._twisted_reactor._reactor, socket)
 
     def connect(self, max_retries=None, factor=None, quiet=False):
         """Connect to the remote Landscape component.
@@ -73,12 +74,15 @@
         @param quiet: A boolean indicating whether to log errors.
         """
 
-        def fire_reconnect(remote):
+        def fire_reconnect(ignored):
             self._twisted_reactor.fire("%s-reconnect" %
                                        self.component.name)
 
         def connected(remote):
-            self._factory.add_notifier(fire_reconnect)
+            # XXX temporary workaround till the AMP cleanup is completed, we
+            # will use the factory directly then.
+            remote._factory.retryOnReconnect = self._retry_on_reconnect
+            self._factory.notifyOnConnect(fire_reconnect)
             return remote
 
         def log_error(failure):

=== modified file 'landscape/broker/amp.py'
--- landscape/broker/amp.py	2013-03-18 03:46:10 +0000
+++ landscape/broker/amp.py	2013-04-23 13:44:28 +0000
@@ -101,6 +101,7 @@
 class BrokerClientProtocolFactory(ComponentProtocolFactory):
 
     protocol = BrokerClientProtocol
+    remote = RemoteBroker
 
 
 class RemoteClient(RemoteObject):

=== modified file 'landscape/lib/amp.py'
--- landscape/lib/amp.py	2013-04-19 11:26:09 +0000
+++ landscape/lib/amp.py	2013-04-23 13:44:28 +0000
@@ -45,7 +45,7 @@
 for more details about the Twisted AMP protocol.
 """
 
-from twisted.internet.defer import Deferred, maybeDeferred
+from twisted.internet.defer import Deferred, maybeDeferred, succeed
 from twisted.internet.protocol import ReconnectingClientFactory
 from twisted.python.failure import Failure
 
@@ -277,7 +277,11 @@
 
 
 class MethodCallClientProtocol(AMP):
-    """XXX Placeholder"""
+    """Send L{MethodCall} commands over the wire using the AMP protocol."""
+
+    def connectionMade(self):
+        """Notify our factory that we're ready to go."""
+        self.factory.clientConnectionMade(self)
 
 
 class MethodCallProtocol(MethodCallServerProtocol, MethodCallClientProtocol):
@@ -288,81 +292,6 @@
         MethodCallClientProtocol.__init__(self)
 
 
-class MethodCallFactory(ReconnectingClientFactory):
-    """
-    Factory for L{MethodCallProtocol}s exposing an object or connecting to
-    L{MethodCall} servers.
-
-    When used to connect, if the connection fails or is lost the factory
-    will keep retrying to establish it.
-
-    @cvar protocol: The factory used to build protocol instances.
-    @cvar factor: The time factor by which the delay between two subsequent
-        connection retries will increase.
-    @cvar maxDelay: Maximum number of seconds between connection attempts.
-    """
-
-    protocol = MethodCallProtocol
-    factor = 1.6180339887498948
-    maxDelay = 30
-
-    def __init__(self, object=None, reactor=None):
-        """
-        @param object: The object exposed by the L{MethodCallProtocol}s
-            instances created by this factory.
-        @param reactor: The reactor used by the created protocols
-            to schedule notifications and timeouts.
-        """
-        self.object = object
-        self.reactor = reactor
-        self.clock = self.reactor
-        self.delay = self.initialDelay
-        self._notifiers = []
-
-    def add_notifier(self, callback, errback=None):
-        """Call the given function on connection, reconnection or give up.
-
-        @param notifier: A function that will be called when the factory builds
-            a new connected protocol or gives up connecting.  It will be passed
-            the new protocol instance as argument, or the connectionf failure.
-        """
-        self._notifiers.append((callback, errback))
-
-    def remove_notifier(self, callback, errback=None):
-        """Remove a notifier."""
-        self._notifiers.remove((callback, errback))
-
-    def notify_success(self, *args, **kwargs):
-        """Notify all registered notifier callbacks."""
-        for callback, _ in self._notifiers:
-            self.reactor.callLater(0, callback, *args, **kwargs)
-
-    def notify_failure(self, failure):
-        """Notify all registered notifier errbacks."""
-        for _, errback in self._notifiers:
-            if errback is not None:
-                self.reactor.callLater(0, errback, failure)
-
-    def clientConnectionFailed(self, connector, reason):
-        ReconnectingClientFactory.clientConnectionFailed(self, connector,
-                                                         reason)
-        if self.maxRetries is not None and (self.retries > self.maxRetries):
-            self.notify_failure(reason)  # Give up
-
-    def buildProtocol(self, addr):
-        self.resetDelay()
-        if self.object is not None:
-            # XXX temporary hack to emulate the behavior of this code before
-            # MethodCallReceiver was introduced
-            locator = MethodCallReceiver(self.object, self.protocol.methods)
-            protocol = AMP(locator=locator)
-            protocol.factory = self
-        else:
-            protocol = ReconnectingClientFactory.buildProtocol(self, addr)
-        self.notify_success(protocol)
-        return protocol
-
-
 class RemoteObject(object):
     """An object able to transparently call methods on a remote object.
 
@@ -371,8 +300,7 @@
     the remote object exposed by the peer.
     """
 
-    def __init__(self, sender, retry_on_reconnect=False, timeout=None,
-                 factory=None):
+    def __init__(self, factory):
         """
         @param protocol: A reference to a connected L{AMP} protocol instance,
             which will be used to send L{MethodCall} commands.
@@ -383,16 +311,10 @@
             can't perform them again successfully within this number of
             seconds, they will errback with a L{MethodCallError}.
         """
-        self._sender = sender
+        self._sender = None
+        self._pending_requests = {}
         self._factory = factory
-        self._retry_on_reconnect = retry_on_reconnect
-        self._timeout = timeout
-        self._pending_requests = {}
-        if self._factory:
-            # XXX temporary hack to emulate the behavior of this code before
-            # MethodCallReceiver was introduced
-            self._reactor = factory.reactor
-            self._factory.add_notifier(self._handle_reconnect)
+        self._factory.notifyOnConnect(self._handle_connect)
 
     def __getattr__(self, method):
         """Return a function sending a L{MethodCall} for the given C{method}.
@@ -445,7 +367,7 @@
             the given deferred.
         """
         is_method_call_error = failure.type is MethodCallError
-        dont_retry = self._retry_on_reconnect is False
+        dont_retry = self._factory.retryOnReconnect is False
 
         if is_method_call_error or dont_retry:
             # This means either that the connection is working, and a
@@ -459,23 +381,26 @@
             deferred.errback(failure)
             return
 
-        if self._timeout and call is None:
+        if self._factory.retryTimeout and call is None:
             # This is the first failure for this request, let's schedule a
             # timeout call.
             timeout = Failure(MethodCallError("timeout"))
-            call = self._reactor.callLater(self._timeout,
-                                           self._handle_failure,
-                                           timeout, method, args,
-                                           kwargs, deferred=deferred)
+            call = self._factory.clock.callLater(self._factory.retryTimeout,
+                                                 self._handle_failure,
+                                                 timeout, method, args,
+                                                 kwargs, deferred=deferred)
 
         self._pending_requests[deferred] = (method, args, kwargs, call)
 
-    def _handle_reconnect(self, protocol):
+    def _handle_connect(self, protocol):
         """Handles a reconnection.
 
         @param protocol: The newly connected protocol instance.
         """
-        self._sender.protocol = protocol
+        if self._sender is None:
+            self._sender = MethodCallSender(protocol, self._factory.clock)
+        else:
+            self._sender.protocol = protocol
         if self._retry_on_reconnect:
             self._retry()
 
@@ -498,13 +423,124 @@
                               deferred=deferred, call=call)
 
 
+class MethodCallClientFactory(ReconnectingClientFactory):
+    """
+    Factory for L{MethodCallProtocol}s exposing an object or connecting to
+    L{MethodCall} servers.
+
+    When used to connect, if the connection fails or is lost the factory
+    will keep retrying to establish it.
+
+    @ivar factor: The time factor by which the delay between two subsequent
+        connection retries will increase.
+    @ivar maxDelay: Maximum number of seconds between connection attempts.
+    @ivar protocol: The factory used to build protocol instances.
+    @ivar remote: The factory used to build remote object instances.
+    @ivar retryOnReconnect: If C{True}, the remote object returned by the
+        C{getRemoteObject} method will retry requests that failed, as a
+        result of a lost connection, as soon as a new connection is available.
+    @param retryTimeout: A timeout for retrying requests, if the remote object
+        can't perform them again successfully within this number of seconds,
+        they will errback with a L{MethodCallError}.
+    """
+
+    factor = 1.6180339887498948
+    maxDelay = 30
+
+    protocol = MethodCallClientProtocol
+    remote = RemoteObject
+
+    retryOnReconnect = False
+    retryTimeout = None
+
+    def __init__(self, reactor=None, object=None):
+        """
+        @param object: The object exposed by the L{MethodCallProtocol}s
+            instances created by this factory.
+        @param reactor: The reactor used by the created protocols
+            to schedule notifications and timeouts.
+        """
+        self.object = object  # XXX
+        self.reactor = reactor
+        self.clock = self.reactor
+        self.delay = self.initialDelay
+        self._connects = []
+        self._requests = []
+        self._remote = None
+
+    def getRemoteObject(self):
+        """Get a L{RemoteObject} as soon as the connection is ready.
+
+        @return: A C{Deferred} firing with a connected L{RemoteObject}.
+        """
+        if self._remote is not None:
+            return succeed(self._remote)
+        deferred = Deferred()
+        self._requests.append(deferred)
+        return deferred
+
+    def notifyOnConnect(self, callback):
+        """Invoke the given C{callback} when a connection is re-established."""
+        self._connects.append(callback)
+
+    def dontNotifyOnConnect(self, callback):
+        """Remove the given C{callback} from listeners."""
+        self._connects.remove(callback)
+
+    def clientConnectionMade(self, protocol):
+        """Called when a newly built protocol gets connected."""
+        if self._remote is None:
+            # This is the first time we successfully connect
+            self._remote = self.remote(self)
+
+        for callback in self._connects:
+            callback(protocol)
+
+        # In all cases fire pending requests
+        self._fire_requests(self._remote)
+
+    def clientConnectionFailed(self, connector, reason):
+        """Try to connect again or errback pending request."""
+        ReconnectingClientFactory.clientConnectionFailed(self, connector,
+                                                         reason)
+        if self._callID is None:
+            # The factory won't retry to connect, so notify that we failed
+            self._fire_requests(reason)
+
+    def buildProtocol(self, addr):
+        self.resetDelay()
+        if self.object is not None:
+            # XXX temporary hack to emulate the behavior of this code before
+            # MethodCallReceiver was introduced
+            locator = MethodCallReceiver(self.object, self.protocol.methods)
+            protocol = AMP(locator=locator)
+            protocol.factory = self
+        else:
+            protocol = ReconnectingClientFactory.buildProtocol(self, addr)
+        return protocol
+
+    def _fire_requests(self, result):
+        """
+        Fire all pending L{getRemoteObject} deferreds with the given C{result}.
+        """
+        requests = self._requests[:]
+        self._requests = []
+
+        for deferred in requests:
+            deferred.callback(result)
+
+
+class MethodCallFactory(MethodCallClientFactory):
+    """XXX placeholder"""
+
+
 class RemoteObjectConnector(object):
     """Connect to remote objects exposed by a L{MethodCallProtocol}."""
 
-    factory = MethodCallFactory
+    factory = MethodCallClientFactory
     remote = RemoteObject
 
-    def __init__(self, reactor, socket_path, *args, **kwargs):
+    def __init__(self, reactor, socket_path):
         """
         @param reactor: A reactor able to connect to Unix sockets.
         @param socket: The path to the socket we want to connect to.
@@ -513,9 +549,6 @@
         """
         self._socket_path = socket_path
         self._reactor = reactor
-        self._args = args
-        self._kwargs = kwargs
-        self._remote = None
         self._factory = None
 
     def connect(self, max_retries=None, factor=None):
@@ -530,37 +563,18 @@
             delay between subsequent retries should increase. Smaller values
             result in a faster reconnection attempts pace.
         """
-        self._connected = Deferred()
         self._factory = self.factory(reactor=self._reactor)
         self._factory.maxRetries = max_retries
         if factor:
             self._factory.factor = factor
-        self._factory.add_notifier(self._success, self._failure)
         self._reactor.connectUNIX(self._socket_path, self._factory)
-        return self._connected
-
-    def _success(self, result):
-        """Called when the first connection has been established"""
-
-        # We did our job, remove our own notifier and let the remote object
-        # handle reconnections.
-        self._factory.remove_notifier(self._success, self._failure)
-        sender = MethodCallSender(result, self._reactor)
-        # XXX temporary hack to emulate the behavior of this code before
-        # MethodCallReceiver was introduced
-        self._kwargs["factory"] = self._factory
-        self._remote = self.remote(sender, *self._args, **self._kwargs)
-        self._connected.callback(self._remote)
-
-    def _failure(self, failure):
-        """Called when the first connection has failed"""
-        self._connected.errback(failure)
+        return self._factory.getRemoteObject()
 
     def disconnect(self):
         """Disconnect the L{RemoteObject} that we have created."""
         if self._factory:
             self._factory.stopTrying()
-        if self._remote:
-            if self._remote._sender.protocol.transport:
-                self._remote._sender.protocol.transport.loseConnection()
-            self._remote = None
+            remote = self._factory._remote
+            if remote:
+                if remote._sender.protocol.transport:
+                    remote._sender.protocol.transport.loseConnection()

=== modified file 'landscape/lib/tests/test_amp.py'
--- landscape/lib/tests/test_amp.py	2013-04-18 09:32:04 +0000
+++ landscape/lib/tests/test_amp.py	2013-04-23 13:44:28 +0000
@@ -4,12 +4,12 @@
 from twisted.internet.error import ConnectionDone, ConnectError
 from twisted.internet.task import Clock
 from twisted.protocols.amp import AMP
+from twisted.python.failure import Failure
 
 from landscape.lib.amp import (
-    MethodCallError, MethodCallProtocol, MethodCallFactory, RemoteObject,
+    MethodCallError, MethodCallProtocol, MethodCallClientFactory, RemoteObject,
     RemoteObjectConnector, MethodCallReceiver, MethodCallSender)
 from landscape.tests.helpers import LandscapeTest
-from landscape.tests.mocker import KWARGS
 
 
 class FakeTransport(object):
@@ -149,11 +149,14 @@
 METHODS = WordsProtocol.methods
 
 
-class WordsFactory(MethodCallFactory):
+class WordsFactory(MethodCallClientFactory):
 
     protocol = WordsProtocol
     factor = 0.19
 
+    retryOnReconnect = True
+    retryTimeout = 0.7
+
 
 class RemoteWordsConnector(RemoteObjectConnector):
 
@@ -339,18 +342,20 @@
         server = AMP(locator=MethodCallReceiver(Words(self.clock), METHODS))
         client = AMP()
         self.connection = FakeConnection(client, server)
-        sender = MethodCallSender(client, self.clock)
-        send_method_call = sender.send_method_call
+        factory = WordsFactory(self.clock)
+        self.remote = RemoteObject(factory)
+        factory.clientConnectionMade(client)
+
+        send_method_call = self.remote._sender.send_method_call
 
         def synchronous_send_method_call(method, args=[], kwargs={}):
             # Transparently flush the connection after a send_method_call
-            # invokation
+            # invocation
             deferred = send_method_call(method, args=args, kwargs=kwargs)
             self.connection.flush()
             return deferred
 
-        sender.send_method_call = synchronous_send_method_call
-        self.remote = RemoteObject(sender)
+        self.remote._sender.send_method_call = synchronous_send_method_call
 
     def test_method_call_sender_with_forbidden_method(self):
         """
@@ -526,12 +531,12 @@
         failure.trap(MethodCallError)
 
 
-class MethodCallFactoryTest(LandscapeTest):
+class MethodCallClientFactoryTest(LandscapeTest):
 
     def setUp(self):
-        super(MethodCallFactoryTest, self).setUp()
+        super(MethodCallClientFactoryTest, self).setUp()
         self.clock = Clock()
-        self.factory = MethodCallFactory(reactor=self.clock)
+        self.factory = MethodCallClientFactory(reactor=self.clock)
 
     def test_max_delay(self):
         """
@@ -540,60 +545,67 @@
         """
         self.assertEqual(self.factory.maxDelay, 30)
 
-    def test_add_notifier(self):
-        """
-        The L{MethodCallClientFactory.add_notifier} method can be used to
-        add a callback function to be called when a connection is made and
-        a new protocol instance built.
-        """
-        protocol = self.factory.protocol()
-        self.factory.protocol = lambda: protocol
-        callback = self.mocker.mock()
-        callback(protocol)
-        self.mocker.replay()
-        self.factory.add_notifier(callback)
-        self.factory.buildProtocol(None)
-        self.clock.advance(0)
-
-    def test_remove_notifier(self):
-        """
-        The L{MethodCallClientFactory.remove_notifier} method can be used to
-        remove a previously added notifier callback.
-        """
-        callback = lambda protocol: 1 / 0
-        self.factory.add_notifier(callback)
-        self.factory.remove_notifier(callback)
-        self.factory.buildProtocol(None)
-        self.clock.advance(0)
+    def test_connect_notifier(self):
+        """
+        The C{notifyOnConnect} method supports specifying a callback that
+        will be invoked when a connection has been established.
+        """
+        protocols = []
+        self.factory.notifyOnConnect(protocols.append)
+        protocol = self.factory.buildProtocol(None)
+        protocol.connectionMade()
+        self.assertEqual([protocol], protocols)
+
+    def test_connect_notifier_with_reconnect(self):
+        """
+        The C{notifyOnConnect} method will also callback when a connection is
+        re-established after it was lost.
+        """
+        protocols = []
+        self.factory.notifyOnConnect(protocols.append)
+        protocol1 = self.factory.buildProtocol(None)
+        protocol1.connectionMade()
+        protocol2 = self.factory.buildProtocol(None)
+        protocol2.connectionMade()
+        self.assertEqual([protocol1, protocol2], protocols)
+
+    def test_get_remote_object(self):
+        """
+        The C{getRemoteObject} method returns a deferred firing with a
+        connected L{RemoteBroker}.
+        """
+        deferred = self.factory.getRemoteObject()
+        protocol = self.factory.buildProtocol(None)
+        protocol.connectionMade()
+        self.assertIsInstance(self.successResultOf(deferred), RemoteObject)
+
+    def test_get_remote_object_failure(self):
+        """
+        If the factory fails to establish a connection the deferreds returned
+        by C{getRemoteObject} will fail.
+        """
+        deferred = self.factory.getRemoteObject()
+        self.factory.continueTrying = False  # Don't retry
+        self.factory.clientConnectionFailed(None, Failure(ConnectError()))
+        self.failureResultOf(deferred).trap(ConnectError)
 
     def test_client_connection_failed(self):
         """
         The L{MethodCallClientFactory} keeps trying to connect if maxRetries
         is not reached.
         """
-        # This is sub-optimal but the ReconnectingFactory in Hardy's Twisted
-        # doesn't support task.Clock
-        self.factory.retry = self.mocker.mock()
-        self.factory.retry(KWARGS)
-        self.mocker.replay()
+        class FakeConnector(object):
+            called = False
+
+            def connect(self):
+                self.called = True
+
+        connector = FakeConnector()
         self.assertEqual(self.factory.retries, 0)
-        self.factory.clientConnectionFailed(None, None)
-
-    def test_client_connection_failed_with_max_retries_reached(self):
-        """
-        The L{MethodCallClientFactory} stops trying to connect if maxRetries
-        is reached.
-        """
-        callback = lambda protocol: 1 / 0
-        errback = self.mocker.mock()
-        errback("failure")
-        self.mocker.replay()
-
-        self.factory.add_notifier(callback, errback)
-        self.factory.maxRetries = 1
-        self.factory.retries = self.factory.maxRetries
-        self.factory.clientConnectionFailed(object(), "failure")
-        self.clock.advance(0)
+        self.factory.clientConnectionFailed(connector, None)
+        self.assertEqual(self.factory.retries, 1)
+        self.clock.advance(5)
+        self.assertTrue(connector.called)
 
 
 class RemoteObjectConnectorTest(LandscapeTest):
@@ -605,9 +617,7 @@
         self.server_factory.protocol = lambda: (
             AMP(locator=MethodCallReceiver(Words(reactor), METHODS)))
         self.port = reactor.listenUNIX(self.socket, self.server_factory)
-        self.connector = RemoteWordsConnector(reactor, self.socket,
-                                              retry_on_reconnect=True,
-                                              timeout=0.7)
+        self.connector = RemoteWordsConnector(reactor, self.socket)
 
         def set_remote(words):
             self.words = words
@@ -664,7 +674,6 @@
         """
         self.connector.disconnect()
         self.connector.disconnect()
-        self.assertIs(self.connector._remote, None)
 
     def test_disconnect_without_connect(self):
         """
@@ -746,8 +755,9 @@
         """
         self.words._sender.protocol.transport.loseConnection()
         self.port.stopListening()
+        real_handle_connect = self.words._handle_connect
 
-        def handle_reconnect(protocol):
+        def handle_connect(protocol):
             # In this precise moment we have a newly connected protocol
             self.words._sender.protocol = protocol
 
@@ -759,8 +769,8 @@
             reactor.callLater(0, self.words._retry)
 
             # Restore the real handler and start listening again very soon
-            self.connector._factory.remove_notifier(handle_reconnect)
-            self.connector._factory.add_notifier(self.words._handle_reconnect)
+            self.connector._factory.dontNotifyOnConnect(handle_connect)
+            self.connector._factory.notifyOnConnect(real_handle_connect)
             reactor.callLater(0.2, restart_listening)
 
         def restart_listening():
@@ -770,8 +780,8 @@
             self.assertEqual(str(error), "Forbidden method 'secret'")
 
         # Use our own reconnect handler
-        self.connector._factory.remove_notifier(self.words._handle_reconnect)
-        self.connector._factory.add_notifier(handle_reconnect)
+        self.connector._factory.dontNotifyOnConnect(real_handle_connect)
+        self.connector._factory.notifyOnConnect(handle_connect)
 
         reactor.callLater(0.2, restart_listening)
         result = self.words.secret()
@@ -828,7 +838,7 @@
             return result
 
         reactor.callLater(0.1, restart_listening)
-        self.words._retry_on_reconnect = False
+        self.words._factory.retryOnReconnect = False
         result = self.words.empty()
         self.assertFailure(result, ConnectionDone)
         reconnected = Deferred()