未调用Tornado io_loop add_future回调

Question

我正在一个Tornado WebSocket服务器上工作，该服务器在收到消息时执行一些异步操作。我添加了一个带有io_loop.add_future的回调，以便在函数结束或引发异常时得到通知，但从未调用此回调。

这是正在发生的事情的一个例子：

import tornado
from tornado.httpclient import HTTPRequest
from tornado.web import Application
from tornado.websocket import websocket_connect
from tornado.testing import AsyncHTTPTestCase, gen_test
def message_processed_callback(*args, **kwargs):
    print 'Callback(args=%r, kwargs=%r)' % (args, kwargs)
class RealtimeHandler(tornado.websocket.WebSocketHandler):
    def initialize(self):
        self.io_loop = tornado.ioloop.IOLoop.instance()
    def on_message(self, message):
        future = self.on_some_message(message)
        print 'The future:', future
        self.io_loop.add_future(future, message_processed_callback)
    @tornado.gen.coroutine
    def on_some_message(self, message):
        print 'Before sleep'
        yield tornado.gen.sleep(3)
        print 'After sleep'
        self.write_message(message)
class ChatTestCase(AsyncHTTPTestCase):
    def get_app(self):
        return Application([
            ('/rt', RealtimeHandler),
        ])
    @gen_test
    def test_reply(self):
        request = HTTPRequest('ws://127.0.0.1:%d/rt' % self.get_http_port())
        ws = yield websocket_connect(request)
        ws.write_message('Hi')
        response = yield ws.read_message()
        print 'Response:', response

执行此测试的结果为：

Before sleep
The future: <tornado.concurrent.Future object at 0x4c01250>
[The sleep takes place here]
After sleep
Response: Hi

Answer 1

为了确保测试之间的隔离，每个测试用例都使用自己独立的IOLoop而不是IOLoop.instance()运行。可以使用IOLoop.current()访问测试的IOLoop（应该几乎总是使用IOLoop.instance()）。

如果由于库超出您的控制范围而要求使用IOLoop.instance()，则可以通过重写AsyncTestCase.get_new_ioloop来使测试使用该IOLoop。