[bug] run callback from thread in events client

2 files changed, 46 insertions, 3 deletions

diff --git a/src/leap/common/events/client.py b/src/leap/common/events/client.py
index 4852b5a..1a026ea 100644
--- a/src/leap/common/events/client.py
+++ b/src/leap/common/events/client.py
@@ -180,14 +180,30 @@ class EventsClient(object):
         """
         Handle an incoming event.
 
-        :param msg: The incoming message.
-        :type msg: list(str)
+        :param event: The event to be sent.
+        :type event: Event
+        :param content: The content of the event.
+        :type content: list
         """
         logger.debug("Handling event %s..." % event)
         for uid in self._callbacks[event].keys():
             callback = self._callbacks[event][uid]
             logger.debug("Executing callback %s." % uid)
-            callback(event, *content)
+            self._run_callback(callback, event, content)
+
+    @abstractmethod
+    def _run_callback(self, callback, event, content):
+        """
+        Run a callback.
+
+        :param callback: The callback to be run.
+        :type callback: callable(event, *content)
+        :param event: The event to be sent.
+        :type event: Event
+        :param content: The content of the event.
+        :type content: list
+        """
+        pass
 
     @abstractmethod
     def _subscribe(self, tag):
@@ -371,6 +387,20 @@ class EventsClientThread(threading.Thread, EventsClient):
         # add send() as a callback for ioloop so it works between threads
         self._loop.add_callback(lambda: self._push.send(data))
 
+    def _run_callback(self, callback, event, content):
+        """
+        Run a callback.
+
+        :param callback: The callback to be run.
+        :type callback: callable(event, *content)
+        :param event: The event to be sent.
+        :type event: Event
+        :param content: The content of the event.
+        :type content: list
+        """
+        from twisted.internet import reactor
+        reactor.callFromThread(callback, event, *content)
+
     def register(self, event, callback, uid=None, replace=False):
         """
         Register a callback to be executed when an event is received.
diff --git a/src/leap/common/events/txclient.py b/src/leap/common/events/txclient.py
index 8206ed5..0dcfc08 100644
--- a/src/leap/common/events/txclient.py
+++ b/src/leap/common/events/txclient.py
@@ -112,6 +112,19 @@ class EventsTxClient(TxZmqClientComponent, EventsClient):
         """
         self._push.send(data)
 
+    def _run_callback(self, callback, event, content):
+        """
+        Run a callback.
+
+        :param callback: The callback to be run.
+        :type callback: callable(event, *content)
+        :param event: The event to be sent.
+        :type event: Event
+        :param content: The content of the event.
+        :type content: list
+        """
+        callback(event, *content)
+
     def shutdown(self):
         TxZmqClientComponent.shutdown(self)
         EventsClient.shutdown(self)