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

[Free Ekanayaka]

This is another branch in the a series that will make most client tests synchronous. In detail:

- Add a landscape.amp.ComponentPublisher class, which is a convenience for Landscape
  client code around the generic low level landscape.lib.amp machinery. It basically
  supports publishing a Python object over a Unix socket using AMP. It is the
  server-side equivalent of RemoteComponentConnector (see next point).

- Make landscape.amp.RemoteComponentConnector not inherit from
  landscape.lib.RemoteObjectConnector. The latter should be eventually dropped, as it's
  transport-specific (it hard-codes using Unix socket), while the machinery in
  landscape.lib.amp should be transport-agnostic, providing plain Twisted factories
  and protocols. It is a sort of replace-inheritance-with-composition change.

- Create sub-classes of ComponentPublisher for each Landscape service that publishes an
  object, this replaces the low-level sub-classing of protocols that we add before.

- Add a MethodCallServerFactory class, which is a little convenience around AMP, mainly
  usedfor instantiating an AMP protocol with the appropriate "locator" parameter, that
  will responde to L{MethodCall} commands send by clients.

This is another branch in the a series that will make most client tests synchronous. In detail:

- Add a landscape.amp.ComponentPublisher class, which is a convenience for Landscape
  client code around the generic low level landscape.lib.amp machinery. It basically
  supports publishing a Python object over a Unix socket using AMP. It is the
  server-side equivalent of RemoteComponentConnector (see next point).

- Make landscape.amp.RemoteComponentConnector not inherit from
  landscape.lib.RemoteObjectConnector. The latter should be eventually dropped, as it's
  transport-specific (it hard-codes using Unix socket), while the machinery in
  landscape.lib.amp should be transport-agnostic, providing plain Twisted factories
  and protocols. It is a sort of replace-inheritance-with-composition change.

- Create sub-classes of ComponentPublisher for each Landscape service that publishes an
  object, this replaces the low-level sub-classing of protocols that we add before.

- Add a MethodCallServerFactory class, which is a little convenience around AMP, mainly
  usedfor instantiating an AMP protocol with the appropriate "locator" parameter, that
  will responde to L{MethodCall} commands send by clients.

There are a few other little things here and there, if you're unsure about them, feel free to ask me.

[Geoff Teale (tealeg)]

+1 Looking good

review: Approve

[Christopher Armstrong (radix)]

I'm taking this opportunity to generally get a better understanding of the AMP code, so there are some general comments here not directly related to your change.

[1] (non-branch) RemoteObject.__init__ docstring is wrong; it documents three parameters that don't actually exist and doesn't document the one that does.

[2] MethodCallServerFactory needs a docstring.

[3] MethodCallServerFactory.__init__'s docstring is wrong. It documents a reactor which isn't accepted but doesn't document the "methods" parameter.

[4] Why didn't you delete RemoteObjectConnector in this branch? I don't see any users of it outside of tests now. I'm eager to reduce the number of classes in the landscape/amp.py and landscape/lib/amp.py code :-)

[5] Do we usually put @params in class docstrings instead of their __init__? The docstring of RemoteComponentConnector seems strange.

[6] I think I would rename ComponentPublisher to UNIXComponentPublisher to more strongly indicate its purpose (binding the component to the unix socket).

[7] (non-branch) I don't really like the subclasses of ComponentPublisher, since all they do is declare which methods are acceptable. It seems like it'd be a lot nicer if the published-object could declare which methods are remote methods, perhaps through a decorator. Then, instead of having a e.g. BrokerServerPublisher, we would just instantiate a ComponentPublisher with the BrokerServer instance.

It's a good refactor. With the docstring comments addressed and a response to [4], I'm happy with merging.

review: Approve

[Free Ekanayaka (free.ekanayaka)]

[1]

Yeah, I noticed that, and fixed it in a later amp-cleanup-N branch. Sorry if changes are a bit scattered, but I'm trying to keep diffs small. To have a feel of what it will finally look like, please see:

lp:~free.ekanayaka/landscape-client/amp-cleanup-9

If you don't mind I'll keep this unchanged in this branch, to avoid conflicts later on.

[2]

Fixed.

[3]

Fixed.

[4]

You're right, I didn't delete it simply because it's used in a number of tests in landscape/lib/tests/test_amp.py. Those tests are not great because they use the reactor and real Unix sockets, however I'd like to keep them as "functional" tests because the guard against changes in the underlying Twisted machinery that wouldn't be necessarily caught by unit-tests. Note that all the logic that we have in lib/amp.py is now properly unit-tested in a synchronous way, by mocking the transport, but as said some functional tests exercising both basic stuff and important corner cases (e.g. disconnections) make me feel more confident. In the last amp-cleanup-9 branch that I mention I'm rewriting all these functional tests that make use of RemoteObjectConnector to simply use MethodCallServerFactory and MethodCallClientFactory (with reactor.listenUNIX/connectUNIX), and at that point I drop RemoteObjectConnector. So please have a look at that branch to see the final result.

[5]

This something I never quite understood myself. In many cases we do put @params in the class doscring for constructors (i.e. __init__), but in other we don't. We should probably set a rule. I think putting them directly under __init__ makes more sense.

[6]

Not quite sure, that would make sense if ComponentPublisher was a generic library, maybe with TCPComponentPublisher and SSHComponentPublisher equivalents. But in this case it's actually a way to abstract away those details, client code doesn't need to know about the internals (e.g. UNIX sockets), it just wants to get back a working RemoteObject. My 2 cents, but I'm open to change it if you really find it should.

[7]

Yeah, sub-classing is not that great, however I didn't want to to couple the published-object with the publisher (after all the published object doesn't need to know it's published). Perhaps an alternative solution would be to have convenience functions that replace those sub-classes. I'll try to think to some way to drop those sub-classes.

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

664. By Free Ekanayaka on 2013-04-29

Fix docstrings

=== modified file 'landscape/amp.py'
--- landscape/amp.py	2013-04-19 08:26:00 +0000
+++ landscape/amp.py	2013-04-29 11:43:28 +0000
@@ -14,7 +14,8 @@
 import logging
 
 from landscape.lib.amp import (
-    MethodCallProtocol, MethodCallFactory, RemoteObjectConnector)
+    MethodCallProtocol, MethodCallFactory, MethodCallServerFactory,
+    RemoteObject)
 
 
 class ComponentProtocol(MethodCallProtocol):
@@ -34,30 +35,64 @@
     initialDelay = 0.05
 
 
-class RemoteComponentConnector(RemoteObjectConnector):
+class ComponentPublisher(object):
+    """Publish a Landscape client component using a UNIX socket.
+
+    Other Landscape client processes can then connect to the socket and invoke
+    methods on the component remotely, using L{MethodCall} commands.
+
+    @param component: The component to publish. It can be any Python object
+        implementing the methods listed in the C{methods} class variable.
+    @param reactor: The L{TwistedReactor} used to listen to the socket.
+    @param config: The L{Configuration} object used to build the socket path.
+    """
+
+    methods = ("ping", "exit")
+
+    def __init__(self, component, reactor, config):
+        self._reactor = reactor
+        self._config = config
+        self._component = component
+        self._port = None
+
+    def start(self):
+        """Start accepting connections."""
+        factory = MethodCallServerFactory(self._component, self.methods)
+        socket_path = _get_socket_path(self._component, self._config)
+        self._port = self._reactor.listen_unix(socket_path, factory)
+
+    def stop(self):
+        """Stop accepting connections."""
+        return self._port.stopListening()
+
+
+class RemoteComponentConnector(object):
     """Utility superclass for creating connections with a Landscape component.
 
     @cvar component: The class of the component to connect to, it is expected
         to define a C{name} class attribute, which will be used to find out
         the socket to use. It must be defined by sub-classes.
+    @cvar factory: The factory class to use for building protocols.
+    @cvar remote: The L{RemoteObject} class or sub-class used for building
+        remote objects.
 
     @param reactor: A L{TwistedReactor} object.
     @param config: A L{LandscapeConfiguration}.
-    @param args: Positional arguments for protocol factory constructor.
-    @param kwargs: Keyword arguments for protocol factory constructor.
+    @param retry_on_reconnect: If C{True} the remote object built by this
+        connector will retry L{MethodCall}s that failed due to lost
+        connections.
 
     @see: L{MethodCallClientFactory}.
     """
-
+    component = None  # Must be defined by sub-classes
     factory = ComponentProtocolFactory
+    remote = RemoteObject
 
     def __init__(self, reactor, config, retry_on_reconnect=False):
-        self._twisted_reactor = reactor
+        self._reactor = reactor
+        self._config = config
         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)
+        self._connector = None
 
     def connect(self, max_retries=None, factor=None, quiet=False):
         """Connect to the remote Landscape component.
@@ -73,28 +108,44 @@
             result in a faster reconnection attempts pace.
         @param quiet: A boolean indicating whether to log errors.
         """
+        reactor = self._reactor._reactor
+        factory = self.factory(reactor)
+        factory.retryOnReconnect = self._retry_on_reconnect
+        factory.remote = self.remote
 
         def fire_reconnect(ignored):
-            self._twisted_reactor.fire("%s-reconnect" %
-                                       self.component.name)
+            self._reactor.fire("%s-reconnect" % self.component.name)
 
         def connected(remote):
-            # 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)
+            factory.notifyOnConnect(fire_reconnect)
             return remote
 
         def log_error(failure):
             logging.error("Error while connecting to %s", self.component.name)
             return failure
 
-        result = super(RemoteComponentConnector, self).connect(
-            max_retries=max_retries, factor=factor)
+        factory.maxRetries = max_retries
+        if factor:
+            factory.factor = factor
+        socket_path = _get_socket_path(self.component, self._config)
+        self._connector = reactor.connectUNIX(socket_path, factory)
+        deferred = factory.getRemoteObject()
+
         if not quiet:
-            result.addErrback(log_error)
-        result.addCallback(connected)
-        return result
+            deferred.addErrback(log_error)
+
+        return deferred.addCallback(connected)
+
+    def disconnect(self):
+        """Disconnect the L{RemoteObject} that we have created."""
+        if self._connector is not None:
+            factory = self._connector.factory
+            factory.stopTrying()
+            # XXX we should be using self._connector.disconnect() here
+            remote = factory._remote
+            if remote:
+                if remote._sender.protocol.transport:
+                    remote._sender.protocol.transport.loseConnection()
 
 
 class RemoteComponentsRegistry(object):
@@ -121,3 +172,7 @@
             that can be used to connect to a certain component.
         """
         cls._by_name[connector_class.component.name] = connector_class
+
+
+def _get_socket_path(component, config):
+    return os.path.join(config.sockets_path, component.name + ".sock")

=== modified file 'landscape/broker/amp.py'
--- landscape/broker/amp.py	2013-04-19 08:26:00 +0000
+++ landscape/broker/amp.py	2013-04-29 11:43:28 +0000
@@ -2,36 +2,31 @@
 
 from landscape.lib.amp import RemoteObject, MethodCallArgument
 from landscape.amp import (
-    ComponentProtocol, ComponentProtocolFactory, RemoteComponentConnector,
-    RemoteComponentsRegistry)
+    RemoteComponentConnector, RemoteComponentsRegistry, ComponentPublisher)
 from landscape.broker.server import BrokerServer
 from landscape.broker.client import BrokerClient
 from landscape.monitor.monitor import Monitor
 from landscape.manager.manager import Manager
 
 
-class BrokerServerProtocol(ComponentProtocol):
+class BrokerServerPublisher(ComponentPublisher):
     """
     Communication protocol between the broker server and its clients.
     """
-    methods = (ComponentProtocol.methods +
-               ["fire_event",
-                "get_accepted_message_types",
-                "get_server_uuid",
-                "is_message_pending",
-                "register",
-                "register_client",
-                "register_client_accepted_message_type",
-                "reload_configuration",
-                "send_message",
-                "stop_clients",
-                "listen_events",
-                "stop_exchanger"])
-
-
-class BrokerServerProtocolFactory(ComponentProtocolFactory):
-
-    protocol = BrokerServerProtocol
+
+    methods = ComponentPublisher.methods + (
+        "fire_event",
+        "get_accepted_message_types",
+        "get_server_uuid",
+        "is_message_pending",
+        "register",
+        "register_client",
+        "register_client_accepted_message_type",
+        "reload_configuration",
+        "send_message",
+        "stop_clients",
+        "listen_events",
+        "stop_exchanger")
 
 
 class RemoteBroker(RemoteObject):
@@ -64,7 +59,6 @@
     def __init__(self, exchanger, message_store, broker_server):
         self.exchanger = exchanger
         self.message_store = message_store
-        self.protocol = BrokerServerProtocol()
         self.broker_server = broker_server
 
     def __getattr__(self, name):
@@ -73,7 +67,7 @@
         that they're encodable with AMP.
         """
         original = getattr(self.broker_server, name, None)
-        if (name in BrokerServerProtocol.methods
+        if (name in BrokerServerPublisher.methods
             and original is not None
             and callable(original)):
             def method(*args, **kwargs):
@@ -92,16 +86,13 @@
         return succeed(None)
 
 
-class BrokerClientProtocol(ComponentProtocol):
-    """Communication protocol between a client and the broker."""
-
-    methods = (ComponentProtocol.methods + ["fire_event", "message"])
-
-
-class BrokerClientProtocolFactory(ComponentProtocolFactory):
-
-    protocol = BrokerClientProtocol
-    remote = RemoteBroker
+class BrokerClientPublisher(ComponentPublisher):
+    """
+    Communication protocol between the broker server and its clients.
+    """
+    methods = ComponentPublisher.methods + (
+        "fire_event",
+        "message")
 
 
 class RemoteClient(RemoteObject):
@@ -111,7 +102,6 @@
 class RemoteBrokerConnector(RemoteComponentConnector):
     """Helper to create connections with the L{BrokerServer}."""
 
-    factory = BrokerClientProtocolFactory
     remote = RemoteBroker
     component = BrokerServer
 
@@ -119,7 +109,6 @@
 class RemoteClientConnector(RemoteComponentConnector):
     """Helper to create connections with the L{BrokerServer}."""
 
-    factory = BrokerServerProtocolFactory
     remote = RemoteClient
     component = BrokerClient
 

=== modified file 'landscape/broker/server.py'
--- landscape/broker/server.py	2013-03-31 22:44:46 +0000
+++ landscape/broker/server.py	2013-04-29 11:43:28 +0000
@@ -66,7 +66,7 @@
         """Register a broker client called C{name}.
 
         Various broker clients interact with the broker server, such as the
-        monitor for example, using the L{BrokerServerProtocol} for performing
+        monitor for example, using the L{BrokerServerConnector} for performing
         remote method calls on the L{BrokerServer}.
 
         They establish connectivity with the broker by connecting and

=== modified file 'landscape/broker/service.py'
--- landscape/broker/service.py	2013-03-31 22:44:46 +0000
+++ landscape/broker/service.py	2013-04-29 11:43:28 +0000
@@ -11,7 +11,7 @@
 from landscape.broker.ping import Pinger
 from landscape.broker.store import get_default_message_store
 from landscape.broker.server import BrokerServer
-from landscape.broker.amp import BrokerServerProtocolFactory
+from landscape.broker.amp import BrokerServerPublisher
 
 
 class BrokerService(LandscapeService):
@@ -65,21 +65,24 @@
         self.broker = BrokerServer(self.config, self.reactor, self.exchanger,
                                    self.registration, self.message_store,
                                    self.pinger)
-        self.factory = BrokerServerProtocolFactory(object=self.broker)
+        self.publisher = BrokerServerPublisher(self.broker, self.reactor,
+                                               self.config)
 
     def startService(self):
         """Start the broker.
 
         Create a L{BrokerServer} listening on C{broker_socket_path} for clients
-        connecting with the L{BrokerClientProtocol}, and start the
+        connecting with the L{BrokerServerConnector}, and start the
         L{MessageExchange} and L{Pinger} services.
         """
         super(BrokerService, self).startService()
+        self.publisher.start()
         self.exchanger.start()
         self.pinger.start()
 
     def stopService(self):
         """Stop the broker."""
+        self.publisher.stop()
         self.exchanger.stop()
         self.pinger.stop()
         super(BrokerService, self).stopService()

=== modified file 'landscape/broker/tests/helpers.py'
--- landscape/broker/tests/helpers.py	2013-04-12 14:14:47 +0000
+++ landscape/broker/tests/helpers.py	2013-04-29 11:43:28 +0000
@@ -20,8 +20,7 @@
 from landscape.broker.config import BrokerConfiguration
 from landscape.broker.server import BrokerServer
 from landscape.broker.amp import (
-    BrokerServerProtocolFactory, BrokerClientProtocolFactory,
-    RemoteBrokerConnector)
+    BrokerServerPublisher, BrokerClientPublisher, RemoteBrokerConnector)
 from landscape.broker.client import BrokerClient
 
 
@@ -176,13 +175,14 @@
     def set_up(self, test_case):
         super(RemoteBrokerHelper, self).set_up(test_case)
 
-        factory = BrokerServerProtocolFactory(object=test_case.broker)
-        socket = os.path.join(test_case.config.sockets_path,
-                              BrokerServer.name + ".sock")
-        self._port = test_case.reactor.listen_unix(socket, factory)
+        self._publisher = BrokerServerPublisher(test_case.broker,
+                                                test_case.reactor,
+                                                test_case.config)
         self._connector = RemoteBrokerConnector(test_case.reactor,
                                                 test_case.config)
 
+        self._publisher.start()
+
         def set_remote(remote):
             test_case.remote = remote
             return remote
@@ -192,7 +192,7 @@
 
     def tear_down(self, test_case):
         self._connector.disconnect()
-        self._port.stopListening()
+        self._publisher.stop()
         super(RemoteBrokerHelper, self).tear_down(test_case)
 
 
@@ -246,11 +246,10 @@
 
         def listen(ignored):
 
-            factory = BrokerClientProtocolFactory(object=test_case.client)
-            socket = os.path.join(test_case.config.sockets_path,
-                                  test_case.client.name + ".sock")
-            self._client_port = test_case.client_reactor.listen_unix(socket,
-                                                                     factory)
+            self._client_publisher = BrokerClientPublisher(test_case.client,
+                                                           test_case.reactor,
+                                                           test_case.config)
+            self._client_publisher.start()
             result = test_case.remote.register_client("client")
             return result.addCallback(set_remote_client)
 
@@ -259,5 +258,5 @@
 
     def tear_down(self, test_case):
         self._client_connector.disconnect()
-        self._client_port.stopListening()
+        self._client_publisher.stop()
         super(RemoteClientHelper, self).tear_down(test_case)

=== modified file 'landscape/broker/tests/test_service.py'
--- landscape/broker/tests/test_service.py	2013-03-31 22:44:46 +0000
+++ landscape/broker/tests/test_service.py	2013-04-29 11:43:28 +0000
@@ -80,5 +80,4 @@
         connected.addCallback(lambda remote: remote.get_server_uuid())
         connected.addCallback(lambda x: connector.disconnect())
         connected.addCallback(lambda x: self.service.stopService())
-        connected.addCallback(lambda x: self.service.port.stopListening())
         return connected

=== modified file 'landscape/lib/amp.py'
--- landscape/lib/amp.py	2013-04-23 13:50:29 +0000
+++ landscape/lib/amp.py	2013-04-29 11:43:28 +0000
@@ -46,7 +46,7 @@
 """
 
 from twisted.internet.defer import Deferred, maybeDeferred, succeed
-from twisted.internet.protocol import ReconnectingClientFactory
+from twisted.internet.protocol import ServerFactory, ReconnectingClientFactory
 from twisted.python.failure import Failure
 
 from twisted.protocols.amp import (
@@ -423,6 +423,28 @@
                               deferred=deferred, call=call)
 
 
+class MethodCallServerFactory(ServerFactory):
+    """Expose a Python object using L{MethodCall} commands over C{AMP}."""
+
+    protocol = AMP
+
+    def __init__(self, object, methods):
+        """
+        @param object: The object exposed by the L{MethodCallProtocol}s
+            instances created by this factory.
+        @param methods: A list of the names of the methods that remote peers
+            are allowed to call on the C{object} that we publish.
+        """
+        self.object = object
+        self.methods = methods
+
+    def buildProtocol(self, addr):
+        locator = MethodCallReceiver(self.object, self.methods)
+        protocol = self.protocol(locator=locator)
+        protocol.factory = self
+        return protocol
+
+
 class MethodCallClientFactory(ReconnectingClientFactory):
     """
     Factory for L{MethodCallProtocol}s exposing an object or connecting to
@@ -550,6 +572,7 @@
         self._socket_path = socket_path
         self._reactor = reactor
         self._factory = None
+        self._connector = None
 
     def connect(self, max_retries=None, factor=None):
         """Connect to a remote object exposed by a L{MethodCallProtocol}.
@@ -563,18 +586,20 @@
             delay between subsequent retries should increase. Smaller values
             result in a faster reconnection attempts pace.
         """
-        self._factory = self.factory(reactor=self._reactor)
-        self._factory.maxRetries = max_retries
+        factory = self.factory(reactor=self._reactor)
+        factory.maxRetries = max_retries
         if factor:
-            self._factory.factor = factor
-        self._reactor.connectUNIX(self._socket_path, self._factory)
-        return self._factory.getRemoteObject()
+            factory.factor = factor
+        self._connector = self._reactor.connectUNIX(self._socket_path, factory)
+        return factory.getRemoteObject()
 
     def disconnect(self):
         """Disconnect the L{RemoteObject} that we have created."""
-        if self._factory:
-            self._factory.stopTrying()
-            remote = self._factory._remote
+        if self._connector is not None:
+            factory = self._connector.factory
+            factory.stopTrying()
+            # XXX we should be using self._connector.disconnect() here
+            remote = 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-23 13:41:26 +0000
+++ landscape/lib/tests/test_amp.py	2013-04-29 11:43:28 +0000
@@ -1,5 +1,4 @@
 from twisted.internet import reactor
-from twisted.internet.protocol import Factory
 from twisted.internet.defer import Deferred, DeferredList
 from twisted.internet.error import ConnectionDone, ConnectError
 from twisted.internet.task import Clock
@@ -7,8 +6,9 @@
 from twisted.python.failure import Failure
 
 from landscape.lib.amp import (
-    MethodCallError, MethodCallProtocol, MethodCallClientFactory, RemoteObject,
-    RemoteObjectConnector, MethodCallReceiver, MethodCallSender)
+    MethodCallError, MethodCallProtocol, MethodCallServerFactory,
+    MethodCallClientFactory, RemoteObject, RemoteObjectConnector,
+    MethodCallReceiver, MethodCallSender)
 from landscape.tests.helpers import LandscapeTest
 
 
@@ -613,9 +613,7 @@
     def setUp(self):
         super(RemoteObjectConnectorTest, self).setUp()
         self.socket = self.mktemp()
-        self.server_factory = Factory()
-        self.server_factory.protocol = lambda: (
-            AMP(locator=MethodCallReceiver(Words(reactor), METHODS)))
+        self.server_factory = MethodCallServerFactory(Words(reactor), METHODS)
         self.port = reactor.listenUNIX(self.socket, self.server_factory)
         self.connector = RemoteWordsConnector(reactor, self.socket)
 
@@ -662,7 +660,7 @@
         self.connector.disconnect()
 
         def assert_factor(ignored):
-            self.assertEqual(self.connector._factory.factor, 1.0)
+            self.assertEqual(self.connector._connector.factory.factor, 1.0)
 
         result = self.connector.connect(factor=1.0)
         return result.addCallback(assert_factor)
@@ -769,8 +767,8 @@
             reactor.callLater(0, self.words._retry)
 
             # Restore the real handler and start listening again very soon
-            self.connector._factory.dontNotifyOnConnect(handle_connect)
-            self.connector._factory.notifyOnConnect(real_handle_connect)
+            factory.dontNotifyOnConnect(handle_connect)
+            factory.notifyOnConnect(real_handle_connect)
             reactor.callLater(0.2, restart_listening)
 
         def restart_listening():
@@ -780,8 +778,9 @@
             self.assertEqual(str(error), "Forbidden method 'secret'")
 
         # Use our own reconnect handler
-        self.connector._factory.dontNotifyOnConnect(real_handle_connect)
-        self.connector._factory.notifyOnConnect(handle_connect)
+        factory = self.connector._connector.factory
+        factory.dontNotifyOnConnect(real_handle_connect)
+        factory.notifyOnConnect(handle_connect)
 
         reactor.callLater(0.2, restart_listening)
         result = self.words.secret()

=== modified file 'landscape/manager/service.py'
--- landscape/manager/service.py	2010-04-12 12:46:07 +0000
+++ landscape/manager/service.py	2013-04-29 11:43:28 +0000
@@ -2,8 +2,7 @@
 
 from landscape.service import LandscapeService, run_landscape_service
 from landscape.manager.config import ManagerConfiguration
-from landscape.broker.amp import (
-    BrokerClientProtocolFactory, RemoteBrokerConnector)
+from landscape.broker.amp import BrokerClientPublisher, RemoteBrokerConnector
 from landscape.manager.manager import Manager
 
 
@@ -19,7 +18,8 @@
         super(ManagerService, self).__init__(config)
         self.plugins = self.get_plugins()
         self.manager = Manager(self.reactor, self.config)
-        self.factory = BrokerClientProtocolFactory(object=self.manager)
+        self.publisher = BrokerClientPublisher(self.manager, self.reactor,
+                                               self.config)
 
     def get_plugins(self):
         """Return instances of all the plugins enabled in the configuration."""
@@ -37,6 +37,7 @@
           - Add all configured plugins, that will in turn register themselves.
         """
         super(ManagerService, self).startService()
+        self.publisher.start()
 
         def start_plugins(broker):
             self.broker = broker
@@ -52,6 +53,7 @@
     def stopService(self):
         """Stop the manager and close the connection with the broker."""
         self.connector.disconnect()
+        self.publisher.stop()
         super(ManagerService, self).stopService()
 
 

=== modified file 'landscape/manager/tests/test_service.py'
--- landscape/manager/tests/test_service.py	2011-12-02 08:35:25 +0000
+++ landscape/manager/tests/test_service.py	2013-04-29 11:43:28 +0000
@@ -48,9 +48,7 @@
             [connector] = self.broker_service.broker.get_connectors()
             connector.disconnect()
             self.service.stopService()
-            self.service.port.stopListening()
             self.broker_service.stopService()
-            self.broker_service.port.stopListening()
 
         def assert_broker_connection(ignored):
             self.assertEqual(len(self.broker_service.broker.get_clients()), 1)

=== modified file 'landscape/monitor/service.py'
--- landscape/monitor/service.py	2010-03-19 08:29:46 +0000
+++ landscape/monitor/service.py	2013-04-29 11:43:28 +0000
@@ -7,8 +7,7 @@
 from landscape.service import LandscapeService, run_landscape_service
 from landscape.monitor.config import MonitorConfiguration
 from landscape.monitor.monitor import Monitor
-from landscape.broker.amp import (
-    BrokerClientProtocolFactory, RemoteBrokerConnector)
+from landscape.broker.amp import BrokerClientPublisher, RemoteBrokerConnector
 
 
 class MonitorService(LandscapeService):
@@ -26,7 +25,8 @@
         self.plugins = self.get_plugins()
         self.monitor = Monitor(self.reactor, self.config, self.persist,
                                persist_filename=self.persist_filename)
-        self.factory = BrokerClientProtocolFactory(object=self.monitor)
+        self.publisher = BrokerClientPublisher(self.monitor, self.reactor,
+                                               self.config)
 
     def get_plugins(self):
         return [namedClass("landscape.monitor.%s.%s"
@@ -36,6 +36,7 @@
     def startService(self):
         """Start the monitor."""
         super(MonitorService, self).startService()
+        self.publisher.start()
 
         def start_plugins(broker):
             self.broker = broker
@@ -54,6 +55,7 @@
         The monitor is flushed to ensure that things like persist databases
         get saved to disk.
         """
+        self.publisher.stop()
         self.monitor.flush()
         self.connector.disconnect()
         super(MonitorService, self).stopService()

=== modified file 'landscape/monitor/tests/test_service.py'
--- landscape/monitor/tests/test_service.py	2012-03-05 14:11:42 +0000
+++ landscape/monitor/tests/test_service.py	2013-04-29 11:43:28 +0000
@@ -58,9 +58,7 @@
             [connector] = self.broker_service.broker.get_connectors()
             connector.disconnect()
             self.service.stopService()
-            self.service.port.stopListening()
             self.broker_service.stopService()
-            self.broker_service.port.stopListening()
 
         def assert_broker_connection(ignored):
             self.assertEqual(len(self.broker_service.broker.get_clients()), 1)
@@ -81,8 +79,7 @@
         self.service.monitor.flush()
         self.service.connector = self.mocker.mock()
         self.service.connector.disconnect()
-        self.service.port = self.mocker.mock()
-        self.service.port.stopListening()
+        self.service.publisher = self.mocker.mock()
+        self.service.publisher.stop()
         self.mocker.replay()
         self.service.stopService()
-        self.service.port.stopListening()

=== modified file 'landscape/service.py'
--- landscape/service.py	2013-02-12 16:26:34 +0000
+++ landscape/service.py	2013-04-29 11:43:28 +0000
@@ -8,7 +8,7 @@
 from landscape.log import rotate_logs
 from landscape.reactor import TwistedReactor
 from landscape.deployment import get_versioned_persist, init_logging
-from landscape.amp import ComponentProtocol
+from landscape.amp import ComponentProtocol, ComponentPublisher
 
 
 class LandscapeService(Service, object):
@@ -50,7 +50,8 @@
         Service.startService(self)
         logging.info("%s started with config %s" % (
             self.service_name.capitalize(), self.config.get_config_filename()))
-        self.port = self.reactor.listen_unix(self.socket, self.factory)
+        if hasattr(self, "factory"):
+            self.port = self.reactor.listen_unix(self.socket, self.factory)
 
     def stopService(self):
         # We don't need to call port.stopListening(), because the reactor

=== modified file 'landscape/tests/helpers.py'
--- landscape/tests/helpers.py	2013-04-15 07:32:54 +0000
+++ landscape/tests/helpers.py	2013-04-29 11:43:28 +0000
@@ -345,14 +345,6 @@
             reactor_factory = FakeReactor
             transport_factory = FakeTransport
 
-            def stopService(service):
-                # We need to explictely stop listening to the socket
-                # because the reactor would still have active selectables
-                # at the end of the test otherwise
-                if os.path.exists(service.port.port):
-                    service.port.connectionLost(None)
-                super(FakeBrokerService, service).stopService()
-
         test_case.broker_service = FakeBrokerService(config)
         test_case.remote = FakeRemoteBroker(
             test_case.broker_service.exchanger,

=== modified file 'landscape/tests/test_watchdog.py'
--- landscape/tests/test_watchdog.py	2013-02-12 16:26:34 +0000
+++ landscape/tests/test_watchdog.py	2013-04-29 11:43:28 +0000
@@ -1453,7 +1453,8 @@
 
 sys.path = %(path)r
 
-from landscape.broker.amp import BrokerServerProtocolFactory
+from landscape.lib.amp import MethodCallServerFactory
+from landscape.broker.amp import BrokerServerPublisher
 
 class StubBroker(object):
 
@@ -1464,7 +1465,8 @@
         reactor.callLater(1, reactor.stop)
 
 stub_broker = StubBroker()
-factory = BrokerServerProtocolFactory(object=stub_broker)
+methods = BrokerServerPublisher.methods
+factory = MethodCallServerFactory(stub_broker, methods)
 reactor.listenUNIX(%(socket)r, factory)
 reactor.run()
 """