added listener support to queues, also added support for non version specific tests

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@542955 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 35 insertions, 13 deletions

diff --git a/python/qpid/delegate.py b/python/qpid/delegate.py
index 90e5c1edc8..8f5033e485 100644
--- a/python/qpid/delegate.py
+++ b/python/qpid/delegate.py
@@ -21,9 +21,12 @@
 Delegate implementation intended for use with the peer module.
 """
 
-import threading, inspect
+import threading, inspect, traceback, sys
 from connection import Method, Request, Response
 
+def _handler_name(method):
+  return "%s_%s" % (method.klass.name, method.name)
+
 class Delegate:
 
   def __init__(self):
@@ -36,11 +39,15 @@ class Delegate:
     try:
       handler = self.handlers[method]
     except KeyError:
-      name = "%s_%s" % (method.klass.name, method.name)
+      name = _handler_name(method)
       handler = getattr(self, name)
       self.handlers[method] = handler
 
-    return handler(channel, frame)
+    try:
+      return handler(channel, frame)
+    except:
+      print >> sys.stderr, "Error in handler: %s\n\n%s" % \
+            (_handler_name(method), traceback.format_exc())
 
   def close(self, reason):
     print "Connection closed: %s" % reason
diff --git a/python/qpid/queue.py b/python/qpid/queue.py
index 5438b328ab..af0565b6cc 100644
--- a/python/qpid/queue.py
+++ b/python/qpid/queue.py
@@ -31,15 +31,29 @@ class Queue(BaseQueue):
 
   END = object()
 
+  def __init__(self, *args, **kwargs):
+    BaseQueue.__init__(self, *args, **kwargs)
+    self._real_put = self.put
+    self.listener = self._real_put
+
   def close(self):
     self.put(Queue.END)
 
   def get(self, block = True, timeout = None):
-    result = BaseQueue.get(self, block, timeout)
-    if result == Queue.END:
-      # this guarantees that any other waiting threads or any future
-      # calls to get will also result in a Closed exception
-      self.put(Queue.END)
-      raise Closed()
-    else:
-      return result
+    self.put = self._real_put
+    try:
+      result = BaseQueue.get(self, block, timeout)
+      if result == Queue.END:
+        # this guarantees that any other waiting threads or any future
+        # calls to get will also result in a Closed exception
+        self.put(Queue.END)
+        raise Closed()
+      else:
+        return result
+    finally:
+      self.put = self.listener
+      pass
+
+  def listen(self, listener):
+    self.listener = listener
+    self.put = self.listener
diff --git a/python/qpid/testlib.py b/python/qpid/testlib.py
index ba2dbe9fc3..2a5e4dcfcd 100644
--- a/python/qpid/testlib.py
+++ b/python/qpid/testlib.py
@@ -122,10 +122,11 @@ Options:
         print "Using specification from:", self.specfile
         self.spec = qpid.spec.load(self.specfile, *self.errata)
         if len(self.tests) == 0:
+            self.tests=findmodules("tests")
             if self.use08spec():
-                self.tests=findmodules("tests_0-8")
+                self.tests+=findmodules("tests_0-8")
             else:
-                self.tests=findmodules("tests_0-9")
+                self.tests+=findmodules("tests_0-9")
 
     def testSuite(self):
         class IgnoringTestSuite(unittest.TestSuite):