Attachment 685286 Details for Bug 574571 – Capture encode and decode exceptions

[patch] Capture encode and decode exceptions

qpid-2453.patch (text/plain), 3.29 KB, created by Ernie on 2013-01-22 16:49:08 UTC

(hide)

Description:

Filename:

MIME Type:

Creator: Ernie

Created: 2013-01-22 16:49:08 UTC

Size: 3.29 KB

patch

obsolete

>diff --git a/qpid/python/qpid/messaging/driver.py b/qpid/python/qpid/messaging/driver.py
>index 3cb62d7..ab70349 100644
>--- a/qpid/python/qpid/messaging/driver.py
>+++ b/qpid/python/qpid/messaging/driver.py
>@@ -511,6 +511,9 @@ class Driver:
>         self.engine.dispatch()
>     except HeartbeatTimeout, e:
>       self.close_engine(e)
>+    except NonBlockingError, e:
>+      msg = compat.format_exc()
>+      self.connection.error = NonBlockingError(text=msg)
>     except:
>       # XXX: Does socket get leaked if this occurs?
>       msg = compat.format_exc()
>@@ -1260,7 +1263,11 @@ class Engine:
>     if msg.ttl is not None:
>       dp.ttl = long(msg.ttl*1000)
>     enc, dec = get_codec(msg.content_type)
>-    body = enc(msg.content)
>+    try:
>+      body = enc(msg.content)
>+    except AttributeError, e:
>+      # convert to non-blocking EncodeError
>+      raise EncodeError(e)
> 
>     # XXX: this is not safe for out of order, can this be triggered by pre_ack?
>     def msg_acked():
>@@ -1309,7 +1316,10 @@ class Engine:
> 
>     ap = mp.application_headers
>     enc, dec = get_codec(mp.content_type)
>-    content = dec(xfr.payload)
>+    try:
>+      content = dec(xfr.payload)
>+    except Exception, e:
>+      raise DecodeError(e)
>     msg = Message(content)
>     msg.id = mp.message_id
>     if ap is not None:
>diff --git a/qpid/python/qpid/messaging/endpoints.py b/qpid/python/qpid/messaging/endpoints.py
>index 95ff551..d187638 100644
>--- a/qpid/python/qpid/messaging/endpoints.py
>+++ b/qpid/python/qpid/messaging/endpoints.py
>@@ -199,7 +199,10 @@ class Connection(Endpoint):
>   def check_error(self):
>     if self.error:
>       self._condition.gc()
>-      raise self.error
>+      e = self.error
>+      if getattr(e, "nonblocking", False):
>+          self.error = None
>+      raise e
> 
>   def get_error(self):
>     return self.error
>@@ -882,8 +885,14 @@ class Sender(Endpoint):
>     if self.synced < mno:
>       self.synced = mno
>       self._wakeup()
>-    if not self._ewait(lambda: self.acked >= mno, timeout=timeout):
>-      raise Timeout("sender sync timed out")
>+    try:
>+      if not self._ewait(lambda: self.acked >= mno, timeout=timeout):
>+        raise Timeout("sender sync timed out")
>+    except NonBlockingError, e:
>+      # clean bad message so we can continue
>+      self.acked = mno
>+      self.session.outgoing.pop(0)
>+      raise Exception(e)
> 
>   @synchronized
>   def close(self, timeout=None):
>diff --git a/qpid/python/qpid/messaging/exceptions.py b/qpid/python/qpid/messaging/exceptions.py
>index 0296d61..4bb8844 100644
>--- a/qpid/python/qpid/messaging/exceptions.py
>+++ b/qpid/python/qpid/messaging/exceptions.py
>@@ -31,6 +31,8 @@ class MessagingError(Exception):
>     if self.code is None:
>       msg = self.text
>     else:
>+      if not self.text:
>+          self.text = self.__class__.__name__
>       msg = "%s(%s)" % (self.text, self.code)
>     if info:
>       msg += " " + ", ".join(["%s=%r" % (k, v) for k, v in self.info.items()])
>@@ -154,3 +156,15 @@ class Empty(FetchError):
>   available within the alloted time.
>   """
>   pass
>+
>+## Non-blocking errors
>+class NonBlockingError(MessagingError):
>+  def __init__(self, code=None, text=None, **info):
>+    MessagingError.__init__(self, code, text)
>+    self.nonblocking = True
>+
>+class EncodeError(NonBlockingError):
>+  pass
>+
>+class DecodeError(NonBlockingError):
>+  pass
>\ No newline at end of file

Actions: View | Diff

Attachments on bug 574571: 679756 | 685286 | 686001 | 737691