# encoding: utf-8 # -*- test-case-name: IPython.kernel.test.test_pendingdeferred -*- """Classes to manage pending Deferreds. A pending deferred is a deferred that may or may not have fired. This module is useful for taking a class whose methods return deferreds and wrapping it to provide API that keeps track of those deferreds for later retrieval. See the tests for examples of its usage. """ __docformat__ = "restructuredtext en" #------------------------------------------------------------------------------- # Copyright (C) 2008 The IPython Development Team # # Distributed under the terms of the BSD License. The full license is in # the file COPYING, distributed as part of this software. #------------------------------------------------------------------------------- #------------------------------------------------------------------------------- # Imports #------------------------------------------------------------------------------- from twisted.application import service from twisted.internet import defer, reactor from twisted.python import log, components, failure from zope.interface import Interface, implements, Attribute from IPython.kernel.twistedutil import gatherBoth from IPython.kernel import error from IPython.external import guid from IPython.tools import growl class PendingDeferredManager(object): """A class to track pending deferreds. To track a pending deferred, the user of this class must first get a deferredID by calling `get_next_deferred_id`. Then the user calls `save_pending_deferred` passing that id and the deferred to be tracked. To later retrieve it, the user calls `get_pending_deferred` passing the id. """ def __init__(self): """Manage pending deferreds.""" self.results = {} # Populated when results are ready self.deferred_ids = [] # List of deferred ids I am managing self.deferreds_to_callback = {} # dict of lists of deferreds to callback def get_deferred_id(self): return guid.generate() def quick_has_id(self, deferred_id): return deferred_id in self.deferred_ids def _save_result(self, result, deferred_id): if self.quick_has_id(deferred_id): self.results[deferred_id] = result self._trigger_callbacks(deferred_id) def _trigger_callbacks(self, deferred_id): # Go through and call the waiting callbacks result = self.results.get(deferred_id) if result is not None: # Only trigger if there is a result try: d = self.deferreds_to_callback.pop(deferred_id) except KeyError: d = None if d is not None: if isinstance(result, failure.Failure): d.errback(result) else: d.callback(result) self.delete_pending_deferred(deferred_id) def save_pending_deferred(self, d, deferred_id=None): """Save the result of a deferred for later retrieval. This works even if the deferred has not fired. Only callbacks and errbacks applied to d before this method is called will be called no the final result. """ if deferred_id is None: deferred_id = self.get_deferred_id() self.deferred_ids.append(deferred_id) d.addBoth(self._save_result, deferred_id) return deferred_id def _protected_del(self, key, container): try: del container[key] except Exception: pass def delete_pending_deferred(self, deferred_id): """Remove a deferred I am tracking and add a null Errback. :Parameters: deferredID : str The id of a deferred that I am tracking. """ if self.quick_has_id(deferred_id): # First go through a errback any deferreds that are still waiting d = self.deferreds_to_callback.get(deferred_id) if d is not None: d.errback(failure.Failure(error.AbortedPendingDeferredError("pending deferred has been deleted: %r"%deferred_id))) # Now delete all references to this deferred_id ind = self.deferred_ids.index(deferred_id) self._protected_del(ind, self.deferred_ids) self._protected_del(deferred_id, self.deferreds_to_callback) self._protected_del(deferred_id, self.results) else: raise error.InvalidDeferredID('invalid deferred_id: %r' % deferred_id) def clear_pending_deferreds(self): """Remove all the deferreds I am tracking.""" for did in self.deferred_ids: self.delete_pending_deferred(did) def _delete_and_pass_through(self, r, deferred_id): self.delete_pending_deferred(deferred_id) return r def get_pending_deferred(self, deferred_id, block): if not self.quick_has_id(deferred_id) or self.deferreds_to_callback.get(deferred_id) is not None: return defer.fail(failure.Failure(error.InvalidDeferredID('invalid deferred_id: %r' + deferred_id))) result = self.results.get(deferred_id) if result is not None: self.delete_pending_deferred(deferred_id) if isinstance(result, failure.Failure): return defer.fail(result) else: return defer.succeed(result) else: # Result is not ready if block: d = defer.Deferred() self.deferreds_to_callback[deferred_id] = d return d else: return defer.fail(failure.Failure(error.ResultNotCompleted("result not completed: %r" % deferred_id))) def two_phase(wrapped_method): """Wrap methods that return a deferred into a two phase process. This transforms:: foo(arg1, arg2, ...) -> foo(arg1, arg2,...,block=True). The wrapped method will then return a deferred to a deferred id. This will only work on method of classes that inherit from `PendingDeferredManager`, as that class provides an API for block is a boolean to determine if we should use the two phase process or just simply call the wrapped method. At this point block does not have a default and it probably won't. """ def wrapper_two_phase(pdm, *args, **kwargs): try: block = kwargs.pop('block') except KeyError: block = True # The default if not specified if block: return wrapped_method(pdm, *args, **kwargs) else: d = wrapped_method(pdm, *args, **kwargs) deferred_id=pdm.save_pending_deferred(d) return defer.succeed(deferred_id) return wrapper_two_phase