summaryrefslogtreecommitdiff
path: root/src/leap/common/events
diff options
context:
space:
mode:
authordrebs <drebs@leap.se>2015-06-26 14:45:19 -0300
committerdrebs <drebs@leap.se>2015-06-26 14:48:56 -0300
commit10d7ebdf5226b74c54850d205a807cc650843efe (patch)
treefba27add1544f23cdac1c399052c9f35a53c98f2 /src/leap/common/events
parentc6107b88ba1eaf7e7ca97d0444e7444634aa98c2 (diff)
[bug] run callback from thread in events client
Diffstat (limited to 'src/leap/common/events')
-rw-r--r--src/leap/common/events/client.py36
-rw-r--r--src/leap/common/events/txclient.py13
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)