concurrent_test.py 文件源码

def test_call_with_both(self):
        class Object(object):
            def __init__(self, io_loop):
                self._io_loop = io_loop
                self.__executor = futures.thread.ThreadPoolExecutor(1)

            @run_on_executor(io_loop='_io_loop', executor='_Object__executor')
            def f(self):
                return 42

        o = Object(io_loop=self.io_loop)
        answer = yield o.f()
        self.assertEqual(answer, 42)