"""
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):
# 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.
"""
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)