threads: errors management

weboob/__init__.py

@@ -18,4 +18,4 @@ Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 
 """
 
-from .ouiboube import Weboob
+from .ouiboube import Weboob, CallErrors

weboob/bcall.py

@@ -21,9 +21,15 @@ Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 from __future__ import with_statement
 
 from logging import debug
+from copy import copy
 from threading import Thread, Event, RLock, Timer
 
-__all__ = ['BackendsCall']
+__all__ = ['BackendsCall', 'CallErrors']
+
+class CallErrors(Exception):
+    def __init__(self, errors):
+        Exception.__init__(self, "Several errors have been raised:\n%s" % ('\n'.join(['%s: %s' % (b, e) for b, e in errors])))
+        self.errors = copy(errors)
 
 class Result(object):
     def __init__(self, backend, result):
@@ -54,7 +60,9 @@ class BackendsCall(object):
         self.response_event = Event()
         # Waiting responses
         self.responses = []
-        # Timers
+        # Errors
+        self.errors = []
+        # Threads
         self.threads = []
 
         # Create jobs for each backend
@@ -68,21 +76,30 @@ class BackendsCall(object):
         with b:
             try:
                 # Call method on backend
-                r = getattr(b, function)(*args, **kwargs)
-                debug('%s: Got answer! %s' % (b, r))
-
-                if hasattr(r, '__iter__'):
-                    # Loop on iterator
-                    for e in r:
-                        # Lock mutex only in loop in case the iterator is slow
-                        # (for example if backend do some parsing operations)
-                        with self.mutex:
-                            self.responses.append((b,e))
-                            self.response_event.set()
-                else:
+                try:
+                    if callable(function):
+                        r = function(b, *args, **kwargs)
+                    else:
+                        r = getattr(b, function)(*args, **kwargs)
+                except Exception, e:
                     with self.mutex:
-                        self.responses.append((b,r))
-                        self.response_event.set()
+                        # TODO save backtrace and/or print it here (with debug)
+                        self.errors.append((b, e))
+                else:
+                    debug('%s: Got answer! %s' % (b, r))
+
+                    if hasattr(r, '__iter__'):
+                        # Loop on iterator
+                        for e in r:
+                            # Lock mutex only in loop in case the iterator is slow
+                            # (for example if backend do some parsing operations)
+                            with self.mutex:
+                                self.responses.append((b,e))
+                                self.response_event.set()
+                    else:
+                        with self.mutex:
+                            self.responses.append((b,r))
+                            self.response_event.set()
             finally:
                 with self.mutex:
                     # This backend is now finished
@@ -93,7 +110,7 @@ class BackendsCall(object):
                     self.finish_event.set()
                     self.response_event.set()
 
-    def _callback_thread_run(self, function):
+    def _callback_thread_run(self, callback, errback):
         responses = []
         while not self.finish_event.isSet():
             self.response_event.wait()
@@ -106,9 +123,14 @@ class BackendsCall(object):
 
             # Consume responses
             while responses:
-                function(*responses.pop())
+                callback(*responses.pop())
 
-    def callback_thread(self, function):
+        if errback:
+            with self.mutex:
+                while self.errors:
+                    errback(*self.errors.pop())
+
+    def callback_thread(self, callback, errback=None):
         """
         Call this method to create a thread which will callback a
         specified function everytimes a new result comes.
@@ -116,11 +138,12 @@ class BackendsCall(object):
         When the process is over, the function will be called with
         both arguments set to None.
 
-        The function prototype is:
-            def function(backend, result)
+        The functions prototypes:
+            def callback(backend, result)
+            def errback(backend, error)
 
         """
-        return Thread(target=self._callback_thread_run, args=(function,))
+        return Thread(target=self._callback_thread_run, args=(callback, errback))
 
     def __iter__(self):
         # Don't know how to factorize with _callback_thread_run
@@ -137,3 +160,8 @@ class BackendsCall(object):
             # Consume responses
             while responses:
                 yield Result(*responses.pop())
+
+        # Raise errors
+        with self.mutex:
+            if self.errors:
+                raise CallErrors(self.errors)

weboob/ouiboube.py

@@ -18,15 +18,17 @@ Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 
 """
 
+from __future__ import with_statement
+
 import os
 from logging import warning
 
-from weboob.bcall import BackendsCall
+from weboob.bcall import BackendsCall, CallErrors
 from weboob.modules import ModulesLoader, BackendsConfig
 from weboob.scheduler import Scheduler
 
 
-__all__ = ['Weboob']
+__all__ = ['Weboob', 'CallErrors']
 
 
 class Weboob(object):
@@ -93,7 +95,8 @@ class Weboob(object):
     def iter_backends(self, caps=None):
         for name, backend in self.backends.iteritems():
             if caps is None or backend.has_caps(caps):
-                yield backend
+                with backend:
+                    yield backend
 
     def do(self, function, *args, **kwargs):
         backends = [b for b in self.iter_backends()]