forked from ipython/ipython
-
Notifications
You must be signed in to change notification settings - Fork 1
/
twistedutil.py
206 lines (164 loc) · 7.71 KB
/
twistedutil.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
#!/usr/bin/env python
# encoding: utf-8
"""Things directly related to all of twisted."""
__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
#-------------------------------------------------------------------------------
import threading, Queue, atexit
import twisted
from twisted.internet import defer, reactor
from twisted.python import log, failure
#-------------------------------------------------------------------------------
# Classes related to twisted and threads
#-------------------------------------------------------------------------------
class ReactorInThread(threading.Thread):
"""Run the twisted reactor in a different thread.
For the process to be able to exit cleanly, do the following:
rit = ReactorInThread()
rit.setDaemon(True)
rit.start()
"""
def run(self):
reactor.run(installSignalHandlers=0)
# self.join()
def stop(self):
# I don't think this does anything useful.
blockingCallFromThread(reactor.stop)
self.join()
if(twisted.version.major >= 8):
import twisted.internet.threads
def blockingCallFromThread(f, *a, **kw):
"""
Run a function in the reactor from a thread, and wait for the result
synchronously, i.e. until the callback chain returned by the function get a
result.
Delegates to twisted.internet.threads.blockingCallFromThread(reactor, f, *a, **kw),
passing twisted.internet.reactor for the first argument.
@param f: the callable to run in the reactor thread
@type f: any callable.
@param a: the arguments to pass to C{f}.
@param kw: the keyword arguments to pass to C{f}.
@return: the result of the callback chain.
@raise: any error raised during the callback chain.
"""
return twisted.internet.threads.blockingCallFromThread(reactor, f, *a, **kw)
else:
def blockingCallFromThread(f, *a, **kw):
"""
Run a function in the reactor from a thread, and wait for the result
synchronously, i.e. until the callback chain returned by the function get a
result.
@param f: the callable to run in the reactor thread
@type f: any callable.
@param a: the arguments to pass to C{f}.
@param kw: the keyword arguments to pass to C{f}.
@return: the result of the callback chain.
@raise: any error raised during the callback chain.
"""
from twisted.internet import reactor
queue = Queue.Queue()
def _callFromThread():
result = defer.maybeDeferred(f, *a, **kw)
result.addBoth(queue.put)
reactor.callFromThread(_callFromThread)
result = queue.get()
if isinstance(result, failure.Failure):
# This makes it easier for the debugger to get access to the instance
try:
result.raiseException()
except Exception, e:
raise e
return result
#-------------------------------------------------------------------------------
# Things for managing deferreds
#-------------------------------------------------------------------------------
def parseResults(results):
"""Pull out results/Failures from a DeferredList."""
return [x[1] for x in results]
def gatherBoth(dlist, fireOnOneCallback=0,
fireOnOneErrback=0,
consumeErrors=0,
logErrors=0):
"""This is like gatherBoth, but sets consumeErrors=1."""
d = DeferredList(dlist, fireOnOneCallback, fireOnOneErrback,
consumeErrors, logErrors)
if not fireOnOneCallback:
d.addCallback(parseResults)
return d
SUCCESS = True
FAILURE = False
class DeferredList(defer.Deferred):
"""I combine a group of deferreds into one callback.
I track a list of L{Deferred}s for their callbacks, and make a single
callback when they have all completed, a list of (success, result)
tuples, 'success' being a boolean.
Note that you can still use a L{Deferred} after putting it in a
DeferredList. For example, you can suppress 'Unhandled error in Deferred'
messages by adding errbacks to the Deferreds *after* putting them in the
DeferredList, as a DeferredList won't swallow the errors. (Although a more
convenient way to do this is simply to set the consumeErrors flag)
Note: This is a modified version of the twisted.internet.defer.DeferredList
"""
fireOnOneCallback = 0
fireOnOneErrback = 0
def __init__(self, deferredList, fireOnOneCallback=0, fireOnOneErrback=0,
consumeErrors=0, logErrors=0):
"""Initialize a DeferredList.
@type deferredList: C{list} of L{Deferred}s
@param deferredList: The list of deferreds to track.
@param fireOnOneCallback: (keyword param) a flag indicating that
only one callback needs to be fired for me to call
my callback
@param fireOnOneErrback: (keyword param) a flag indicating that
only one errback needs to be fired for me to call
my errback
@param consumeErrors: (keyword param) a flag indicating that any errors
raised in the original deferreds should be
consumed by this DeferredList. This is useful to
prevent spurious warnings being logged.
"""
self.resultList = [None] * len(deferredList)
defer.Deferred.__init__(self)
if len(deferredList) == 0 and not fireOnOneCallback:
self.callback(self.resultList)
# These flags need to be set *before* attaching callbacks to the
# deferreds, because the callbacks use these flags, and will run
# synchronously if any of the deferreds are already fired.
self.fireOnOneCallback = fireOnOneCallback
self.fireOnOneErrback = fireOnOneErrback
self.consumeErrors = consumeErrors
self.logErrors = logErrors
self.finishedCount = 0
index = 0
for deferred in deferredList:
deferred.addCallbacks(self._cbDeferred, self._cbDeferred,
callbackArgs=(index,SUCCESS),
errbackArgs=(index,FAILURE))
index = index + 1
def _cbDeferred(self, result, index, succeeded):
"""(internal) Callback for when one of my deferreds fires.
"""
self.resultList[index] = (succeeded, result)
self.finishedCount += 1
if not self.called:
if succeeded == SUCCESS and self.fireOnOneCallback:
self.callback((result, index))
elif succeeded == FAILURE and self.fireOnOneErrback:
# We have modified this to fire the errback chain with the actual
# Failure instance the originally occured rather than twisted's
# FirstError which wraps the failure
self.errback(result)
elif self.finishedCount == len(self.resultList):
self.callback(self.resultList)
if succeeded == FAILURE and self.logErrors:
log.err(result)
if succeeded == FAILURE and self.consumeErrors:
result = None
return result