python类run_on_executor()的实例源码-第2页-面圈网

concurrent_test.py 文件源码项目：PyQYT 作者: collinsctk 项目源码文件源码阅读 22 收藏 0 点赞 0 评论 0

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

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

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

concurrent_test.py 文件源码项目：PyQYT 作者: collinsctk 项目源码文件源码阅读 16 收藏 0 点赞 0 评论 0

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)

concurrent_test.py 文件源码项目：ProgrameFacil 作者: Gpzim98 项目源码文件源码阅读 20 收藏 0 点赞 0 评论 0

def test_no_calling(self):
        class Object(object):
            def __init__(self, io_loop):
                self.io_loop = io_loop
                self.executor = futures.thread.ThreadPoolExecutor(1)

            @run_on_executor
            def f(self):
                return 42

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

concurrent_test.py 文件源码项目：ProgrameFacil 作者: Gpzim98 项目源码文件源码阅读 29 收藏 0 点赞 0 评论 0

def test_call_with_no_args(self):
        class Object(object):
            def __init__(self, io_loop):
                self.io_loop = io_loop
                self.executor = futures.thread.ThreadPoolExecutor(1)

            @run_on_executor()
            def f(self):
                return 42

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

concurrent_test.py 文件源码项目：ProgrameFacil 作者: Gpzim98 项目源码文件源码阅读 21 收藏 0 点赞 0 评论 0

def test_call_with_io_loop(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')
            def f(self):
                return 42

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

concurrent_test.py 文件源码项目：ProgrameFacil 作者: Gpzim98 项目源码文件源码阅读 21 收藏 0 点赞 0 评论 0

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

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

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

concurrent_test.py 文件源码项目：ProgrameFacil 作者: Gpzim98 项目源码文件源码阅读 27 收藏 0 点赞 0 评论 0

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)

concurrent_test.py 文件源码项目：ProgrameFacil 作者: Gpzim98 项目源码文件源码阅读 19 收藏 0 点赞 0 评论 0

def test_call_with_no_args(self):
        class Object(object):
            def __init__(self, io_loop):
                self.io_loop = io_loop
                self.executor = futures.thread.ThreadPoolExecutor(1)

            @run_on_executor()
            def f(self):
                return 42

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

concurrent_test.py 文件源码项目：ProgrameFacil 作者: Gpzim98 项目源码文件源码阅读 25 收藏 0 点赞 0 评论 0

def test_call_with_io_loop(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')
            def f(self):
                return 42

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

concurrent_test.py 文件源码项目：ProgrameFacil 作者: Gpzim98 项目源码文件源码阅读 21 收藏 0 点赞 0 评论 0

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

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

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

concurrent_test.py 文件源码项目：ProgrameFacil 作者: Gpzim98 项目源码文件源码阅读 28 收藏 0 点赞 0 评论 0

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)

concurrent_test.py 文件源码项目：projects-2017-2 作者: ncss 项目源码文件源码阅读 20 收藏 0 点赞 0 评论 0

def test_call_with_io_loop(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')
            def f(self):
                return 42

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

concurrent_test.py 文件源码项目：projects-2017-2 作者: ncss 项目源码文件源码阅读 22 收藏 0 点赞 0 评论 0

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

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

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

concurrent_test.py 文件源码项目：projects-2017-2 作者: ncss 项目源码文件源码阅读 26 收藏 0 点赞 0 评论 0

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)

concurrent_test.py 文件源码项目：aweasome_learning 作者: Knight-ZXW 项目源码文件源码阅读 18 收藏 0 点赞 0 评论 0

def test_no_calling(self):
        class Object(object):
            def __init__(self, io_loop):
                self.io_loop = io_loop
                self.executor = futures.thread.ThreadPoolExecutor(1)

            @run_on_executor
            def f(self):
                return 42

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

concurrent_test.py 文件源码项目：aweasome_learning 作者: Knight-ZXW 项目源码文件源码阅读 22 收藏 0 点赞 0 评论 0

def test_call_with_no_args(self):
        class Object(object):
            def __init__(self, io_loop):
                self.io_loop = io_loop
                self.executor = futures.thread.ThreadPoolExecutor(1)

            @run_on_executor()
            def f(self):
                return 42

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

concurrent_test.py 文件源码项目：aweasome_learning 作者: Knight-ZXW 项目源码文件源码阅读 17 收藏 0 点赞 0 评论 0

def test_call_with_io_loop(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')
            def f(self):
                return 42

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

concurrent_test.py 文件源码项目：aweasome_learning 作者: Knight-ZXW 项目源码文件源码阅读 23 收藏 0 点赞 0 评论 0

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

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

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

concurrent_test.py 文件源码项目：aweasome_learning 作者: Knight-ZXW 项目源码文件源码阅读 23 收藏 0 点赞 0 评论 0

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)

concurrent_test.py 文件源码项目：zenchmarks 作者: squeaky-pl 项目源码文件源码阅读 20 收藏 0 点赞 0 评论 0

def test_no_calling(self):
        class Object(object):
            def __init__(self, io_loop):
                self.io_loop = io_loop
                self.executor = futures.thread.ThreadPoolExecutor(1)

            @run_on_executor
            def f(self):
                return 42

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