RPC Client

class oslo_messaging.RPCClient(transport, target, timeout=None, version_cap=None, serializer=None, retry=None, call_monitor_timeout=None, transport_options=None)

A class for invoking methods on remote RPC servers.

The RPCClient class is responsible for sending method invocations to and receiving return values from remote RPC servers via a messaging transport.

Two RPC patterns are supported: RPC calls and RPC casts.

An RPC cast is used when an RPC method does not return a value to the caller. An RPC call is used when a return value is expected from the method. For further information see the cast() and call() methods.

The default target used for all subsequent calls and casts is supplied to the RPCClient constructor. The client uses the target to control how the RPC request is delivered to a server. If only the target’s topic (and optionally exchange) are set, then the RPC can be serviced by any server that is listening to that topic (and exchange). If multiple servers are listening on that topic/exchange, then one server is picked using a best-effort round-robin algorithm. Alternatively, the client can set the Target’s server attribute to the name of a specific server to send the RPC request to one particular server. In the case of RPC cast, the RPC request can be broadcast to all servers listening to the Target’s topic/exchange by setting the Target’s fanout property to True.

While the default target is set on construction, target attributes can be overridden for individual method invocations using the prepare() method.

A method invocation consists of a request context dictionary, a method name and a dictionary of arguments.

This class is intended to be used by wrapping it in another class which provides methods on the subclass to perform the remote invocation using call() or cast():

class TestClient(object):

    def __init__(self, transport):
        target = messaging.Target(topic='test', version='2.0')
        self._client = messaging.RPCClient(transport, target)

    def test(self, ctxt, arg):
        return self._client.call(ctxt, 'test', arg=arg)

An example of using the prepare() method to override some attributes of the default target:

def test(self, ctxt, arg):
    cctxt = self._client.prepare(version='2.5')
    return cctxt.call(ctxt, 'test', arg=arg)

RPCClient have a number of other properties - for example, timeout and version_cap - which may make sense to override for some method invocations, so they too can be passed to prepare():

def test(self, ctxt, arg):
    cctxt = self._client.prepare(timeout=10)
    return cctxt.call(ctxt, 'test', arg=arg)

However, this class can be used directly without wrapping it another class. For example:

transport = messaging.get_rpc_transport(cfg.CONF)
target = messaging.Target(topic='test', version='2.0')
client = messaging.RPCClient(transport, target)
client.call(ctxt, 'test', arg=arg)

but this is probably only useful in limited circumstances as a wrapper class will usually help to make the code much more obvious.

If the connection to the messaging service is not active when an RPC request is made the client will block waiting for the connection to complete. If the connection fails to complete, the client will try to re-establish that connection. By default this will continue indefinitely until the connection completes. However, the retry parameter can be used to have the RPC request fail with a MessageDeliveryFailure after the given number of retries. For example:

client = messaging.RPCClient(transport, target, retry=None)
client.call(ctxt, 'sync')
try:
    client.prepare(retry=0).cast(ctxt, 'ping')
except messaging.MessageDeliveryFailure:
    LOG.error("Failed to send ping message")
call(ctxt, method, **kwargs)

Invoke a method and wait for a reply.

The call() method is used to invoke RPC methods that return a value. Since only a single return value is permitted it is not possible to call() to a fanout target.

call() will block the calling thread until the messaging transport provides the return value, a timeout occurs, or a non-recoverable error occurs.

call() guarantees that the RPC request is done ‘at-most-once’ which ensures that the call will never be duplicated. However if the call should fail or time out before the return value arrives then there are no guarantees whether or not the method was invoked.

Since call() blocks until completion of the RPC method, call()s from the same thread are guaranteed to be processed in-order.

Method arguments must either be primitive types or types supported by the client’s serializer (if any). Similarly, the request context must be a dict unless the client’s serializer supports serializing another type.

The semantics of how any errors raised by the remote RPC endpoint method are handled are quite subtle.

Firstly, if the remote exception is contained in one of the modules listed in the allow_remote_exmods messaging.get_rpc_transport() parameter, then it this exception will be re-raised by call(). However, such locally re-raised remote exceptions are distinguishable from the same exception type raised locally because re-raised remote exceptions are modified such that their class name ends with the ‘_Remote’ suffix so you may do:

if ex.__class__.__name__.endswith('_Remote'):
    # Some special case for locally re-raised remote exceptions

Secondly, if a remote exception is not from a module listed in the allowed_remote_exmods list, then a messaging.RemoteError exception is raised with all details of the remote exception.

Parameters

  • ctxt (dict) – a request context dict

  • method (str) – the method name

  • kwargs (dict) – a dict of method arguments

Raises

MessagingTimeout, RemoteError, MessageDeliveryFailure

can_send_version(version=<object object>)

Check to see if a version is compatible with the version cap.

cast(ctxt, method, **kwargs)

Invoke a method without blocking for a return value.

The cast() method is used to invoke an RPC method that does not return a value. cast() RPC requests may be broadcast to all Servers listening on a given topic by setting the fanout Target property to True.

The cast() operation is best-effort: cast() will block the calling thread until the RPC request method is accepted by the messaging transport, but cast() does not verify that the RPC method has been invoked by the server. cast() does guarantee that the method will be not executed twice on a destination (e.g. ‘at-most-once’ execution).

There are no ordering guarantees across successive casts, even among casts to the same destination. Therefore methods may be executed in an order different from the order in which they are cast.

Method arguments must either be primitive types or types supported by the client’s serializer (if any).

Similarly, the request context must be a dict unless the client’s serializer supports serializing another type.

Parameters

  • ctxt (dict) – a request context dict

  • method (str) – the method name

  • kwargs (dict) – a dict of method arguments

Raises

MessageDeliveryFailure if the messaging transport fails to accept the request.

prepare(exchange=<object object>, topic=<object object>, namespace=<object object>, version=<object object>, server=<object object>, fanout=<object object>, timeout=<object object>, version_cap=<object object>, retry=<object object>, call_monitor_timeout=<object object>, transport_options=<object object>)

Prepare a method invocation context.

Use this method to override client properties for an individual method invocation. For example:

def test(self, ctxt, arg):
    cctxt = self.prepare(version='2.5')
    return cctxt.call(ctxt, 'test', arg=arg)
Parameters

  • exchange (str) – see Target.exchange

  • topic (str) – see Target.topic

  • namespace (str) – see Target.namespace

  • version (str) – requirement the server must support, see Target.version

  • server (str) – send to a specific server, see Target.server

  • fanout (bool) – send to all servers on topic, see Target.fanout

  • timeout (int or float) – an optional default timeout (in seconds) for call()s

  • version_cap (str) – raise a RPCVersionCapError version exceeds this cap

  • retry (int) – an optional connection retries configuration: None or -1 means to retry forever. 0 means no retry is attempted. N means attempt at most N retries.

  • transport_options (dictionary) – additional parameters to configure the driver for example to send parameters as “mandatory” flag in RabbitMQ

  • call_monitor_timeout (int) – an optional timeout (in seconds) for active call heartbeating. If specified, requires the server to heartbeat long-running calls at this interval (less than the overall timeout parameter).

exception oslo_messaging.RemoteError(exc_type=None, value=None, traceback=None)

Signifies that a remote endpoint method has raised an exception.

Contains a string representation of the type of the original exception, the value of the original exception, and the traceback. These are sent to the parent as a joined string so printing the exception contains all of the relevant info.