Python asyncio 模块，coroutine() 实例源码

def on(self, event, callback, selector=None):
        """
        Wrapper around :meth:`~.application.Application.register`.
        If :meth:`~.application.View.on` is called, for instance, during :meth:`~.application.View.build`,
        the event handlers will be enqueued and registered when the view is loaded. Similarly,
        if :meth:`~.application.View.on` is called once the view is loaded (for example, in a button callback),
        the event handler will be registered immediately.

        :param event: Name of the event to monitor
        :param callback: Callback function for when the event is received (Params: event, interface).
        :param selector: `(Optional)` CSS selector for the element(s) you want to monitor
        """
        cbk = asyncio.coroutine(callback)
        self._event_cache.append(
            {'event': event, 'callback': cbk, 'selector': selector})
        if self.is_loaded:
            self.context.register(event, cbk, selector)

def get_event(self):
        """Fetch an event.

        This coroutine will swallow events for removed watches.
        """
        while True:
            prefix = yield from self._stream.readexactly(PREFIX.size)
            if prefix == b'':
                # We got closed, return None.
                return
            wd, flags, cookie, length = PREFIX.unpack(prefix)
            path = yield from self._stream.readexactly(length)

            # All async performed, time to look at the event's content.
            if wd not in self.aliases:
                # Event for a removed watch, skip it.
                continue

            decoded_path = struct.unpack('%ds' % length, path)[0].rstrip(b'\x00').decode('utf-8')
            return Event(
                flags=flags,
                cookie=cookie,
                name=decoded_path,
                alias=self.aliases[wd],
            )

def test_poll(self):
        @asyncio.coroutine
        def test():
            a, b = self.create_bound_pair(zmq.PUSH, zmq.PULL)
            f = b.poll(timeout=0)
            yield from asyncio.sleep(0)
            self.assertEqual(f.result(), 0)

            f = b.poll(timeout=1)
            assert not f.done()
            evt = yield from f

            self.assertEqual(evt, 0)

            f = b.poll(timeout=1000)
            assert not f.done()
            yield from a.send_multipart([b'hi', b'there'])
            evt = yield from f
            self.assertEqual(evt, zmq.POLLIN)
            recvd = yield from b.recv_multipart()
            self.assertEqual(recvd, [b'hi', b'there'])
        self.loop.run_until_complete(test())

def generate_request_url(self, scopes: tuple):
        """Generate OAuth request url.

        :param scopes: github access scopes
                       (https://developer.github.com/v3/oauth/#scopes)
        :param callback: callback to be called when user authorize request
                         (may be coroutine)
        """
        state = base64.b64encode(os.urandom(15)).decode("ascii")
        self._requested_scopes[state] = scopes
        qs = parse.urlencode({
            "client_id": self._client_id,
            "scope": ",".join(scopes),
            "state": state,
        })
        return "%s?%s" % (REQ_ACCESS_URL, qs)

def test_put_cancelled(self):
        q = asyncio.Queue(loop=self.loop)

        @asyncio.coroutine
        def queue_put():
            yield from q.put(1)
            return True

        @asyncio.coroutine
        def test():
            return (yield from q.get())

        t = asyncio.Task(queue_put(), loop=self.loop)
        self.assertEqual(1, self.loop.run_until_complete(test()))
        self.assertTrue(t.done())
        self.assertTrue(t.result())

def test_subprocess_shell_invalid_args(self):
        @asyncio.coroutine
        def connect(cmd=None, **kwds):
            if not cmd:
                cmd = 'pwd'
            yield from self.loop.subprocess_shell(
                asyncio.SubprocessProtocol,
                cmd, **kwds)

        with self.assertRaises(ValueError):
            self.loop.run_until_complete(connect(['ls', '-l']))
        with self.assertRaises(ValueError):
            self.loop.run_until_complete(connect(universal_newlines=True))
        with self.assertRaises(ValueError):
            self.loop.run_until_complete(connect(bufsize=4096))
        with self.assertRaises(ValueError):
            self.loop.run_until_complete(connect(shell=False))

def test_context_manager_cant_reuse(self):
        lock = asyncio.Lock(loop=self.loop)

        @asyncio.coroutine
        def acquire_lock():
            return (yield from lock)

        # This spells "yield from lock" outside a generator.
        cm = self.loop.run_until_complete(acquire_lock())
        with cm:
            self.assertTrue(lock.locked())

        self.assertFalse(lock.locked())

        with self.assertRaises(AttributeError):
            with cm:
                pass

def test_semaphore(self):
        sem = asyncio.Semaphore(loop=self.loop)
        self.assertEqual(1, sem._value)

        @asyncio.coroutine
        def acquire_lock():
            return (yield from sem)

        res = self.loop.run_until_complete(acquire_lock())

        self.assertTrue(res)
        self.assertTrue(sem.locked())
        self.assertEqual(0, sem._value)

        sem.release()
        self.assertFalse(sem.locked())
        self.assertEqual(1, sem._value)

def test_communicate(self):
        args = PROGRAM_CAT

        @asyncio.coroutine
        def run(data):
            proc = yield from asyncio.create_subprocess_exec(
                                          *args,
                                          stdin=subprocess.PIPE,
                                          stdout=subprocess.PIPE,
                                          loop=self.loop)
            stdout, stderr = yield from proc.communicate(data)
            return proc.returncode, stdout

        task = run(b'some data')
        task = asyncio.wait_for(task, 60.0, loop=self.loop)
        exitcode, stdout = self.loop.run_until_complete(task)
        self.assertEqual(exitcode, 0)
        self.assertEqual(stdout, b'some data')

def test_send_signal(self):
        code = 'import time; print("sleeping", flush=True); time.sleep(3600)'
        args = [sys.executable, '-c', code]
        create = asyncio.create_subprocess_exec(*args,
                                                stdout=subprocess.PIPE,
                                                loop=self.loop)
        proc = self.loop.run_until_complete(create)

        @asyncio.coroutine
        def send_signal(proc):
            # basic synchronization to wait until the program is sleeping
            line = yield from proc.stdout.readline()
            self.assertEqual(line, b'sleeping\n')

            proc.send_signal(signal.SIGHUP)
            returncode = (yield from proc.wait())
            return returncode

        returncode = self.loop.run_until_complete(send_signal(proc))
        self.assertEqual(-signal.SIGHUP, returncode)

def test_stdin_not_inheritable(self):
        # Tulip issue #209: stdin must not be inheritable, otherwise
        # the Process.communicate() hangs
        @asyncio.coroutine
        def len_message(message):
            code = 'import sys; data = sys.stdin.read(); print(len(data))'
            proc = yield from asyncio.create_subprocess_exec(
                                          sys.executable, '-c', code,
                                          stdin=asyncio.subprocess.PIPE,
                                          stdout=asyncio.subprocess.PIPE,
                                          stderr=asyncio.subprocess.PIPE,
                                          close_fds=False,
                                          loop=self.loop)
            stdout, stderr = yield from proc.communicate(message)
            exitcode = yield from proc.wait()
            return (stdout, exitcode)

        output, exitcode = self.loop.run_until_complete(len_message(b'abc'))
        self.assertEqual(output.rstrip(), b'3')
        self.assertEqual(exitcode, 0)

def test_cancel_make_subprocess_transport_exec(self):
        @asyncio.coroutine
        def cancel_make_transport():
            coro = asyncio.create_subprocess_exec(*PROGRAM_BLOCKED,
                                                  loop=self.loop)
            task = self.loop.create_task(coro)

            self.loop.call_soon(task.cancel)
            try:
                yield from task
            except asyncio.CancelledError:
                pass

        # ignore the log:
        # "Exception during subprocess creation, kill the subprocess"
        with test_utils.disable_logger():
            self.loop.run_until_complete(cancel_make_transport())

def test_cancel_post_init(self):
        @asyncio.coroutine
        def cancel_make_transport():
            coro = self.loop.subprocess_exec(asyncio.SubprocessProtocol,
                                             *PROGRAM_BLOCKED)
            task = self.loop.create_task(coro)

            self.loop.call_soon(task.cancel)
            try:
                yield from task
            except asyncio.CancelledError:
                pass

        # ignore the log:
        # "Exception during subprocess creation, kill the subprocess"
        with test_utils.disable_logger():
            self.loop.run_until_complete(cancel_make_transport())
            test_utils.run_briefly(self.loop)

def test_create_task(self):
        class MyTask(asyncio.Task):
            pass

        @asyncio.coroutine
        def test():
            pass

        class EventLoop(base_events.BaseEventLoop):
            def create_task(self, coro):
                return MyTask(coro, loop=loop)

        loop = EventLoop()
        self.set_event_loop(loop)

        coro = test()
        task = asyncio.async(coro, loop=loop)
        self.assertIsInstance(task, MyTask)

        # make warnings quiet
        task._log_destroy_pending = False
        coro.close()

def test_run_forever_keyboard_interrupt(self):
        # Python issue #22601: ensure that the temporary task created by
        # run_forever() consumes the KeyboardInterrupt and so don't log
        # a warning
        @asyncio.coroutine
        def raise_keyboard_interrupt():
            raise KeyboardInterrupt

        self.loop._process_events = mock.Mock()
        self.loop.call_exception_handler = mock.Mock()

        try:
            self.loop.run_until_complete(raise_keyboard_interrupt())
        except KeyboardInterrupt:
            pass
        self.loop.close()
        support.gc_collect()

        self.assertFalse(self.loop.call_exception_handler.called)

def test_create_connection_connect_err(self):
        @asyncio.coroutine
        def getaddrinfo(*args, **kw):
            yield from []
            return [(2, 1, 6, '', ('107.6.106.82', 80))]

        def getaddrinfo_task(*args, **kwds):
            return asyncio.Task(getaddrinfo(*args, **kwds), loop=self.loop)

        self.loop.getaddrinfo = getaddrinfo_task
        self.loop.sock_connect = mock.Mock()
        self.loop.sock_connect.side_effect = OSError

        coro = self.loop.create_connection(MyProto, 'example.com', 80)
        self.assertRaises(
            OSError, self.loop.run_until_complete, coro)

def test_create_connection_multiple(self):
        @asyncio.coroutine
        def getaddrinfo(*args, **kw):
            return [(2, 1, 6, '', ('0.0.0.1', 80)),
                    (2, 1, 6, '', ('0.0.0.2', 80))]

        def getaddrinfo_task(*args, **kwds):
            return asyncio.Task(getaddrinfo(*args, **kwds), loop=self.loop)

        self.loop.getaddrinfo = getaddrinfo_task
        self.loop.sock_connect = mock.Mock()
        self.loop.sock_connect.side_effect = OSError

        coro = self.loop.create_connection(
            MyProto, 'example.com', 80, family=socket.AF_INET)
        with self.assertRaises(OSError):
            self.loop.run_until_complete(coro)

def test_create_server_empty_host(self):
        # if host is empty string use None instead
        host = object()

        @asyncio.coroutine
        def getaddrinfo(*args, **kw):
            nonlocal host
            host = args[0]
            yield from []

        def getaddrinfo_task(*args, **kwds):
            return asyncio.Task(getaddrinfo(*args, **kwds), loop=self.loop)

        self.loop.getaddrinfo = getaddrinfo_task
        fut = self.loop.create_server(MyProto, '', 0)
        self.assertRaises(OSError, self.loop.run_until_complete, fut)
        self.assertIsNone(host)

def test_call_coroutine(self):
        @asyncio.coroutine
        def simple_coroutine():
            pass

        coro_func = simple_coroutine
        coro_obj = coro_func()
        self.addCleanup(coro_obj.close)
        for func in (coro_func, coro_obj):
            with self.assertRaises(TypeError):
                self.loop.call_soon(func)
            with self.assertRaises(TypeError):
                self.loop.call_soon_threadsafe(func)
            with self.assertRaises(TypeError):
                self.loop.call_later(60, func)
            with self.assertRaises(TypeError):
                self.loop.call_at(self.loop.time() + 60, func)
            with self.assertRaises(TypeError):
                self.loop.run_in_executor(None, func)

def test_add_signal_handler_coroutine_error(self, m_signal):
        m_signal.NSIG = signal.NSIG

        @asyncio.coroutine
        def simple_coroutine():
            yield from []

        # callback must not be a coroutine function
        coro_func = simple_coroutine
        coro_obj = coro_func()
        self.addCleanup(coro_obj.close)
        for func in (coro_func, coro_obj):
            self.assertRaisesRegex(
                TypeError, 'coroutines cannot be used with add_signal_handler',
                self.loop.add_signal_handler,
                signal.SIGINT, func)

def test_task_class(self):
        @asyncio.coroutine
        def notmuch():
            return 'ok'
        t = asyncio.Task(notmuch(), loop=self.loop)
        self.loop.run_until_complete(t)
        self.assertTrue(t.done())
        self.assertEqual(t.result(), 'ok')
        self.assertIs(t._loop, self.loop)

        loop = asyncio.new_event_loop()
        self.set_event_loop(loop)
        t = asyncio.Task(notmuch(), loop=loop)
        self.assertIs(t._loop, loop)
        loop.run_until_complete(t)
        loop.close()

def test_async_task(self):
        @asyncio.coroutine
        def notmuch():
            return 'ok'
        t_orig = asyncio.Task(notmuch(), loop=self.loop)
        t = asyncio.async(t_orig)
        self.loop.run_until_complete(t)
        self.assertTrue(t.done())
        self.assertEqual(t.result(), 'ok')
        self.assertIs(t, t_orig)

        loop = asyncio.new_event_loop()
        self.set_event_loop(loop)

        with self.assertRaises(ValueError):
            t = asyncio.async(t_orig, loop=loop)

        loop.close()

        t = asyncio.async(t_orig, loop=self.loop)
        self.assertIs(t, t_orig)

def test_task_basics(self):
        @asyncio.coroutine
        def outer():
            a = yield from inner1()
            b = yield from inner2()
            return a+b

        @asyncio.coroutine
        def inner1():
            return 42

        @asyncio.coroutine
        def inner2():
            return 1000

        t = outer()
        self.assertEqual(self.loop.run_until_complete(t), 1042)

def test_cancel(self):

        def gen():
            when = yield
            self.assertAlmostEqual(10.0, when)
            yield 0

        loop = self.new_test_loop(gen)

        @asyncio.coroutine
        def task():
            yield from asyncio.sleep(10.0, loop=loop)
            return 12

        t = asyncio.Task(task(), loop=loop)
        loop.call_soon(t.cancel)
        with self.assertRaises(asyncio.CancelledError):
            loop.run_until_complete(t)
        self.assertTrue(t.done())
        self.assertTrue(t.cancelled())
        self.assertFalse(t.cancel())

def test_cancel_both_task_and_inner_future(self):
        f = asyncio.Future(loop=self.loop)

        @asyncio.coroutine
        def task():
            yield from f
            return 12

        t = asyncio.Task(task(), loop=self.loop)
        test_utils.run_briefly(self.loop)

        f.cancel()
        t.cancel()

        with self.assertRaises(asyncio.CancelledError):
            self.loop.run_until_complete(t)

        self.assertTrue(t.done())
        self.assertTrue(f.cancelled())
        self.assertTrue(t.cancelled())

def test_cancel_task_catching(self):
        fut1 = asyncio.Future(loop=self.loop)
        fut2 = asyncio.Future(loop=self.loop)

        @asyncio.coroutine
        def task():
            yield from fut1
            try:
                yield from fut2
            except asyncio.CancelledError:
                return 42

        t = asyncio.Task(task(), loop=self.loop)
        test_utils.run_briefly(self.loop)
        self.assertIs(t._fut_waiter, fut1)  # White-box test.
        fut1.set_result(None)
        test_utils.run_briefly(self.loop)
        self.assertIs(t._fut_waiter, fut2)  # White-box test.
        t.cancel()
        self.assertTrue(fut2.cancelled())
        res = self.loop.run_until_complete(t)
        self.assertEqual(res, 42)
        self.assertFalse(t.cancelled())

def test_cancel_current_task(self):
        loop = asyncio.new_event_loop()
        self.set_event_loop(loop)

        @asyncio.coroutine
        def task():
            t.cancel()
            self.assertTrue(t._must_cancel)  # White-box test.
            # The sleep should be cancelled immediately.
            yield from asyncio.sleep(100, loop=loop)
            return 12

        t = asyncio.Task(task(), loop=loop)
        self.assertRaises(
            asyncio.CancelledError, loop.run_until_complete, t)
        self.assertTrue(t.done())
        self.assertFalse(t._must_cancel)  # White-box test.
        self.assertFalse(t.cancel())

def test_as_completed_with_unused_timeout(self):

        def gen():
            yield
            yield 0
            yield 0.01

        loop = self.new_test_loop(gen)

        a = asyncio.sleep(0.01, 'a', loop=loop)

        @asyncio.coroutine
        def foo():
            for f in asyncio.as_completed([a], timeout=1, loop=loop):
                v = yield from f
                self.assertEqual(v, 'a')

        loop.run_until_complete(asyncio.Task(foo(), loop=loop))

def test_as_completed_duplicate_coroutines(self):

        @asyncio.coroutine
        def coro(s):
            return s

        @asyncio.coroutine
        def runner():
            result = []
            c = coro('ham')
            for f in asyncio.as_completed([c, c, coro('spam')],
                                          loop=self.loop):
                result.append((yield from f))
            return result

        fut = asyncio.Task(runner(), loop=self.loop)
        self.loop.run_until_complete(fut)
        result = fut.result()
        self.assertEqual(set(result), {'ham', 'spam'})
        self.assertEqual(len(result), 2)

def test_sleep(self):

        def gen():
            when = yield
            self.assertAlmostEqual(0.05, when)
            when = yield 0.05
            self.assertAlmostEqual(0.1, when)
            yield 0.05

        loop = self.new_test_loop(gen)

        @asyncio.coroutine
        def sleeper(dt, arg):
            yield from asyncio.sleep(dt/2, loop=loop)
            res = yield from asyncio.sleep(dt/2, arg, loop=loop)
            return res

        t = asyncio.Task(sleeper(0.1, 'yeah'), loop=loop)
        loop.run_until_complete(t)
        self.assertTrue(t.done())
        self.assertEqual(t.result(), 'yeah')
        self.assertAlmostEqual(0.1, loop.time())

def test_yield_vs_yield_from_generator(self):
        @asyncio.coroutine
        def coro():
            yield

        @asyncio.coroutine
        def wait_for_future():
            gen = coro()
            try:
                yield gen
            finally:
                gen.close()

        task = wait_for_future()
        self.assertRaises(
            RuntimeError,
            self.loop.run_until_complete, task)

def test_coroutine_non_gen_function_return_future(self):
        fut = asyncio.Future(loop=self.loop)

        @asyncio.coroutine
        def func():
            return fut

        @asyncio.coroutine
        def coro():
            fut.set_result('test')

        t1 = asyncio.Task(func(), loop=self.loop)
        t2 = asyncio.Task(coro(), loop=self.loop)
        res = self.loop.run_until_complete(t1)
        self.assertEqual(res, 'test')
        self.assertIsNone(t2.result())

def test_shield_effect(self):
        # Cancelling outer() does not affect inner().
        proof = 0
        waiter = asyncio.Future(loop=self.loop)

        @asyncio.coroutine
        def inner():
            nonlocal proof
            yield from waiter
            proof += 1

        @asyncio.coroutine
        def outer():
            nonlocal proof
            yield from asyncio.shield(inner(), loop=self.loop)
            proof += 100

        f = asyncio.async(outer(), loop=self.loop)
        test_utils.run_briefly(self.loop)
        f.cancel()
        with self.assertRaises(asyncio.CancelledError):
            self.loop.run_until_complete(f)
        waiter.set_result(None)
        test_utils.run_briefly(self.loop)
        self.assertEqual(proof, 1)

def test_yield_from_corowrapper(self):
        old_debug = asyncio.coroutines._DEBUG
        asyncio.coroutines._DEBUG = True
        try:
            @asyncio.coroutine
            def t1():
                return (yield from t2())

            @asyncio.coroutine
            def t2():
                f = asyncio.Future(loop=self.loop)
                asyncio.Task(t3(f), loop=self.loop)
                return (yield from f)

            @asyncio.coroutine
            def t3(f):
                f.set_result((1, 2, 3))

            task = asyncio.Task(t1(), loop=self.loop)
            val = self.loop.run_until_complete(task)
            self.assertEqual(val, (1, 2, 3))
        finally:
            asyncio.coroutines._DEBUG = old_debug

def _test_cancel_wait_for(self, timeout):
        loop = asyncio.new_event_loop()
        self.addCleanup(loop.close)

        @asyncio.coroutine
        def blocking_coroutine():
            fut = asyncio.Future(loop=loop)
            # Block: fut result is never set
            yield from fut

        task = loop.create_task(blocking_coroutine())

        wait = loop.create_task(asyncio.wait_for(task, timeout, loop=loop))
        loop.call_soon(wait.cancel)

        self.assertRaises(asyncio.CancelledError,
                          loop.run_until_complete, wait)

        # Python issue #23219: cancelling the wait must also cancel the task
        self.assertTrue(task.cancelled())

def test_constructor_loop_selection(self):
        @asyncio.coroutine
        def coro():
            return 'abc'
        gen1 = coro()
        gen2 = coro()
        fut = asyncio.gather(gen1, gen2)
        self.assertIs(fut._loop, self.one_loop)
        self.one_loop.run_until_complete(fut)

        self.set_event_loop(self.other_loop, cleanup=False)
        gen3 = coro()
        gen4 = coro()
        fut2 = asyncio.gather(gen3, gen4, loop=self.other_loop)
        self.assertIs(fut2._loop, self.other_loop)
        self.other_loop.run_until_complete(fut2)

def test_exception_marking(self):
        # Test for the first line marked "Mark exception retrieved."

        @asyncio.coroutine
        def inner(f):
            yield from f
            raise RuntimeError('should not be ignored')

        a = asyncio.Future(loop=self.one_loop)
        b = asyncio.Future(loop=self.one_loop)

        @asyncio.coroutine
        def outer():
            yield from asyncio.gather(inner(a), inner(b), loop=self.one_loop)

        f = asyncio.async(outer(), loop=self.one_loop)
        test_utils.run_briefly(self.one_loop)
        a.set_result(None)
        test_utils.run_briefly(self.one_loop)
        b.set_result(None)
        test_utils.run_briefly(self.one_loop)
        self.assertIsInstance(f.exception(), RuntimeError)

def async_handle_subscribe(request, subscribed_clients, state_variables):
    callback_url = request.headers.get('CALLBACK')[1:-1]
    sid = 'uuid:' + str(uuid.uuid4())
    subscribed_clients[sid] = callback_url

    headers = {
        'SID': sid
    }

    @asyncio.coroutine
    def async_push_later(state_variable):
        yield from asyncio.sleep(0.5)
        yield from state_variable.async_notify_listeners()

    for state_variable in state_variables.values():
        LOGGER.debug('Pushing state_variable on SUBSCRIBE: %s', state_variable.name)
        asyncio.get_event_loop().create_task(async_push_later(state_variable))

    return web.Response(status=200, headers=headers)
#endregion


#region Unsubscribe

def _connect(self):
        """
            Connect to the stream

        Returns
        -------
        asyncio.coroutine
            The streaming response
        """
        logger.debug("connecting to the stream")
        await self.client.setup.early
        if self.session is None:
            self.session = self.client._session
        kwargs = await self.client.headers.prepare_request(**self.kwargs)
        request = self.client.error_handler(self.session.request)

        return await request(*self.args, timeout=0, **kwargs)

def safe_send(self, dest, content: str, **kwargs):
        """ Sends a message and then deletes it after a certain time has passed.

        :param dest: Where the message will be sent.
        :param content: The content of the message to send.
        """
        tts = kwargs.pop('tts', False)
        delete_after = kwargs.pop('delete_after', 0)

        message = await self.send_message(
            lib.as_object(dest) if isinstance(dest, str) else dest,
            content, tts=tts)

        if message and delete_after > 0:
            @asyncio.coroutine
            def delete():
                yield from asyncio.sleep(delete_after)
                yield from self.delete_message(message)

            asyncio.ensure_future(delete(), loop=self.loop)

def create_connection(event_loop):

    conns = []

    @asyncio.coroutine
    def f(*args, **kwargs):
        kwargs.setdefault('loop', event_loop)
        conn = yield from aiossdb.create_connection(*args, **kwargs)
        # ??????????????????????yield?????return
        # ???????????????
        conns.append(conn)
        return conn

    try:
        yield f
    finally:
        waiters = []
        while conns:
            conn = conns.pop()
            conn.close()
            waiters.append(conn.wait_closed())
        if waiters:
            event_loop.run_until_complete(asyncio.gather(*waiters, loop=event_loop))

def create_connection_pool(event_loop):
    pools = []

    @asyncio.coroutine
    def f(*args, **kwargs):
        kwargs.setdefault('loop', event_loop)
        pool = yield from aiossdb.create_pool(*args, **kwargs)
        # ??????????????????????yield?????return
        # ???????????????
        pools.append(pool)
        return pool

    try:
        yield f
    finally:
        waiters = []
        while pools:
            conn = pools.pop()
            conn.close()
            waiters.append(conn.wait_closed())
        if waiters:
            event_loop.run_until_complete(asyncio.gather(*waiters, loop=event_loop))

def iscoroutinepartial(fn):
    """
    Function returns True if function it's a partial instance of coroutine. See additional information here_.

    :param fn: Function
    :return: bool

    .. _here: https://goo.gl/C0S4sQ

    """

    while True:
        parent = fn

        fn = getattr(parent, 'func', None)

        if fn is None:
            break

    return asyncio.iscoroutinefunction(parent)

def timeout(timeout_sec=5):
    def decorator(func):
        @asyncio.coroutine
        @wraps(func)
        def wrap(self, *args, **kwargs):
            future = asyncio.Future(loop=self.loop)

            def on_timeout(future: asyncio.Future):
                if future.done():
                    return

                future.set_exception(TimeoutError)

            self.loop.call_later(timeout_sec, on_timeout, future)

            result = yield from asyncio.coroutine(func)(self, *args, **kwargs)

            if not future.done():
                future.set_result(result)

            return (yield from future)

        return wrap
    return decorator

def test_get_ws_connection_invalid_error(client):
    """Test error opening a websocket connection with an engineIO session."""
    mock_ws = AsyncMock()
    receive_queue = asyncio.Queue(loop=client.loop)
    mock_ws.receive_str = receive_queue.get

    @asyncio.coroutine
    def mock_send_str(data):
        if data == "2probe":
            yield from receive_queue.put("3probe")
            return

        if data == "5":
            yield from receive_queue.put('44[[[')

    mock_ws.send_str = mock_send_str
    client._client_session.ws_connect.return_value = mock_ws
    session_data = {
        "sid": "mock_session_id",
        "pingTimeout": 12.345,
        "pingInterval": 23.456,
    }
    with pytest.raises(exceptions.ProtocolError):
        client.loop.run_until_complete(
            client._get_ws_connection(session_data))

def test_ws_loop_exception(client):
    """Test websocket loop exception."""
    @asyncio.coroutine
    def side_effect(*args, **kwargs):
        raise aiohttp.ClientError("Mock Exception")
    mock_ws = AsyncMock()
    mock_ws.receive.side_effect = side_effect

    client._ws_connection = mock_ws
    client.ws_close = AsyncMock()
    client._handle_event = MagicMock()

    with pytest.raises(exceptions.TransportError):
        client.loop.run_until_complete(client._ws_loop())

    assert client.ws_close.called
    assert len(client.ws_close.mock_calls) == 1
    assert client._handle_event.called
    assert len(client._handle_event.mock_calls) == 1
    assert client._handle_event.mock_calls[0][1][0] == 'closed'
    assert client._handle_event.mock_calls[0][1][1] is None

def test_ws_close_exception(client):
    """Test websocket close exception."""
    @asyncio.coroutine
    def side_effect(*args, **kwargs):
        raise aiohttp.ClientError("Mock Exception")
    mock_ws = AsyncMock()
    mock_ws.send_str.side_effect = side_effect

    client._ws_connection = mock_ws
    client._ws_session_data = {}
    client._handle_event = MagicMock()

    client.loop.run_until_complete(client.ws_close())

    assert mock_ws.close.called
    assert len(mock_ws.close.mock_calls) == 1
    assert mock_ws.send_str.called
    assert len(mock_ws.send_str.mock_calls) == 1
    assert mock_ws.send_str.mock_calls[0][1][0] == '41'

def test_reader(subscription):
    reader = subscription.get_reader()
    assert not reader.is_active
    m = mock.Mock()

    @reader.callback
    def callback(channel_name, model):
        m(model)
        return False

    @asyncio.coroutine
    def go():
        listener = yield from reader.listen()
        assert reader.is_active
        listener.cancel()
        with pytest.raises(asyncio.CancelledError):
            yield from listener

        assert not reader.is_active
        assert not m.called

        yield from reader.manager.stop()
        assert reader.manager.closed

    LOOP.run_until_complete(go())

def test_websocket_wrapper_authentication_error():
    loop = asyncio.get_event_loop()

    @websocket("/", authenticate=True)
    def handler(ws, params, **kwargs):
        ws.send_str("hello, world!")

    @asyncio.coroutine
    def start_server(loop):
        app = Application()
        app.router.add_route(*handler.route)
        srv = yield from loop.create_server(app.make_handler(), "localhost", 9000)
        return srv

    @asyncio.coroutine
    def go(loop):
        srv = yield from start_server(loop)
        with pytest.raises(WSServerHandshakeError):
            client = yield from ws_connect("http://localhost:9000")
            yield from client.close()

        srv.close()
        yield from srv.wait_closed()

    loop.run_until_complete(go(loop))