# License for the specific language governing permissions and limitations
# under the License.
-"""
-A twisted-style reactor for the Tornado IOLoop.
+# Note: This module's docs are not currently extracted automatically,
+# so changes must be made manually to twisted.rst
+# TODO: refactor doc build process to use an appropriate virtualenv
+"""A Twisted reactor built on the Tornado IOLoop.
+
+This module lets you run applications and libraries written for
+Twisted in a Tornado application. To use it, simply call `install` at
+the beginning of the application::
+
+ import tornado.platform.twisted
+ tornado.platform.twisted.install()
+ from twisted.internet import reactor
+
+When the app is ready to start, call `IOLoop.instance().start()`
+instead of `reactor.run()`. This will allow you to use a mixture of
+Twisted and Tornado code in the same process.
+
+It is also possible to create a non-global reactor by calling
+`tornado.platform.twisted.TornadoReactor(io_loop)`. However, if
+the `IOLoop` and reactor are to be short-lived (such as those used in
+unit tests), additional cleanup may be required. Specifically, it is
+recommended to call::
+
+ reactor.fireSystemEvent('shutdown')
+ reactor.disconnectAll()
-To use it, add the following to your twisted application:
+before closing the `IOLoop`.
-import tornado.platform.twisted
-tornado.platform.twisted.install()
-from twisted.internet import reactor
+This module has been tested with Twisted versions 11.0.0 and 11.1.0.
"""
from __future__ import with_statement, absolute_import
from twisted.internet.posixbase import PosixReactorBase
from twisted.internet.interfaces import \
IReactorFDSet, IDelayedCall, IReactorTime
-from twisted.python import failure
+from twisted.python import failure, log
from twisted.internet import error
from zope.interface import implements
class TornadoDelayedCall(object):
- """
- DelayedCall object for Tornado.
- """
+ """DelayedCall object for Tornado."""
implements(IDelayedCall)
def __init__(self, reactor, seconds, f, *args, **kw):
return self._active
class TornadoReactor(PosixReactorBase):
- """
- Twisted style reactor for Tornado.
+ """Twisted reactor built on the Tornado IOLoop.
+
+ Since it is intented to be used in applications where the top-level
+ event loop is ``io_loop.start()`` rather than ``reactor.run()``,
+ it is implemented a little differently than other Twisted reactors.
+ We override `mainLoop` instead of `doIteration` and must implement
+ timed call functionality on top of `IOLoop.add_timeout` rather than
+ using the implementation in `PosixReactorBase`.
"""
implements(IReactorTime, IReactorFDSet)
if not io_loop:
io_loop = tornado.ioloop.IOLoop.instance()
self._io_loop = io_loop
- self._readers = {}
- self._writers = {}
+ self._readers = {} # map of reader objects to fd
+ self._writers = {} # map of writer objects to fd
self._fds = {} # a map of fd to a (reader, writer) tuple
self._delayedCalls = {}
- self._running = False
- self._closed = False
PosixReactorBase.__init__(self)
+ # IOLoop.start() bypasses some of the reactor initialization.
+ # Fire off the necessary events if they weren't already triggered
+ # by reactor.run().
+ def start_if_necessary():
+ if not self._started:
+ self.fireSystemEvent('startup')
+ self._io_loop.add_callback(start_if_necessary)
+
# IReactorTime
def seconds(self):
return time.time()
# IReactorThreads
def callFromThread(self, f, *args, **kw):
- """
- See L{twisted.internet.interfaces.IReactorThreads.callFromThread}.
- """
+ """See `twisted.internet.interfaces.IReactorThreads.callFromThread`"""
assert callable(f), "%s is not callable" % f
p = functools.partial(f, *args, **kw)
self._io_loop.add_callback(p)
# IReactorFDSet
def _invoke_callback(self, fd, events):
(reader, writer) = self._fds[fd]
- if events & IOLoop.READ and reader:
- reader.doRead()
- if events & IOLoop.WRITE and writer:
- writer.doWrite()
- if events & IOLoop.ERROR:
- if reader:
- reader.readConnectionLost(failure.Failure(error.ConnectionLost()))
- if writer:
- writer.connectionLost(failure.Failure(error.ConnectionLost()))
+ if reader:
+ err = None
+ if reader.fileno() == -1:
+ err = error.ConnectionLost()
+ elif events & IOLoop.READ:
+ err = log.callWithLogger(reader, reader.doRead)
+ if err is None and events & IOLoop.ERROR:
+ err = error.ConnectionLost()
+ if err is not None:
+ self.removeReader(reader)
+ reader.readConnectionLost(failure.Failure(err))
+ if writer:
+ err = None
+ if writer.fileno() == -1:
+ err = error.ConnectionLost()
+ elif events & IOLoop.WRITE:
+ err = log.callWithLogger(writer, writer.doWrite)
+ if err is None and events & IOLoop.ERROR:
+ err = error.ConnectionLost()
+ if err is not None:
+ self.removeWriter(writer)
+ writer.writeConnectionLost(failure.Failure(err))
def addReader(self, reader):
- """
- Add a FileDescriptor for notification of data available to read.
- """
+ """Add a FileDescriptor for notification of data available to read."""
if reader in self._readers:
# Don't add the reader if it's already there
return
- self._readers[reader] = True
fd = reader.fileno()
+ self._readers[reader] = fd
if fd in self._fds:
(_, writer) = self._fds[fd]
self._fds[fd] = (reader, writer)
IOLoop.READ)
def addWriter(self, writer):
- """
- Add a FileDescriptor for notification of data available to write.
- """
+ """Add a FileDescriptor for notification of data available to write."""
if writer in self._writers:
return
- self._writers[writer] = True
fd = writer.fileno()
+ self._writers[writer] = fd
if fd in self._fds:
(reader, _) = self._fds[fd]
self._fds[fd] = (reader, writer)
IOLoop.WRITE)
def removeReader(self, reader):
- """
- Remove a Selectable for notification of data available to read.
- """
- fd = reader.fileno()
+ """Remove a Selectable for notification of data available to read."""
if reader in self._readers:
- del self._readers[reader]
- if self._closed: return
+ fd = self._readers.pop(reader)
(_, writer) = self._fds[fd]
if writer:
# We have a writer so we need to update the IOLoop for
self._io_loop.remove_handler(fd)
def removeWriter(self, writer):
- """
- Remove a Selectable for notification of data available to write.
- """
- fd = writer.fileno()
+ """Remove a Selectable for notification of data available to write."""
if writer in self._writers:
- del self._writers[writer]
- if self._closed: return
+ fd = self._writers.pop(writer)
(reader, _) = self._fds[fd]
if reader:
# We have a reader so we need to update the IOLoop for
def getWriters(self):
return self._writers.keys()
+ # The following functions are mainly used in twisted-style test cases;
+ # it is expected that most users of the TornadoReactor will call
+ # IOLoop.start() instead of Reactor.run().
def stop(self):
- """
- Implement L{IReactorCore.stop}.
- """
- self._running = False
PosixReactorBase.stop(self)
- self.runUntilCurrent()
- try:
- self._io_loop.stop()
- self._io_loop.close()
- except:
- # Ignore any exceptions thrown by IOLoop
- pass
- self._closed = True
+ self._io_loop.stop()
def crash(self):
- if not self._running:
- return
- self._running = False
PosixReactorBase.crash(self)
- self.runUntilCurrent()
- try:
- self._io_loop.stop()
- self._io_loop.close()
- except:
- # Ignore any exceptions thrown by IOLoop
- pass
- self._closed = True
+ self._io_loop.stop()
def doIteration(self, delay):
raise NotImplementedError("doIteration")
def mainLoop(self):
- self._running = True
self._io_loop.start()
+ if self._stopped:
+ self.fireSystemEvent("shutdown")
class _TestReactor(TornadoReactor):
"""Subclass of TornadoReactor for use in unittests.
This can't go in the test.py file because of import-order dependencies
- with the twisted reactor test builder.
+ with the Twisted reactor test builder.
"""
def __init__(self):
# always use a new ioloop
return super(_TestReactor, self).listenTCP(
port, factory, backlog=backlog, interface=interface)
+ def listenUDP(self, port, protocol, interface='', maxPacketSize=8192):
+ if not interface:
+ interface = '127.0.0.1'
+ return super(_TestReactor, self).listenUDP(
+ port, protocol, interface=interface, maxPacketSize=maxPacketSize)
+
def install(io_loop=None):
- """
- Install the Tornado reactor.
- """
+ """Install this package as the default Twisted reactor."""
if not io_loop:
io_loop = tornado.ioloop.IOLoop.instance()
reactor = TornadoReactor(io_loop)
try:
import fcntl
import twisted
+ from twisted.internet.defer import Deferred
from twisted.internet.interfaces import IReadDescriptor, IWriteDescriptor
+ from twisted.internet.protocol import Protocol
+ from twisted.web.client import Agent
+ from twisted.web.resource import Resource
+ from twisted.web.server import Site
+ from twisted.python import log
from tornado.platform.twisted import TornadoReactor
from zope.interface import implements
except ImportError:
IReadDescriptor = IWriteDescriptor = None
def implements(f): pass
+from tornado.httpclient import AsyncHTTPClient
from tornado.ioloop import IOLoop
from tornado.platform.auto import set_close_exec
+from tornado.testing import get_unused_port
from tornado.util import import_object
+from tornado.web import RequestHandler, Application
-class ReactorWhenRunningTest(unittest.TestCase):
+class ReactorTestCase(unittest.TestCase):
def setUp(self):
- self._reactor = TornadoReactor(IOLoop())
+ self._io_loop = IOLoop()
+ self._reactor = TornadoReactor(self._io_loop)
+ def tearDown(self):
+ self._io_loop.close(all_fds=True)
+
+class ReactorWhenRunningTest(ReactorTestCase):
def test_whenRunning(self):
self._whenRunningCalled = False
self._anotherWhenRunningCalled = False
def anotherWhenRunningCallback(self):
self._anotherWhenRunningCalled = True
-class ReactorCallLaterTest(unittest.TestCase):
- def setUp(self):
- self._reactor = TornadoReactor(IOLoop())
-
+class ReactorCallLaterTest(ReactorTestCase):
def test_callLater(self):
self._laterCalled = False
self._now = self._reactor.seconds()
self._called = self._reactor.seconds()
self._reactor.stop()
-class ReactorTwoCallLaterTest(unittest.TestCase):
- def setUp(self):
- self._reactor = TornadoReactor(IOLoop())
-
+class ReactorTwoCallLaterTest(ReactorTestCase):
def test_callLater(self):
self._later1Called = False
self._later2Called = False
self._called2 = self._reactor.seconds()
self._reactor.stop()
-class ReactorCallFromThreadTest(unittest.TestCase):
+class ReactorCallFromThreadTest(ReactorTestCase):
def setUp(self):
- self._reactor = TornadoReactor(IOLoop())
+ super(ReactorCallFromThreadTest, self).setUp()
self._mainThread = thread.get_ident()
def tearDown(self):
self._thread.join()
+ super(ReactorCallFromThreadTest, self).tearDown()
def _newThreadRun(self):
self.assertNotEqual(self._mainThread, thread.get_ident())
self._reactor.callWhenRunning(self._whenRunningCallback)
self._reactor.run()
-class ReactorCallInThread(unittest.TestCase):
+class ReactorCallInThread(ReactorTestCase):
def setUp(self):
- self._reactor = TornadoReactor(IOLoop())
+ super(ReactorCallInThread, self).setUp()
self._mainThread = thread.get_ident()
def _fnCalledInThread(self, *args, **kwargs):
def doWrite(self):
self._callback(self._fd)
-class ReactorReaderWriterTest(unittest.TestCase):
+class ReactorReaderWriterTest(ReactorTestCase):
def _set_nonblocking(self, fd):
flags = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
def setUp(self):
- self._reactor = TornadoReactor(IOLoop())
+ super(ReactorReaderWriterTest, self).setUp()
r, w = os.pipe()
self._set_nonblocking(r)
self._set_nonblocking(w)
self._p1 = os.fdopen(r, "rb", 0)
self._p2 = os.fdopen(w, "wb", 0)
+ def tearDown(self):
+ super(ReactorReaderWriterTest, self).tearDown()
+ self._p1.close()
+ self._p2.close()
+
def _testReadWrite(self):
"""
In this test the writer writes an 'x' to its fd. The reader
self._reactor.callWhenRunning(self._testNoWriter)
self._reactor.run()
+# Test various combinations of twisted and tornado http servers,
+# http clients, and event loop interfaces.
+class CompatibilityTests(unittest.TestCase):
+ def setUp(self):
+ self.io_loop = IOLoop()
+ self.reactor = TornadoReactor(self.io_loop)
+
+ def tearDown(self):
+ self.reactor.disconnectAll()
+ self.io_loop.close(all_fds=True)
+
+ def start_twisted_server(self):
+ class HelloResource(Resource):
+ isLeaf = True
+ def render_GET(self, request):
+ return "Hello from twisted!"
+ site = Site(HelloResource())
+ self.twisted_port = get_unused_port()
+ self.reactor.listenTCP(self.twisted_port, site, interface='127.0.0.1')
+
+ def start_tornado_server(self):
+ class HelloHandler(RequestHandler):
+ def get(self):
+ self.write("Hello from tornado!")
+ app = Application([('/', HelloHandler)],
+ log_function=lambda x: None)
+ self.tornado_port = get_unused_port()
+ app.listen(self.tornado_port, address='127.0.0.1', io_loop=self.io_loop)
+
+ def run_ioloop(self):
+ self.stop_loop = self.io_loop.stop
+ self.io_loop.start()
+ self.reactor.fireSystemEvent('shutdown')
+
+ def run_reactor(self):
+ self.stop_loop = self.reactor.stop
+ self.stop = self.reactor.stop
+ self.reactor.run()
+
+ def tornado_fetch(self, url, runner):
+ responses = []
+ client = AsyncHTTPClient(self.io_loop)
+ def callback(response):
+ responses.append(response)
+ self.stop_loop()
+ client.fetch(url, callback=callback)
+ runner()
+ self.assertEqual(len(responses), 1)
+ responses[0].rethrow()
+ return responses[0]
+
+ def twisted_fetch(self, url, runner):
+ # http://twistedmatrix.com/documents/current/web/howto/client.html
+ chunks = []
+ client = Agent(self.reactor)
+ d = client.request('GET', url)
+ class Accumulator(Protocol):
+ def __init__(self, finished):
+ self.finished = finished
+ def dataReceived(self, data):
+ chunks.append(data)
+ def connectionLost(self, reason):
+ self.finished.callback(None)
+ def callback(response):
+ finished = Deferred()
+ response.deliverBody(Accumulator(finished))
+ return finished
+ d.addCallback(callback)
+ def shutdown(ignored):
+ self.stop_loop()
+ d.addBoth(shutdown)
+ runner()
+ self.assertTrue(chunks)
+ return ''.join(chunks)
+
+ def testTwistedServerTornadoClientIOLoop(self):
+ self.start_twisted_server()
+ response = self.tornado_fetch(
+ 'http://localhost:%d' % self.twisted_port, self.run_ioloop)
+ self.assertEqual(response.body, 'Hello from twisted!')
+
+ def testTwistedServerTornadoClientReactor(self):
+ self.start_twisted_server()
+ response = self.tornado_fetch(
+ 'http://localhost:%d' % self.twisted_port, self.run_reactor)
+ self.assertEqual(response.body, 'Hello from twisted!')
+
+ def testTornadoServerTwistedClientIOLoop(self):
+ self.start_tornado_server()
+ response = self.twisted_fetch(
+ 'http://localhost:%d' % self.tornado_port, self.run_ioloop)
+ self.assertEqual(response, 'Hello from tornado!')
+
+ def testTornadoServerTwistedClientReactor(self):
+ self.start_tornado_server()
+ response = self.twisted_fetch(
+ 'http://localhost:%d' % self.tornado_port, self.run_reactor)
+ self.assertEqual(response, 'Hello from tornado!')
+
+
if twisted is None:
del ReactorWhenRunningTest
del ReactorCallLaterTest
del ReactorCallFromThreadTest
del ReactorCallInThread
del ReactorReaderWriterTest
+ del CompatibilityTests
else:
# Import and run as much of twisted's test suite as possible.
# This is unfortunately rather dependent on implementation details,
# but there doesn't appear to be a clean all-in-one conformance test
# suite for reactors.
+ #
# This is a list of all test suites using the ReactorBuilder
- # available in Twisted 11.0.0. Tests that do not currently pass
- # with the TornadoReactor are commented out.
- twisted_tests = [
- 'twisted.internet.test.test_core.ObjectModelIntegrationTest',
- #'twisted.internet.test.test_core.SystemEventTestsBuilder',
- 'twisted.internet.test.test_fdset.ReactorFDSetTestsBuilder',
- #'twisted.internet.test.test_process.ProcessTestsBuilder',
- #'twisted.internet.test.test_process.PTYProcessTestsBuilder',
- #'twisted.internet.test.test_tcp.TCPClientTestsBuilder',
- 'twisted.internet.test.test_tcp.TCPPortTestsBuilder',
- 'twisted.internet.test.test_tcp.TCPConnectionTestsBuilder',
- 'twisted.internet.test.test_threads.ThreadTestsBuilder',
- 'twisted.internet.test.test_time.TimeTestsBuilder',
- #'twisted.internet.test.test_tls.SSLClientTestsMixin',
- 'twisted.internet.test.test_udp.UDPServerTestsBuilder',
- #'twisted.internet.test.test_unix.UNIXTestsBuilder',
- #'twisted.internet.test.test_unix.UNIXDatagramTestsBuilder',
- ]
- for test_name in twisted_tests:
+ # available in Twisted 11.0.0 and 11.1.0 (and a blacklist of
+ # specific test methods to be disabled).
+ twisted_tests = {
+ 'twisted.internet.test.test_core.ObjectModelIntegrationTest': [],
+ 'twisted.internet.test.test_core.SystemEventTestsBuilder': [
+ 'test_iterate', # deliberately not supported
+ ],
+ 'twisted.internet.test.test_fdset.ReactorFDSetTestsBuilder': [
+ "test_lostFileDescriptor", # incompatible with epoll and kqueue
+ ],
+ 'twisted.internet.test.test_process.ProcessTestsBuilder': [
+ # Doesn't work on python 2.5
+ 'test_systemCallUninterruptedByChildExit',
+ # Doesn't clean up its temp files
+ 'test_shebang',
+ ],
+ 'twisted.internet.test.test_process.PTYProcessTestsBuilder': [
+ 'test_systemCallUninterruptedByChildExit',
+ ],
+ 'twisted.internet.test.test_tcp.TCPClientTestsBuilder': [],
+ 'twisted.internet.test.test_tcp.TCPPortTestsBuilder': [],
+ 'twisted.internet.test.test_tcp.TCPConnectionTestsBuilder': [],
+ 'twisted.internet.test.test_tcp.WriteSequenceTests': [],
+ 'twisted.internet.test.test_tcp.AbortConnectionTestCase': [],
+ 'twisted.internet.test.test_threads.ThreadTestsBuilder': [],
+ 'twisted.internet.test.test_time.TimeTestsBuilder': [],
+ # Extra third-party dependencies (pyOpenSSL)
+ #'twisted.internet.test.test_tls.SSLClientTestsMixin': [],
+ 'twisted.internet.test.test_udp.UDPServerTestsBuilder': [],
+ 'twisted.internet.test.test_unix.UNIXTestsBuilder': [
+ # Platform-specific. These tests would be skipped automatically
+ # if we were running twisted's own test runner.
+ 'test_connectToLinuxAbstractNamespace',
+ 'test_listenOnLinuxAbstractNamespace',
+ ],
+ 'twisted.internet.test.test_unix.UNIXDatagramTestsBuilder': [
+ 'test_listenOnLinuxAbstractNamespace',
+ ],
+ 'twisted.internet.test.test_unix.UNIXPortTestsBuilder': [],
+ }
+ for test_name, blacklist in twisted_tests.iteritems():
try:
- test = import_object(test_name)
+ test_class = import_object(test_name)
except (ImportError, AttributeError):
continue
- class TornadoTest(test):
- _reactors = ["tornado.platform.twisted._TestReactor"]
- TornadoTest.__name__ = test.__name__
- globals().update(TornadoTest.makeTestCaseClasses())
+ for test_func in blacklist:
+ if hasattr(test_class, test_func):
+ # The test_func may be defined in a mixin, so clobber
+ # it instead of delattr()
+ setattr(test_class, test_func, lambda self: None)
+ def make_test_subclass(test_class):
+ class TornadoTest(test_class):
+ _reactors = ["tornado.platform.twisted._TestReactor"]
+ def unbuildReactor(self, reactor):
+ test_class.unbuildReactor(self, reactor)
+ # Clean up file descriptors (especially epoll/kqueue
+ # objects) eagerly instead of leaving them for the
+ # GC. Unfortunately we can't do this in reactor.stop
+ # since twisted expects to be able to unregister
+ # connections in a post-shutdown hook.
+ reactor._io_loop.close(all_fds=True)
+ TornadoTest.__name__ = test_class.__name__
+ return TornadoTest
+ test_subclass = make_test_subclass(test_class)
+ globals().update(test_subclass.makeTestCaseClasses())
+
+ # Since we're not using twisted's test runner, it's tricky to get
+ # logging set up well. Most of the time it's easiest to just
+ # leave it turned off, but while working on these tests you may want
+ # to uncomment one of the other lines instead.
+ log.defaultObserver.stop()
+ #import sys; log.startLogging(sys.stderr, setStdout=0)
+ #log.startLoggingWithObserver(log.PythonLoggingObserver().emit, setStdout=0)
if __name__ == "__main__":
unittest.main()
projects / thirdparty / tornado.git / commitdiff
