diff --git a/neutron/common/rpc_compat.py b/neutron/common/rpc_compat.py index 2e95ee2bd52..bf8d40c35c6 100644 --- a/neutron/common/rpc_compat.py +++ b/neutron/common/rpc_compat.py @@ -13,8 +13,15 @@ # License for the specific language governing permissions and limitations # under the License. +from neutron.openstack.common import log as logging +from neutron.openstack.common import rpc from neutron.openstack.common.rpc import common as rpc_common +from neutron.openstack.common.rpc import dispatcher as rpc_dispatcher from neutron.openstack.common.rpc import proxy +from neutron.openstack.common import service + + +LOG = logging.getLogger(__name__) class RpcProxy(proxy.RpcProxy): @@ -35,6 +42,57 @@ class RpcCallback(object): ''' +class Service(service.Service): + """Service object for binaries running on hosts. + + A service enables rpc by listening to queues based on topic and host. + """ + def __init__(self, host, topic, manager=None, serializer=None): + super(Service, self).__init__() + self.host = host + self.topic = topic + self.serializer = serializer + if manager is None: + self.manager = self + else: + self.manager = manager + + def start(self): + super(Service, self).start() + + self.conn = rpc.create_connection(new=True) + LOG.debug("Creating Consumer connection for Service %s" % + self.topic) + + dispatcher = rpc_dispatcher.RpcDispatcher([self.manager], + self.serializer) + + # Share this same connection for these Consumers + self.conn.create_consumer(self.topic, dispatcher, fanout=False) + + node_topic = '%s.%s' % (self.topic, self.host) + self.conn.create_consumer(node_topic, dispatcher, fanout=False) + + self.conn.create_consumer(self.topic, dispatcher, fanout=True) + + # Hook to allow the manager to do other initializations after + # the rpc connection is created. + if callable(getattr(self.manager, 'initialize_service_hook', None)): + self.manager.initialize_service_hook(self) + + # Consume from all consumers in a thread + self.conn.consume_in_thread() + + def stop(self): + # Try to shut the connection down, but if we get any sort of + # errors, go ahead and ignore them.. as we're shutting down anyway + try: + self.conn.close() + except Exception: + pass + super(Service, self).stop() + + # exceptions RPCException = rpc_common.RPCException RemoteError = rpc_common.RemoteError diff --git a/neutron/service.py b/neutron/service.py index a3224545bf3..9b3073b5fb4 100644 --- a/neutron/service.py +++ b/neutron/service.py @@ -22,6 +22,7 @@ import random from oslo.config import cfg from neutron.common import config +from neutron.common import rpc_compat from neutron import context from neutron.db import api as session from neutron import manager @@ -29,7 +30,6 @@ from neutron.openstack.common import excutils from neutron.openstack.common import importutils from neutron.openstack.common import log as logging from neutron.openstack.common import loopingcall -from neutron.openstack.common.rpc import service from neutron.openstack.common import service as common_service from neutron import wsgi @@ -178,7 +178,7 @@ def _run_wsgi(app_name): return server -class Service(service.Service): +class Service(rpc_compat.Service): """Service object for binaries running on hosts. A service takes a manager and enables rpc by listening to queues based diff --git a/neutron/services/loadbalancer/agent/agent.py b/neutron/services/loadbalancer/agent/agent.py index 0ec14554d9b..7a830c63150 100644 --- a/neutron/services/loadbalancer/agent/agent.py +++ b/neutron/services/loadbalancer/agent/agent.py @@ -23,8 +23,8 @@ from oslo.config import cfg from neutron.agent.common import config from neutron.agent.linux import interface +from neutron.common import rpc_compat from neutron.common import topics -from neutron.openstack.common.rpc import service as rpc_service from neutron.openstack.common import service from neutron.services.loadbalancer.agent import agent_manager as manager @@ -37,7 +37,7 @@ OPTS = [ ] -class LbaasAgentService(rpc_service.Service): +class LbaasAgentService(rpc_compat.Service): def start(self): super(LbaasAgentService, self).start() self.tg.add_timer( diff --git a/neutron/tests/unit/services/loadbalancer/agent/test_agent.py b/neutron/tests/unit/services/loadbalancer/agent/test_agent.py index b00c5233008..955d6e1a842 100644 --- a/neutron/tests/unit/services/loadbalancer/agent/test_agent.py +++ b/neutron/tests/unit/services/loadbalancer/agent/test_agent.py @@ -27,7 +27,7 @@ from neutron.tests import base class TestLbaasService(base.BaseTestCase): def test_start(self): with mock.patch.object( - agent.rpc_service.Service, 'start' + agent.rpc_compat.Service, 'start' ) as mock_start: mgr = mock.Mock()