# HG changeset patch # User Goffi # Date 1439669607 -7200 # Node ID 227856b13d7aed9554595abd93934633ab675c31 # Parent b003dbd2b4e925b434558ac21a8b95a561f84c40 core: new tools.sat_defer module, and implementation of RTDeferredSessions: The real-time deferred session is a class which manage session of several deferreds at once (pretty much like a DeferredList), with the ability to get intermediate results. diff -r b003dbd2b4e9 -r 227856b13d7a src/core/constants.py --- a/src/core/constants.py Sat Aug 15 22:13:27 2015 +0200 +++ b/src/core/constants.py Sat Aug 15 22:13:27 2015 +0200 @@ -201,6 +201,7 @@ SAVEFILE_DATABASE = APP_NAME_FILE + ".db" IQ_SET = '/iq[@type="set"]' ENV_PREFIX = 'SAT_' # Prefix used for environment variables + IGNORE = 'ignore' ## ANSI escape sequences ## diff -r b003dbd2b4e9 -r 227856b13d7a src/tools/sat_defer.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/src/tools/sat_defer.py Sat Aug 15 22:13:27 2015 +0200 @@ -0,0 +1,200 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- + +# SàT: a XMPP client +# Copyright (C) 2009, 2010, 2011, 2012, 2013, 2014, 2015 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. + +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""tools related to deferred""" + +from sat.core.log import getLogger +log = getLogger(__name__) +from sat.core import exceptions +from twisted.internet import defer +from twisted.python import failure +from sat.core.constants import Const as C +from sat.memory import memory + +KEY_DEFERREDS = 'deferreds' +KEY_NEXT = 'next_defer' + + +class RTDeferredSessions(memory.Sessions): + """Real Time Deferred Sessions""" + + + def __init__(self, timeout=120): + """Manage list of Deferreds in real-time, allowing to get intermediate results + + @param timeout (int): nb of seconds before deferreds cancellation + """ + super(RTDeferredSessions, self).__init__(timeout=timeout, resettable_timeout=False) + + def newSession(self, deferreds, profile): + """Launch a new session with a list of deferreds + + @param deferreds(list[defer.Deferred]): list of deferred to call + @param profile: %(doc_profile)s + @param return (tupe[str, defer.Deferred]): tuple with session id and a deferred wich fire *WITHOUT RESULT* when all results are received + """ + data = {KEY_NEXT: defer.Deferred()} + session_id, session_data = super(RTDeferredSessions, self).newSession(data, profile=profile) + if isinstance(deferreds, dict): + session_data[KEY_DEFERREDS] = deferreds.values() + iterator = deferreds.iteritems() + else: + session_data[KEY_DEFERREDS] = deferreds + iterator = enumerate(deferreds) + + for idx, d in iterator: + d._RTDeferred_index = idx + d._RTDeferred_return = None + d.addCallback(self._callback, d, session_id, profile) + d.addErrback(self._errback, d, session_id, profile) + return session_id + + def _purgeSession(self, session_id, reason=u"timeout", no_warning=False, got_result=False): + """Purge the session + + @param session_id(str): id of the session to purge + @param reason (unicode): human readable reason why the session is purged + @param no_warning(bool): if True, no warning will be put in logs + @param got_result(bool): True if the session is purged after normal ending (i.e.: all the results have been gotten). + reason and no_warning are ignored if got_result is True. + @raise KeyError: session doesn't exists (anymore ?) + """ + if not got_result: + try: + timer, session_data, profile = self._sessions[session_id] + except ValueError: + raise exceptions.InternalError(u'was expecting timer, session_data and profile; is profile set ?') + + # next_defer must be called before deferreds, + # else its callback will be called by _gotResult + next_defer = session_data[KEY_NEXT] + if not next_defer.called: + next_defer.errback(failure.Failure(defer.CancelledError(reason))) + + deferreds = session_data[KEY_DEFERREDS] + for d in deferreds: + d.cancel() + + if not no_warning: + log.warning(u"RTDeferredList cancelled: {} (profile {})".format(reason, profile)) + + super(RTDeferredSessions, self)._purgeSession(session_id) + + def _gotResult(self, session_id, profile): + """Method called after each callback or errback + + manage the next_defer deferred + """ + session_data = self.profileGet(session_id, profile) + defer_next = session_data[KEY_NEXT] + if not defer_next.called: + defer_next.callback(None) + + def _callback(self, result, deferred, session_id, profile): + deferred._RTDeferred_return = (True, result) + self._gotResult(session_id, profile) + + def _errback(self, failure, deferred, session_id, profile): + deferred._RTDeferred_return = (False, failure) + self._gotResult(session_id, profile) + + def cancel(self, session_id, reason=u"timeout", no_log=False): + """Stop this RTDeferredList + + Cancel all remaining deferred, and call self.final_defer.errback + @param reason (unicode): reason of the cancellation + @param no_log(bool): if True, don't log the cancellation + """ + self._purgeSession(session_id, reason=reason, no_warning=no_log) + + def getResults(self, session_id, on_success=None, on_error=None, profile=C.PROF_KEY_NONE): + """Get current results of a real-time deferred session + + result already gotten are deleted + @param session_id(str): session id + @param on_success: can be: + - None: add success normaly to results + - callable: replace result by the return value of on_success(result) (may be deferred) + @param on_error: can be: + - None: add error normaly to results + - C.IGNORE: don't put errors in results + - callable: replace failure by the return value of on_error(failure) (may be deferred) + @param profile=%(doc_profile)s + @param result(tuple): tuple(remaining, results) where: + - remaining[int] is the number of remaining deferred + (deferreds from which we don't have result yet) + - results is a dict where: + - key is the index of the deferred if deferred is a list, or its key if it's a dict + - value = (success, result) where: + - success is True if the deferred was successful + - result is the result in case of success, else the failure + If remaining == 0, the session is ended + @raise KeyError: the session is already finished or doesn't exists at all + """ + if profile == C.PROF_KEY_NONE: + raise exceptions.ProfileNotSetError + session_data = self.profileGet(session_id, profile) + + @defer.inlineCallbacks + def next_cb(dummy): + # we got one or several results + results = {} + filtered_data = [] # used to keep deferreds without results + deferreds = session_data[KEY_DEFERREDS] + + for d in deferreds: + if d._RTDeferred_return: # we don't use d.called as called is True before the full callbacks chain has been called + # we have a result + idx = d._RTDeferred_index + success, result = d._RTDeferred_return + if success: + if on_success is not None: + if callable(on_success): + result = yield on_success(result) + else: + raise exceptions.InternalError('Unknown value of on_success: {}'.format(on_success)) + + else: + if on_error is not None: + if on_error == C.IGNORE: + continue + elif callable(on_error): + result = yield on_error(result) + else: + raise exceptions.InternalError('Unknown value of on_error: {}'.format(on_error)) + results[idx] = (success, result) + else: + filtered_data.append(d) + + # we change the deferred with the filtered list + # in other terms, we don't want anymore deferred from which we have got the result + session_data[KEY_DEFERREDS] = filtered_data + + if filtered_data: + # we create a new next_defer only if we are still waiting for results + session_data[KEY_NEXT] = defer.Deferred() + else: + # no more data to get, the result have been gotten, + # we can cleanly finish the session + self._purgeSession(session_id, got_result=True) + + defer.returnValue((len(filtered_data), results)) + + # we wait for a result + return session_data[KEY_NEXT].addCallback(next_cb)