2013-04-30 06:49:56 +01:00
|
|
|
|
|
|
|
# Copyright 2013 Red Hat, Inc.
|
|
|
|
#
|
|
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
|
|
# not use this file except in compliance with the License. You may obtain
|
|
|
|
# a copy of the License at
|
|
|
|
#
|
|
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
#
|
|
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
|
|
# License for the specific language governing permissions and limitations
|
|
|
|
# under the License.
|
|
|
|
|
|
|
|
import abc
|
|
|
|
|
2013-10-14 10:42:15 -04:00
|
|
|
import six
|
|
|
|
|
2013-06-15 07:43:39 +01:00
|
|
|
from oslo.messaging import exceptions
|
2013-06-14 14:45:40 +01:00
|
|
|
|
|
|
|
|
|
|
|
class TransportDriverError(exceptions.MessagingException):
|
|
|
|
"""Base class for transport driver specific exceptions."""
|
|
|
|
|
2013-04-30 06:49:56 +01:00
|
|
|
|
2013-10-14 10:42:15 -04:00
|
|
|
@six.add_metaclass(abc.ABCMeta)
|
Add IncomingMessage abstraction
When an executor polls for a message, the driver needs to return the
request context and the message.
Later, if the executor wishes to send a reply, it needs to pass back
a handle which identifies the message so that the transport can deliver
the reply.
In the current interface, we're just passing back the context and
message to the transport so this presumes that the transport would
attach whatever it needs to one of these objects.
In the AMQP drivers in openstack.common.rpc, we set attributes like
reply_q and msg_id on the returned context. However, it would be much
better if we never touched the user-supplied context and, instead, had
some other way to pass this info to the executor and then have it passed
back to the transport.
To achieve that, add an IncomingMessage abstract class which wraps the
context and message and has a reply() method. That way, transports can
subclass this class, add whatver attributes they want and implement a
reply method.
To repeat what this means ... we can allow users of the API to use
read-only mapping objects as a context, rather than requiring it to be
an object we can set arbitrary attributes on.
2013-05-13 11:33:41 +01:00
|
|
|
class IncomingMessage(object):
|
2013-04-30 06:49:56 +01:00
|
|
|
|
Add IncomingMessage abstraction
When an executor polls for a message, the driver needs to return the
request context and the message.
Later, if the executor wishes to send a reply, it needs to pass back
a handle which identifies the message so that the transport can deliver
the reply.
In the current interface, we're just passing back the context and
message to the transport so this presumes that the transport would
attach whatever it needs to one of these objects.
In the AMQP drivers in openstack.common.rpc, we set attributes like
reply_q and msg_id on the returned context. However, it would be much
better if we never touched the user-supplied context and, instead, had
some other way to pass this info to the executor and then have it passed
back to the transport.
To achieve that, add an IncomingMessage abstract class which wraps the
context and message and has a reply() method. That way, transports can
subclass this class, add whatver attributes they want and implement a
reply method.
To repeat what this means ... we can allow users of the API to use
read-only mapping objects as a context, rather than requiring it to be
an object we can set arbitrary attributes on.
2013-05-13 11:33:41 +01:00
|
|
|
def __init__(self, listener, ctxt, message):
|
|
|
|
self.conf = listener.conf
|
|
|
|
self.listener = listener
|
|
|
|
self.ctxt = ctxt
|
|
|
|
self.message = message
|
2013-04-30 06:49:56 +01:00
|
|
|
|
|
|
|
@abc.abstractmethod
|
2013-08-07 09:00:44 +01:00
|
|
|
def reply(self, reply=None, failure=None, log_failure=True):
|
2013-06-15 19:13:45 +01:00
|
|
|
"Send a reply or failure back to the client."
|
2013-04-30 06:49:56 +01:00
|
|
|
|
Add IncomingMessage abstraction
When an executor polls for a message, the driver needs to return the
request context and the message.
Later, if the executor wishes to send a reply, it needs to pass back
a handle which identifies the message so that the transport can deliver
the reply.
In the current interface, we're just passing back the context and
message to the transport so this presumes that the transport would
attach whatever it needs to one of these objects.
In the AMQP drivers in openstack.common.rpc, we set attributes like
reply_q and msg_id on the returned context. However, it would be much
better if we never touched the user-supplied context and, instead, had
some other way to pass this info to the executor and then have it passed
back to the transport.
To achieve that, add an IncomingMessage abstract class which wraps the
context and message and has a reply() method. That way, transports can
subclass this class, add whatver attributes they want and implement a
reply method.
To repeat what this means ... we can allow users of the API to use
read-only mapping objects as a context, rather than requiring it to be
an object we can set arbitrary attributes on.
2013-05-13 11:33:41 +01:00
|
|
|
|
2013-10-14 10:42:15 -04:00
|
|
|
@six.add_metaclass(abc.ABCMeta)
|
Add IncomingMessage abstraction
When an executor polls for a message, the driver needs to return the
request context and the message.
Later, if the executor wishes to send a reply, it needs to pass back
a handle which identifies the message so that the transport can deliver
the reply.
In the current interface, we're just passing back the context and
message to the transport so this presumes that the transport would
attach whatever it needs to one of these objects.
In the AMQP drivers in openstack.common.rpc, we set attributes like
reply_q and msg_id on the returned context. However, it would be much
better if we never touched the user-supplied context and, instead, had
some other way to pass this info to the executor and then have it passed
back to the transport.
To achieve that, add an IncomingMessage abstract class which wraps the
context and message and has a reply() method. That way, transports can
subclass this class, add whatver attributes they want and implement a
reply method.
To repeat what this means ... we can allow users of the API to use
read-only mapping objects as a context, rather than requiring it to be
an object we can set arbitrary attributes on.
2013-05-13 11:33:41 +01:00
|
|
|
class Listener(object):
|
|
|
|
|
|
|
|
def __init__(self, driver, target):
|
|
|
|
self.conf = driver.conf
|
|
|
|
self.driver = driver
|
|
|
|
self.target = target
|
|
|
|
|
2013-05-11 13:19:59 +01:00
|
|
|
@abc.abstractmethod
|
Add IncomingMessage abstraction
When an executor polls for a message, the driver needs to return the
request context and the message.
Later, if the executor wishes to send a reply, it needs to pass back
a handle which identifies the message so that the transport can deliver
the reply.
In the current interface, we're just passing back the context and
message to the transport so this presumes that the transport would
attach whatever it needs to one of these objects.
In the AMQP drivers in openstack.common.rpc, we set attributes like
reply_q and msg_id on the returned context. However, it would be much
better if we never touched the user-supplied context and, instead, had
some other way to pass this info to the executor and then have it passed
back to the transport.
To achieve that, add an IncomingMessage abstract class which wraps the
context and message and has a reply() method. That way, transports can
subclass this class, add whatver attributes they want and implement a
reply method.
To repeat what this means ... we can allow users of the API to use
read-only mapping objects as a context, rather than requiring it to be
an object we can set arbitrary attributes on.
2013-05-13 11:33:41 +01:00
|
|
|
def poll(self):
|
2013-06-15 19:13:45 +01:00
|
|
|
"Blocking until a message is pending and return IncomingMessage."
|
2013-05-11 13:19:59 +01:00
|
|
|
|
2013-04-30 06:49:56 +01:00
|
|
|
|
2013-10-14 10:42:15 -04:00
|
|
|
@six.add_metaclass(abc.ABCMeta)
|
2013-04-30 06:49:56 +01:00
|
|
|
class BaseDriver(object):
|
|
|
|
|
Add a TransportURL class to the public API
Nova's cells/rpc_driver.py has some code which allows user of the REST
API to update elements of a cell's transport URL (say, the host name of
the message broker) stored in the database. To achieve this, it has
a parse_transport_url() method which breaks the URL into its constituent
parts and an unparse_transport_url() which re-forms it again after
updating some of its parts.
This is all fine and, since it's fairly specialized, it wouldn't be a
big deal to leave this code in Nova for now ... except the unparse
method looks at CONF.rpc_backend to know what scheme to use in the
returned URL if now backend was specified.
oslo.messaging registers the rpc_backend option, but the ability to
reference any option registered by the library should not be relied upon
by users of the library. Imagine, for instance, if we renamed the option
in future (with backwards compat for old configurations), then this
would mean API breakage.
So, long story short - an API along these lines makes some sense, but
especially since not having it would mean we'd need to add some way to
query the name of the transport driver.
In this commit, we add a simple new TransportURL class:
>>> url = messaging.TransportURL.parse(cfg.CONF, 'foo:///')
>>> str(url), url
('foo:///', <TransportURL transport='foo'>)
>>> url.hosts.append(messaging.TransportHost(hostname='localhost'))
>>> str(url), url
('foo://localhost/', <TransportURL transport='foo', hosts=[<TransportHost hostname='localhost'>]>)
>>> url.transport = None
>>> str(url), url
('kombu://localhost/', <TransportURL transport='kombu', hosts=[<TransportHost hostname='localhost'>]>)
>>> cfg.CONF.set_override('rpc_backend', 'bar')
>>> str(url), url
('bar://localhost/', <TransportURL transport='bar', hosts=[<TransportHost hostname='localhost'>]>)
The TransportURL.parse() method equates to parse_transport_url() and
TransportURL.__str__() equates to unparse_transport().
The transport drivers are also updated to take a TransportURL as a
required argument, which simplifies the handling of transport URLs in
the drivers.
Change-Id: Ic04173476329858e4a2c2d2707e9d4aeb212d127
2013-08-12 17:16:44 +01:00
|
|
|
def __init__(self, conf, url,
|
|
|
|
default_exchange=None, allowed_remote_exmods=[]):
|
2013-04-30 06:49:56 +01:00
|
|
|
self.conf = conf
|
|
|
|
self._url = url
|
2013-05-11 16:22:37 +01:00
|
|
|
self._default_exchange = default_exchange
|
2013-08-07 13:07:05 +01:00
|
|
|
self._allowed_remote_exmods = allowed_remote_exmods
|
2013-04-30 06:49:56 +01:00
|
|
|
|
|
|
|
@abc.abstractmethod
|
2013-05-17 18:48:55 +01:00
|
|
|
def send(self, target, ctxt, message,
|
|
|
|
wait_for_reply=None, timeout=None, envelope=False):
|
2013-04-30 06:49:56 +01:00
|
|
|
"""Send a message to the given target."""
|
|
|
|
|
2013-08-01 23:32:17 +01:00
|
|
|
@abc.abstractmethod
|
|
|
|
def send_notification(self, target, ctxt, message, version):
|
|
|
|
"""Send a notification message to the given target."""
|
|
|
|
|
2013-04-30 06:49:56 +01:00
|
|
|
@abc.abstractmethod
|
2013-05-06 15:29:37 -04:00
|
|
|
def listen(self, target):
|
2013-04-30 06:49:56 +01:00
|
|
|
"""Construct a Listener for the given target."""
|
2013-07-31 07:54:25 +01:00
|
|
|
|
|
|
|
@abc.abstractmethod
|
|
|
|
def cleanup(self):
|
|
|
|
"""Release all resources."""
|