b5955b6ca9
In this change router was redesigned in a way most appropriate for routing concept of zmq.ROUTER socket. DEALER(cli)-ROUTER(proxy)-DEALER(srv) instead of DEALER-ROUTER-DEALER-ROUTER (3 layers instead of 4) The main reason is to use zmq.DEALER identity in message routing. For this reason DealerConsumer was introduced server-side. RouterConsumer is left for peer-to-peer DEALER-ROUTER deployment option. Also handled assertions in receive-methods in order to not stop server when received message with wrong format. Change-Id: If25edf500fa8d220d4233bb13d67121824e841c6 Closes-Bug: #1558601 Related-Bug: #1555007
142 lines
4.3 KiB
Python
142 lines
4.3 KiB
Python
# Copyright 2015 Mirantis, 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
|
|
import logging
|
|
import uuid
|
|
|
|
import six
|
|
|
|
from oslo_messaging._drivers.zmq_driver.client import zmq_envelope
|
|
from oslo_messaging._drivers.zmq_driver import zmq_async
|
|
from oslo_messaging._drivers.zmq_driver import zmq_names
|
|
from oslo_messaging._i18n import _LE
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
zmq = zmq_async.import_zmq()
|
|
|
|
|
|
@six.add_metaclass(abc.ABCMeta)
|
|
class Request(object):
|
|
|
|
"""Zmq request abstract class
|
|
|
|
Represents socket (publisher) independent data object to publish.
|
|
Request object should contain all needed information for a publisher
|
|
to publish it, for instance: message payload, target, timeout
|
|
and retries etc.
|
|
"""
|
|
|
|
def __init__(self, target, context=None, message=None, retry=None):
|
|
|
|
"""Construct request object
|
|
|
|
:param target: Message destination target
|
|
:type target: oslo_messaging.Target
|
|
:param context: Message context
|
|
:type context: dict
|
|
:param message: Message payload to pass
|
|
:type message: dict
|
|
:param retry: an optional default connection retries configuration
|
|
None or -1 means to retry forever
|
|
0 means no retry
|
|
N means N retries
|
|
:type retry: int
|
|
"""
|
|
|
|
if self.msg_type not in zmq_names.MESSAGE_TYPES:
|
|
raise RuntimeError("Unknown message type!")
|
|
|
|
self.target = target
|
|
self.context = context
|
|
self.message = message
|
|
|
|
self.retry = retry
|
|
if not isinstance(retry, int) and retry is not None:
|
|
raise ValueError(
|
|
"retry must be an integer, not {0}".format(type(retry)))
|
|
|
|
self.message_id = str(uuid.uuid1())
|
|
|
|
def create_envelope(self, routing_key=None, reply_id=None):
|
|
envelope = zmq_envelope.Envelope(msg_type=self.msg_type,
|
|
message_id=self.message_id,
|
|
target=self.target,
|
|
routing_key=routing_key)
|
|
envelope.reply_id = reply_id
|
|
return envelope
|
|
|
|
@abc.abstractproperty
|
|
def msg_type(self):
|
|
"""ZMQ message type"""
|
|
|
|
def close(self):
|
|
"""Nothing to close in base request"""
|
|
|
|
|
|
class RpcRequest(Request):
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
message = kwargs.get("message")
|
|
if message['method'] is None:
|
|
errmsg = _LE("No method specified for RPC call")
|
|
LOG.error(_LE("No method specified for RPC call"))
|
|
raise KeyError(errmsg)
|
|
|
|
super(RpcRequest, self).__init__(*args, **kwargs)
|
|
|
|
|
|
class CallRequest(RpcRequest):
|
|
|
|
msg_type = zmq_names.CALL_TYPE
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.allowed_remote_exmods = kwargs.pop("allowed_remote_exmods")
|
|
|
|
self.timeout = kwargs.pop("timeout")
|
|
if self.timeout is None:
|
|
raise ValueError("Timeout should be specified for a RPC call!")
|
|
elif not isinstance(self.timeout, int):
|
|
raise ValueError(
|
|
"timeout must be an integer, not {0}"
|
|
.format(type(self.timeout)))
|
|
|
|
super(CallRequest, self).__init__(*args, **kwargs)
|
|
|
|
def create_envelope(self, routing_key=None, reply_id=None):
|
|
envelope = super(CallRequest, self).create_envelope(
|
|
routing_key, reply_id)
|
|
envelope.set('timeout', self.timeout)
|
|
return envelope
|
|
|
|
|
|
class CastRequest(RpcRequest):
|
|
|
|
msg_type = zmq_names.CAST_TYPE
|
|
|
|
|
|
class FanoutRequest(RpcRequest):
|
|
|
|
msg_type = zmq_names.CAST_FANOUT_TYPE
|
|
|
|
|
|
class NotificationRequest(Request):
|
|
|
|
msg_type = zmq_names.NOTIFY_TYPE
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.version = kwargs.pop("version")
|
|
super(NotificationRequest, self).__init__(*args, **kwargs)
|