improve error handling on callbacks

errors in callback functions should be reported to log,
but not stop the callback chain

Change-Id: I4fc509b7121960ebe59e1ad4f4b4746dfb4d5ba3
Reviewed-on: https://forge.frm2.tum.de/review/c/secop/frappy/+/30950
Tested-by: Jenkins Automated Tests <pedersen+jenkins@frm2.tum.de>
Reviewed-by: Markus Zolliker <markus.zolliker@psi.ch>

frappy/client/__init__.py

@@ -46,10 +46,10 @@ UPDATE_MESSAGES = {EVENTREPLY, READREPLY, WRITEREPLY, ERRORPREFIX + READREQUEST,
 
 VERSIONFMT= re.compile(r'^[^,]*?ISSE[^,]*,SECoP,')
 
-class UNREGISTER:
-    """a magic value, used a returned value in a callback
 
-    to indicate it has to be unregistered
+class UnregisterCallback(Exception):
+    """raise in a callback to indicate it has to be unregistered
+
     used to implement one shot callbacks
     """
 
@@ -157,8 +157,8 @@ class CacheItem(tuple):
 class ProxyClient:
     """common functionality for proxy clients"""
 
-    CALLBACK_NAMES = ('updateEvent', 'updateItem', 'descriptiveDataChange',
+    CALLBACK_NAMES = {'updateEvent', 'updateItem', 'descriptiveDataChange',
-                      'nodeStateChange', 'unhandledMessage')
+                      'nodeStateChange', 'unhandledMessage'}
     online = False  # connected or reconnecting since a short time
     state = 'disconnected'  # further possible values: 'connecting', 'reconnecting', 'connected'
     log = None
@@ -181,42 +181,41 @@ class ProxyClient:
         """
         for cbfunc in args:
             kwds[cbfunc.__name__] = cbfunc
-        for cbname in self.CALLBACK_NAMES:
+        for cbname, cbfunc in kwds.items():
-            cbfunc = kwds.pop(cbname, None)
+            if cbname not in self.CALLBACK_NAMES:
-            if not cbfunc:
+                raise TypeError(f"unknown callback: {', '.join(kwds)}")
-                continue
-            cbdict = self.callbacks[cbname]
-            cbdict[key].append(cbfunc)
 
             # immediately call for some callback types
-            if cbname == 'updateItem':
+            if cbname in ('updateItem', 'updateEvent'):
-                if key is None:
+                if key is None:  # case generic callback
-                    for (mname, pname), data in self.cache.items():
+                    cbargs = [(m, p, d) for (m, p), d in self.cache.items()]
-                        cbfunc(mname, pname, data)
                 else:
                     data = self.cache.get(key, None)
-                    if data:
+                    if data:  # case single parameter
-                        cbfunc(*key, data)  # case single parameter
+                        cbargs = [key + (data,)]
                     else:  # case key = module
-                        for (mname, pname), data in self.cache.items():
+                        cbargs = [(m, p, d) for (m, p), d in self.cache.items() if m == key]
-                            if mname == key:
+
-                                cbfunc(mname, pname, data)
+                if cbname == 'updateEvent':
-            elif cbname == 'updateEvent':
+                    # expand entry argument to (value, timestamp, readerror)
-                if key is None:
+                    cbargs = [a[0:2] + a[2] for a in cbargs]
-                    for (mname, pname), data in self.cache.items():
+
-                        cbfunc(mname, pname, *data)
-                else:
-                    data = self.cache.get(key, None)
-                    if data:
-                        cbfunc(*key, *data)  # case single parameter
-                    else:  # case key = module
-                        for (mname, pname), data in self.cache.items():
-                            if mname == key:
-                                cbfunc(mname, pname, *data)
             elif cbname == 'nodeStateChange':
-                cbfunc(self.online, self.state)
+                cbargs = [(self.online, self.state)]
-        if kwds:
+            else:
-            raise TypeError(f"unknown callback: {', '.join(kwds)}")
+                cbargs = []
+
+            do_append = True
+            for args in cbargs:
+                try:
+                    cbfunc(*args)
+                except UnregisterCallback:
+                    do_append = False
+                except Exception as e:
+                    if self.log:
+                        self.log.error('error %r calling %s%r', e, cbfunc.__name__, args)
+            if do_append:
+                self.callbacks[cbname][key].append(cbfunc)
 
     def unregister_callback(self, key, *args, **kwds):
         """unregister a callback
@@ -240,7 +239,14 @@ class ProxyClient:
         key=(<module name>, <parameter name): callbacks for specified parameter
         """
         cblist = self.callbacks[cbname].get(key, [])
-        self.callbacks[cbname][key] = [cb for cb in cblist if cb(*args) is not UNREGISTER]
+        for cbfunc in list(cblist):
+            try:
+                cbfunc(*args)
+            except UnregisterCallback:
+                cblist.remove(cbfunc)
+            except Exception as e:
+                if self.log:
+                    self.log.error('error %r calling %s%r', e, cbfunc.__name__, args)
         return bool(cblist)
 
     def updateValue(self, module, param, value, timestamp, readerror):
@@ -251,9 +257,9 @@ class ProxyClient:
         self.callback(module, 'updateItem', module, param, entry)
         self.callback((module, param), 'updateItem', module, param, entry)
         # TODO: change clients to use updateItem instead of updateEvent
-        self.callback(None, 'updateEvent', module, param, value, timestamp, readerror)
+        self.callback(None, 'updateEvent', module, param, *entry)
-        self.callback(module, 'updateEvent', module, param, value, timestamp, readerror)
+        self.callback(module, 'updateEvent', module, param, *entry)
-        self.callback((module, param), 'updateEvent', module, param, value, timestamp, readerror)
+        self.callback((module, param), 'updateEvent', module, param, *entry)
 
 
 class SecopClient(ProxyClient):

frappy/protocol/router.py

@@ -122,8 +122,7 @@ class Router(frappy.protocol.dispatcher.Dispatcher):
                             self.node_by_module[module] = node
                         self.nodes.append(node)
                         self.restart()
-                        return frappy.client.UNREGISTER
+                        raise frappy.client.UnregisterCallback()
-                    return None
 
                 node.register_callback(None, nodeStateChange)
                 logger.warning('can not connect to node %r', node.nodename)