Don't raise a driver-specific error on send

Similar to doing listen() on the server side, if the driver throws an
exception when we do a cast() or call() we should translate it into
a transport-agnostic exception.

oslo/messaging/__init__.py

@@ -44,6 +44,7 @@ DriverLoadFailure = transport.DriverLoadFailure
 InvalidTransportURL = transport.InvalidTransportURL
 
 RPCVersionCapError = client.RPCVersionCapError
+ClientSendError = client.ClientSendError
 
 MessagingServerError = server.MessagingServerError
 ExecutorLoadFailure = server.ExecutorLoadFailure

oslo/messaging/rpc/client.py

@@ -20,6 +20,7 @@ import inspect
 
 from oslo.config import cfg
 
+from oslo.messaging._drivers import base as driver_base
 from oslo.messaging import _utils as utils
 from oslo.messaging import exceptions
 from oslo.messaging.openstack.common import log as logging
@@ -45,6 +46,16 @@ class RPCVersionCapError(exceptions.MessagingException):
         super(RPCVersionCapError, self).__init__(msg)
 
 
+class ClientSendError(exceptions.MessagingException):
+    """Raised if we failed to send a message to a target."""
+
+    def __init__(self, target, ex):
+        msg = 'Failed to send to target "%s": %s' % (target, ex)
+        super(ClientSendError, self).__init__(msg)
+        self.target = target
+        self.ex = ex
+
+
 class _CallContext(object):
 
     def __init__(self, transport, target, serializer,
@@ -84,7 +95,10 @@ class _CallContext(object):
         msg = self._make_message(ctxt, method, kwargs)
         if self.version_cap:
             self._check_version_cap(msg.get('version'))
-        self.transport._send(self.target, ctxt, msg)
+        try:
+            self.transport._send(self.target, ctxt, msg)
+        except driver_base.TransportDriverError as ex:
+            raise ClientSendError(self.target, ex)
 
     def _check_for_lock(self):
         locks_held = self.check_for_lock(self.conf)
@@ -108,8 +122,11 @@ class _CallContext(object):
         if self.version_cap:
             self._check_version_cap(msg.get('version'))
 
-        result = self.transport._send(self.target, ctxt, msg,
-                                      wait_for_reply=True, timeout=timeout)
+        try:
+            result = self.transport._send(self.target, ctxt, msg,
+                                          wait_for_reply=True, timeout=timeout)
+        except driver_base.TransportDriverError as ex:
+            raise ClientSendError(self.target, ex)
         return self.serializer.deserialize_entity(ctxt, result)
 
     _marker = object()

tests/test_rpc_server.py

@@ -116,7 +116,7 @@ class TestRPCServer(test_utils.BaseTestCase, ServerSetupMixin):
         else:
             self.assertTrue(False)
 
-    def test_no_target_topic(self):
+    def test_no_server_topic(self):
         transport = messaging.get_transport(self.conf, url='fake:')
         target = messaging.Target(server='testserver')
         server = messaging.get_rpc_server(transport, target, [])
@@ -128,6 +128,27 @@ class TestRPCServer(test_utils.BaseTestCase, ServerSetupMixin):
         else:
             self.assertTrue(False)
 
+    def _test_no_client_topic(self, call=True):
+        transport = messaging.get_transport(self.conf, url='fake:')
+
+        client = self._setup_client(transport, topic=None)
+
+        method = client.call if call else client.cast
+
+        try:
+            method({}, 'ping', arg='foo')
+        except Exception as ex:
+            self.assertTrue(isinstance(ex, messaging.ClientSendError), ex)
+            self.assertTrue(ex.target is not None)
+        else:
+            self.assertTrue(False)
+
+    def test_no_client_topic_call(self):
+        self._test_no_client_topic(call=True)
+
+    def test_no_client_topic_cast(self):
+        self._test_no_client_topic(call=False)
+
     def test_unknown_executor(self):
         transport = messaging.get_transport(self.conf, url='fake:')