Merge "Set access_policy for messaging's dispatcher"
This commit is contained in:
commit
7ba3434215
|
@ -15,6 +15,7 @@
|
||||||
from oslo_config import cfg
|
from oslo_config import cfg
|
||||||
from oslo_log import log as logging
|
from oslo_log import log as logging
|
||||||
import oslo_messaging as messaging
|
import oslo_messaging as messaging
|
||||||
|
from oslo_messaging.rpc import dispatcher
|
||||||
from oslo_service import service
|
from oslo_service import service
|
||||||
|
|
||||||
from octavia.controller.queue import endpoint
|
from octavia.controller.queue import endpoint
|
||||||
|
@ -35,8 +36,10 @@ class Consumer(service.Service):
|
||||||
transport = messaging.get_transport(cfg.CONF)
|
transport = messaging.get_transport(cfg.CONF)
|
||||||
target = messaging.Target(topic=topic, server=server, fanout=False)
|
target = messaging.Target(topic=topic, server=server, fanout=False)
|
||||||
endpoints = [endpoint.Endpoint()]
|
endpoints = [endpoint.Endpoint()]
|
||||||
|
access_policy = dispatcher.DefaultRPCAccessPolicy
|
||||||
self.server = messaging.get_rpc_server(transport, target, endpoints,
|
self.server = messaging.get_rpc_server(transport, target, endpoints,
|
||||||
executor='eventlet')
|
executor='eventlet',
|
||||||
|
access_policy=access_policy)
|
||||||
LOG.info(_LI('Starting consumer...'))
|
LOG.info(_LI('Starting consumer...'))
|
||||||
self.server.start()
|
self.server.start()
|
||||||
super(Consumer, self).start()
|
super(Consumer, self).start()
|
||||||
|
|
|
@ -16,6 +16,7 @@ import mock
|
||||||
from oslo_config import cfg
|
from oslo_config import cfg
|
||||||
from oslo_config import fixture as oslo_fixture
|
from oslo_config import fixture as oslo_fixture
|
||||||
import oslo_messaging as messaging
|
import oslo_messaging as messaging
|
||||||
|
from oslo_messaging.rpc import dispatcher
|
||||||
|
|
||||||
from octavia.controller.queue import consumer
|
from octavia.controller.queue import consumer
|
||||||
from octavia.controller.queue import endpoint
|
from octavia.controller.queue import endpoint
|
||||||
|
@ -51,10 +52,12 @@ class TestConsumer(base.TestCase):
|
||||||
mock_target.assert_called_once_with(topic='foo_topic',
|
mock_target.assert_called_once_with(topic='foo_topic',
|
||||||
server='foo_host', fanout=False)
|
server='foo_host', fanout=False)
|
||||||
mock_endpoint.assert_called_once_with()
|
mock_endpoint.assert_called_once_with()
|
||||||
|
access_policy = dispatcher.DefaultRPCAccessPolicy
|
||||||
mock_rpc_server.assert_called_once_with(mock_get_transport_rv,
|
mock_rpc_server.assert_called_once_with(mock_get_transport_rv,
|
||||||
mock_target_rv,
|
mock_target_rv,
|
||||||
[mock_endpoint_rv],
|
[mock_endpoint_rv],
|
||||||
executor='eventlet')
|
executor='eventlet',
|
||||||
|
access_policy=access_policy)
|
||||||
|
|
||||||
def test_consumer_stop(self, mock_rpc_server, mock_endpoint, mock_target,
|
def test_consumer_stop(self, mock_rpc_server, mock_endpoint, mock_target,
|
||||||
mock_get_transport):
|
mock_get_transport):
|
||||||
|
|
Loading…
Reference in New Issue