promises.py 7.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243
  1. """Promise implementation."""
  2. from __future__ import absolute_import, unicode_literals
  3. import sys
  4. from collections import deque
  5. import inspect
  6. from weakref import ref
  7. try:
  8. from weakref import WeakMethod
  9. except ImportError:
  10. from vine.backports.weakref_backports import WeakMethod
  11. from .abstract import Thenable
  12. from .five import python_2_unicode_compatible, reraise
  13. __all__ = ['promise']
  14. @Thenable.register
  15. @python_2_unicode_compatible
  16. class promise(object):
  17. """Promise of future evaluation.
  18. This is a special implementation of promises in that it can
  19. be used both for "promise of a value" and lazy evaluation.
  20. The biggest upside for this is that everything in a promise can also be
  21. a promise, e.g. filters, callbacks and errbacks can all be promises.
  22. Usage examples:
  23. .. code-block:: python
  24. >>> from __future__ import print_statement # noqa
  25. >>> p = promise()
  26. >>> p.then(promise(print, ('OK',))) # noqa
  27. >>> p.on_error = promise(print, ('ERROR',)) # noqa
  28. >>> p(20)
  29. OK, 20
  30. >>> p.then(promise(print, ('hello',))) # noqa
  31. hello, 20
  32. >>> p.throw(KeyError('foo'))
  33. ERROR, KeyError('foo')
  34. >>> p2 = promise()
  35. >>> p2.then(print) # noqa
  36. >>> p2.cancel()
  37. >>> p(30)
  38. Example:
  39. .. code-block:: python
  40. from vine import promise, wrap
  41. class Protocol(object):
  42. def __init__(self):
  43. self.buffer = []
  44. def receive_message(self):
  45. return self.read_header().then(
  46. self.read_body).then(
  47. wrap(self.prepare_body))
  48. def read(self, size, callback=None):
  49. callback = callback or promise()
  50. tell_eventloop_to_read(size, callback)
  51. return callback
  52. def read_header(self, callback=None):
  53. return self.read(4, callback)
  54. def read_body(self, header, callback=None):
  55. body_size, = unpack('>L', header)
  56. return self.read(body_size, callback)
  57. def prepare_body(self, value):
  58. self.buffer.append(value)
  59. """
  60. if not hasattr(sys, 'pypy_version_info'): # pragma: no cover
  61. __slots__ = (
  62. 'fun', 'args', 'kwargs', 'ready', 'failed',
  63. 'value', 'reason', '_svpending', '_lvpending',
  64. 'on_error', 'cancelled', 'weak', '__weakref__',
  65. )
  66. def __init__(self, fun=None, args=None, kwargs=None,
  67. callback=None, on_error=None, weak=False):
  68. self.weak = weak
  69. self.fun = self._get_fun_or_weakref(fun=fun, weak=weak)
  70. self.args = args or ()
  71. self.kwargs = kwargs or {}
  72. self.ready = False
  73. self.failed = False
  74. self.value = None
  75. self.reason = None
  76. # Optimization
  77. # Most promises will only have one callback, so we optimize for this
  78. # case by using a list only when there are multiple callbacks.
  79. # s(calar) pending / l(ist) pending
  80. self._svpending = None
  81. self._lvpending = None
  82. self.on_error = on_error
  83. self.cancelled = False
  84. if callback is not None:
  85. self.then(callback)
  86. if self.fun:
  87. assert self.fun and callable(fun)
  88. @staticmethod
  89. def _get_fun_or_weakref(fun, weak):
  90. """Return the callable or a weak reference.
  91. Handles both bound and unbound methods.
  92. """
  93. if not weak:
  94. return fun
  95. if inspect.ismethod(fun):
  96. return WeakMethod(fun)
  97. else:
  98. return ref(fun)
  99. def __repr__(self):
  100. return ('<{0} --> {1!r}>' if self.fun else '<{0}>').format(
  101. '{0}@0x{1:x}'.format(type(self).__name__, id(self)), self.fun,
  102. )
  103. def cancel(self):
  104. self.cancelled = True
  105. try:
  106. if self._svpending is not None:
  107. self._svpending.cancel()
  108. if self._lvpending is not None:
  109. for pending in self._lvpending:
  110. pending.cancel()
  111. if isinstance(self.on_error, Thenable):
  112. self.on_error.cancel()
  113. finally:
  114. self._svpending = self._lvpending = self.on_error = None
  115. def __call__(self, *args, **kwargs):
  116. retval = None
  117. if self.cancelled:
  118. return
  119. final_args = self.args + args if args else self.args
  120. final_kwargs = dict(self.kwargs, **kwargs) if kwargs else self.kwargs
  121. # self.fun may be a weakref
  122. fun = self._fun_is_alive(self.fun)
  123. if fun is not None:
  124. try:
  125. retval = fun(*final_args, **final_kwargs)
  126. self.value = (ca, ck) = (retval,), {}
  127. except Exception:
  128. return self.throw()
  129. else:
  130. self.value = (ca, ck) = final_args, final_kwargs
  131. self.ready = True
  132. svpending = self._svpending
  133. if svpending is not None:
  134. try:
  135. svpending(*ca, **ck)
  136. finally:
  137. self._svpending = None
  138. else:
  139. lvpending = self._lvpending
  140. try:
  141. while lvpending:
  142. p = lvpending.popleft()
  143. p(*ca, **ck)
  144. finally:
  145. self._lvpending = None
  146. return retval
  147. def _fun_is_alive(self, fun):
  148. return fun() if self.weak else self.fun
  149. def then(self, callback, on_error=None):
  150. if not isinstance(callback, Thenable):
  151. callback = promise(callback, on_error=on_error)
  152. if self.cancelled:
  153. callback.cancel()
  154. return callback
  155. if self.failed:
  156. callback.throw(self.reason)
  157. elif self.ready:
  158. args, kwargs = self.value
  159. callback(*args, **kwargs)
  160. if self._lvpending is None:
  161. svpending = self._svpending
  162. if svpending is not None:
  163. self._svpending, self._lvpending = None, deque([svpending])
  164. else:
  165. self._svpending = callback
  166. return callback
  167. self._lvpending.append(callback)
  168. return callback
  169. def throw1(self, exc=None):
  170. if not self.cancelled:
  171. exc = exc if exc is not None else sys.exc_info()[1]
  172. self.failed, self.reason = True, exc
  173. if self.on_error:
  174. self.on_error(*self.args + (exc,), **self.kwargs)
  175. def throw(self, exc=None, tb=None, propagate=True):
  176. if not self.cancelled:
  177. current_exc = sys.exc_info()[1]
  178. exc = exc if exc is not None else current_exc
  179. try:
  180. self.throw1(exc)
  181. svpending = self._svpending
  182. if svpending is not None:
  183. try:
  184. svpending.throw1(exc)
  185. finally:
  186. self._svpending = None
  187. else:
  188. lvpending = self._lvpending
  189. try:
  190. while lvpending:
  191. lvpending.popleft().throw1(exc)
  192. finally:
  193. self._lvpending = None
  194. finally:
  195. if self.on_error is None and propagate:
  196. if tb is None and (exc is None or exc is current_exc):
  197. raise
  198. reraise(type(exc), exc, tb)
  199. @property
  200. def listeners(self):
  201. if self._lvpending:
  202. return self._lvpending
  203. return [self._svpending]