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

def execute_receiver(receiver, args, kwargs, ignore_exceptions=False):
        try:
            if asyncio.iscoroutinefunction(receiver):
                if len(args) > 0:
                    return receiver, await receiver(*args, **kwargs)
                return receiver, await receiver(**kwargs)

            if len(args) > 0:
                return receiver, receiver(*args, **kwargs)
            return receiver, receiver(**kwargs)
        except Exception as exc:
            if not ignore_exceptions:
                raise

            logger.exception(SignalException(
                'Signal receiver \'{}\' => {} thrown an exception!'.format(receiver.__module__, receiver.__name__)
            ), exc_info=False)

            # Handle, will send to sentry if it's related to the core/contrib apps.
            handle_exception(exc, receiver.__module__, receiver.__name__)

            # Log the actual exception.
            logger.exception(exc)
            return receiver, exc

def set_value(self, value):
        """
        Set the value, this will serialize and save the setting to the data storage.

        :param value: Python value input.
        :raise: NotFound / SerializationException
        """
        old_value = self._value[0] if self._value and len(self._value) > 0 else None

        model = await self.get_model()
        model.value = self.serialize_value(value)
        self._value = (True, model.value)
        await model.save()

        # Call the change target.
        if self.change_target and callable(self.change_target):
            if iscoroutinefunction(self.change_target):
                await self.change_target(old_value, model.value)
            else:
                self.change_target(old_value, model.value)

def subscribe(self, topics=None):
        """Decorator"""

        if not topics:
            topics = set()

        if isinstance(topics, str):
            topics = {topics}

        def real_decorator(f):
            # if function is a coro, add some new functions
            if asyncio.iscoroutinefunction(f):
                if not topics:
                    log.error("Empty topic fount in function '{}'. Skipping "
                              "it.".format(f.__name__))
                for topic in topics:
                    self.topics_subscribers[topic].add(f)
            return f

        return real_decorator

def add_task(self, function: callable, name: str = None) -> callable:
        if not asyncio.iscoroutinefunction(function):
            log.warning("Function '{}' is not a coroutine and can't be added "
                        "as a task".format(function.__name__))
            return

        function.delay = partial(self.context_class,
                                 function,
                                 self.task_list_name,
                                 self.poller,
                                 self._loop_delay)

        if name:
            function_name = name
        else:
            function_name = function.__name__

        function.function_name = function_name

        self.task_available_tasks[function_name] = function

def message_handler(self, message_type):
        """
        Set handler for message type:
            - text
            - picture
            - contact
            - location
            - file
            - video
            - sticker
            - rich_media
            - url
        """

        def decorator(coro):
            assert aio.iscoroutinefunction(coro), 'Decorated function should be coroutine'

            self.add_handler(message_type, coro)
            return coro

        return decorator

def event_handler(self, event_type):
        """
        Set callback for specific event type:
            - delivered
            - seen
            - conversation_started
            - message
            - subscribed
            - unsubscribed
            - failed
            - webhook
        """
        assert event_type in EventType.not_message_events(), 'Wrong event type'

        def wrap(coro):
            assert aio.iscoroutinefunction(coro), 'Decorated function should be coroutine'

            self._events_callbacks[event_type] = coro
            return coro

        return wrap

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 install(context, request):
    data = await request.json()
    id_to_install = data.get('id', None)
    if id_to_install not in app_settings['available_addons']:
        return ErrorResponse(
            'RequiredParam',
            _("Property 'id' is required to be valid"))

    registry = request.site_settings
    config = registry.for_interface(IAddons)

    if id_to_install in config.enabled:
        return ErrorResponse(
            'Duplicate',
            _("Addon already installed"))
    handler = app_settings['available_addons'][id_to_install]['handler']
    if asyncio.iscoroutinefunction(handler.install):
        await handler.install(context, request)
    else:
        handler.install(context, request)
    config.enabled |= {id_to_install}
    return await get_addons(context, request)()

def uninstall(context, request):
    data = await request.json()
    id_to_install = data.get('id', None)
    if id_to_install not in app_settings['available_addons']:
        return ErrorResponse(
            'RequiredParam',
            _("Property 'id' is required to be valid"))

    registry = request.site_settings
    config = registry.for_interface(IAddons)

    if id_to_install not in config.enabled:
        return ErrorResponse(
            'Duplicate',
            _("Addon not installed"))

    handler = app_settings['available_addons'][id_to_install]['handler']
    if asyncio.iscoroutinefunction(handler.install):
        await handler.uninstall(context, request)
    else:
        handler.uninstall(context, request)
    config.enabled -= {id_to_install}

def add_irc_event(self, irc_event: event.event, insert: bool = False):
        logger.debug('Adding %s to events', irc_event)
        if not asyncio.iscoroutinefunction(irc_event.callback):
            raise ValueError('Event handlers must be coroutines')

        # key is used to link irc_events_re and irc_events
        matcher = irc_event.compile(self.config)
        key = irc_event.key  # = regexp

        if key not in self.irc_events:
            if insert:
                self.irc_events_re.appendleft((key, matcher))
            else:
                self.irc_events_re.append((key, matcher))

        if insert:
            self.irc_events[key].appendleft(irc_event)
        else:
            self.irc_events[key].append(irc_event)

def trigger_event(self, event, *args):
        """Dispatch an event to the proper handler method.

        In the most common usage, this method is not overloaded by subclasses,
        as it performs the routing of events to methods. However, this
        method can be overriden if special dispatching rules are needed, or if
        having a single method that catches all events is desired.

        Note: this method is a coroutine.
        """
        handler_name = 'on_' + event
        if hasattr(self, handler_name):
            handler = getattr(self, handler_name)
            if asyncio.iscoroutinefunction(handler) is True:
                try:
                    ret = await handler(*args)
                except asyncio.CancelledError:  # pragma: no cover
                    pass
            else:
                ret = handler(*args)
            return ret

def trigger_callback(self, sid, namespace, id, data):
        """Invoke an application callback.

        Note: this method is a coroutine.
        """
        callback = None
        try:
            callback = self.callbacks[sid][namespace][id]
        except KeyError:
            # if we get an unknown callback we just ignore it
            self.server.logger.warning('Unknown callback received, ignoring.')
        else:
            del self.callbacks[sid][namespace][id]
        if callback is not None:
            if asyncio.iscoroutinefunction(callback) is True:
                try:
                    await callback(*data)
                except asyncio.CancelledError:  # pragma: no cover
                    pass
            else:
                callback(*data)

def _trigger_event(self, event, namespace, *args):
        """Invoke an application event handler."""
        # first see if we have an explicit handler for the event
        if namespace in self.handlers and event in self.handlers[namespace]:
            if asyncio.iscoroutinefunction(self.handlers[namespace][event]) \
                    is True:
                try:
                    ret = await self.handlers[namespace][event](*args)
                except asyncio.CancelledError:  # pragma: no cover
                    pass
            else:
                ret = self.handlers[namespace][event](*args)
            return ret

        # or else, forward the event to a namepsace handler if one exists
        elif namespace in self.namespace_handlers:
            return await self.namespace_handlers[namespace].trigger_event(
                event, *args)

def __run(self, app, settings):
        if asyncio.iscoroutinefunction(self.run):
            # Blocking call which returns when finished
            loop = asyncio.get_event_loop()
            loop.run_until_complete(self.run(self.arguments, settings, app))
            loop.run_until_complete(self.wait_for_tasks())
            loop.close()
        else:
            self.run(self.arguments, settings, app)

        if self.profiler is not None:
            if self.arguments.profile_output:
                self.profiler.dump_stats(self.arguments.profile_output)
            else:
                # dump to screen
                self.profiler.print_stats(-1)
        if self.line_profiler is not None:
            self.line_profiler.disable_by_count()
            if self.arguments.line_profiler_output:
                self.line_profiler.dump_stats(self.arguments.line_profiler_output)
            else:
                self.line_profiler.print_stats()

def add_event(self, loop, instance, fn):
        """
        Wraps the given function in a corutine which calls it every N seconds
        :param loop: The event loop (cls._loop)
        :param instance: The instantiated class
        :param fn: The function to be called
        :return: None
        """
        # Verify function signature
        argspec = inspect.getfullargspec(fn)
        n_args = len(argspec.args) - 1 if 'self' in argspec.args else len(argspec.args)
        n_required_args = n_args - (len(argspec.defaults) if argspec.defaults else 0)
        assert n_required_args == 0, 'Functions decorated with @Periodic cannot have any required parameters.'

        @asyncio.coroutine
        def periodic_fn():
            # If coroutine yield from else call normally
            is_coroutine = asyncio.iscoroutinefunction(fn)

            while True:
                last_exec = time.time()
                (yield from fn()) if is_coroutine else fn()
                yield from asyncio.sleep(max(0, self.time + last_exec - time.time()))

        super().add_event(loop, instance, periodic_fn())

def command(fn, name=None):
    """Decorator for functions that should be exposed as commands."""
    module = sys.modules[fn.__module__]
    if name is None:
        name = fn.__name__
    if asyncio.iscoroutinefunction(fn if inspect.isfunction(fn) else fn.__func__ if inspect.ismethod(fn) else fn.__call__): # Get the actual function for coroutine check
        @functools.wraps(fn)
        async def wrapper(*args, **kwargs):
            try:
                frame = inspect.currentframe()
                ctx = frame.f_back.f_locals['ctx']
                return await fn(ctx, *args, **kwargs)
            finally:
                del frame
    else:
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            try:
                frame = inspect.currentframe()
                ctx = frame.f_back.f_locals['ctx']
                return fn(ctx, *args, **kwargs)
            finally:
                del frame
    vars(module).setdefault('commands', {})[fn.__name__] = wrapper
    return wrapper

def invoke(self, ctx):
        locals_ = locals().copy()
        try:
            load_function(self.code, dict(globals(), **ctx.bot.commands), locals_)
        except SyntaxError as err:
            traceback.print_exception(type(err), err, err.__traceback__)
            return 'SyntaxError: %s' % err

        try:
            ret = await locals_['evaluation'](ctx)
            if asyncio.iscoroutine(ret):
                ret = await ret
            elif ret in ctx.bot.commands.values():
                ret = await ret() if asyncio.iscoroutinefunction(ret) else ret()
        except Exception as err:
            traceback.print_exception(type(err), err, err.__traceback__)
            return '%s: %s' % (type(err).__name__, err)
        else:
            return str(ret)

def on_run(self):
        try:
            while True:
                await asyncio.wait_for(
                    self.revive_event.wait(),
                    timeout=self.timeout,
                    loop=self.loop,
                )
                self.revive_event.clear()
        except asyncio.TimeoutError:
            try:
                if asyncio.iscoroutinefunction(self.callback):
                    await self.callback()
                else:
                    self.callback()
            except Exception:
                logger.exception("Error in timeout callback execution.")

def on_run(self):
        while True:
            try:
                await asyncio.wait_for(
                    self.reset_event.wait(),
                    timeout=self.period,
                    loop=self.loop,
                )
            except asyncio.TimeoutError:
                try:
                    if asyncio.iscoroutinefunction(self.callback):
                        await self.callback()
                    else:
                        self.callback()
                except Exception:
                    logger.exception("Error in timer callback execution.")
            else:
                self.reset_event.clear()

def send(self, group_resolver):
        coros = []
        for i, g in self._signals:
            if not group_resolver.match(g):
                continue
            if asyncio.iscoroutinefunction(i):
                params = inspect.signature(i).parameters
                if 'context' in params:
                    coro = i(self)
                else:
                    coro = i()
            elif asyncio.iscoroutine(i):
                coro = i
            else:
                continue
            coros.append(coro)
        await self._context.wait_all(coros)

def _make_response(self, request, response):
        if callable(response):
            if asyncio.iscoroutinefunction(response):
                return await response(request)
            else:
                return response(request)
        elif isinstance(response, dict):
            return web.json_response(response)

        rqdata = await request.json()
        if 'method' in rqdata and rqdata['method'] == 'session-get':
            return web.json_response(SESSION_GET_RESPONSE)
        elif response is None:
            raise RuntimeError('Set the response property before making a request!')
        else:
            return web.Response(text=response)

def run(self, TORRENT_FILTER, FILE_FILTER, columns):
        columns = self.cfg['columns.files'].value if columns is None else columns
        try:
            columns = self.get_file_columns(columns)
            tfilter = self.select_torrents(TORRENT_FILTER,
                                           allow_no_filter=False,
                                           discover_torrent=True)
            ffilter = self.select_files(FILE_FILTER,
                                        allow_no_filter=True,
                                        discover_file=False)
        except ValueError as e:
            log.error(e)
            return False

        log.debug('Listing %s files of %s torrents', ffilter, tfilter)

        if asyncio.iscoroutinefunction(self.make_flist):
            return await self.make_flist(tfilter, ffilter, columns)
        else:
            return self.make_flist(tfilter, ffilter, columns)

def run(self, TORRENT_FILTER, sort, columns):
        sort = self.cfg['sort.torrents'].value if sort is None else sort
        columns = self.cfg['columns.torrents'].value if columns is None else columns
        try:
            columns = self.get_torrent_columns(columns)
            tfilter = self.select_torrents(TORRENT_FILTER,
                                           allow_no_filter=True,
                                           discover_torrent=False)
            sort = self.get_torrent_sorter(sort)
        except ValueError as e:
            log.error(e)
            return False
        else:
            log.debug('Listing %s torrents sorted by %s', tfilter, sort)
            if asyncio.iscoroutinefunction(self.make_tlist):
                return await self.make_tlist(tfilter, sort, columns)
            else:
                return self.make_tlist(tfilter, sort, columns)

def pytest_pyfunc_call(pyfuncitem):
    """
    Run asyncio marked test functions in an event loop instead of a normal
    function call.
    """
    if 'run_loop' in pyfuncitem.keywords:
        funcargs = pyfuncitem.funcargs
        loop = funcargs['loop']
        testargs = {arg: funcargs[arg]
                    for arg in pyfuncitem._fixtureinfo.argnames}

        if not asyncio.iscoroutinefunction(pyfuncitem.obj):
            func = asyncio.coroutine(pyfuncitem.obj)
        else:
            func = pyfuncitem.obj
        loop.run_until_complete(func(**testargs))
        return True

def register(self, match, func, flags=0, mention=False, admin=False,
                 channel_id='*'):

        logger.debug('Registering message: %s, %s from %s',
                     match,
                     func.__name__,
                     inspect.getabsfile(func))

        if not asyncio.iscoroutinefunction(func):
            func = asyncio.coroutine(func)

        option = {
            'func': func,
            'mention': mention,
            'admin': admin,
            'channel_id': channel_id
        }

        self._endpoints[re.compile(match, flags)].append(option)

def retry(*dargs, **dkw):
    """Wrap a function with a new `Retrying` object.

    :param dargs: positional arguments passed to Retrying object
    :param dkw: keyword arguments passed to the Retrying object
    """
    # support both @retry and @retry() as valid syntax
    if len(dargs) == 1 and callable(dargs[0]):
        return retry()(dargs[0])
    else:
        def wrap(f):
            if asyncio and asyncio.iscoroutinefunction(f):
                r = AsyncRetrying(*dargs, **dkw)
            elif tornado and tornado.gen.is_coroutine_function(f):
                r = TornadoRetrying(*dargs, **dkw)
            else:
                r = Retrying(*dargs, **dkw)

            return r.wraps(f)

        return wrap

def observe(self, event, fn):
        """
        Arguments:
            event (str): event to subscribe.
            fn (function|coroutinefunction): function to trigger.

        Raises:
            TypeError: if fn argument is not valid
        """
        iscoroutine = asyncio.iscoroutinefunction(fn)
        if not iscoroutine and not isfunction(fn):
            raise TypeError('fn param must be a callable object '
                            'or coroutine function')

        observers = self._pool.get(event)
        if not observers:
            observers = self._pool[event] = []

        # Register the observer
        observers.append(fn if iscoroutine else coroutine_wrapper(fn))

def on_packet(self, data, address):
        self.log_packet(data, address, received=True)

        handlers = self.handlers.get(self.packet_id(data), ())
        origin_handlers = [i for i in handlers if i[1] is None or i[1] == address]
        if not origin_handlers:
            print("No handlers for the previously received message")

        data = self.handler_data(data)
        for handler_tuple in origin_handlers:
            handler, origin_filter = handler_tuple
            stream = BitStream(data)
            if asyncio.iscoroutinefunction(handler):
                asyncio.ensure_future(handler(stream, address))
            else:
                handler(stream, address)

    # Packet callbacks

def command(command, flags=re.IGNORECASE, role=None, ban_groups=[]):
    """
    Decorator that registers a function as a command.
    This is regex.
    """

    def inner(function):
        if not asyncio.iscoroutinefunction(function):
            logger.warning("Attempted to register non-coroutine %s!", function)
            function = asyncio.coroutine(function)

        pattern = re.compile(command, flags)
        commands[pattern] = function
        function.role_requirement = role
        function.ban_groups = ban_groups
        return function

    return inner

def extract_info(self, *args, on_error=None, retry_on_error=False, **kwargs):
        if callable(on_error):
            try:
                return await self.loop.run_in_executor(self.thread_pool, partial(self.unsafe_ytdl.extract_info, *args, **kwargs))
            except Exception as e:
                if asyncio.iscoroutinefunction(on_error):
                    asyncio.ensure_future(on_error(e), loop=self.loop)
                elif asyncio.iscourutine(on_error):
                    asyncio.ensure_future(on_error, loop=loop)
                else:
                    loop.call_soon_threadsafe(on_error, e)

                if retry_on_error:
                    return await self.safe_extract_info(loop, *args, **kwargs)

        else:
            return await self.loop.run_in_executor(self.thread_pool, partial(self.unsafe_ytdl.extract_info, *args, **kwargs))

def pytest_pyfunc_call(pyfuncitem):
    if 'run_loop' in pyfuncitem.keywords:
        funcargs = pyfuncitem.funcargs

        loop = funcargs['loop']

        testargs = {
            arg: funcargs[arg]
            for arg in pyfuncitem._fixtureinfo.argnames
        }

        assert asyncio.iscoroutinefunction(pyfuncitem.obj)

        loop.run_until_complete(pyfuncitem.obj(**testargs))

        return True

def run(self, context):
        """
        Run this step and record the results.

        :param context: A context object too pass into this steps function.
        """
        try:
            if asyncio.iscoroutinefunction(self.function):
                await self.function(context, **self.kwargs)
            else:
                self.function(context, **self.kwargs)
        except Exception:
            raise
        else:
            self.passed = True
        finally:
            self.ran = True

def test_transfermethod_success_registering_decorated_transfermethod_without_schedule(self):
        '''transfermethod object should be able to register a decorated transfer_method without schedule '''
        def func(param):
            pass
        tm=transfer_methods.transfermethod(f_params={'param':'param'})
        self.assertTrue(isinstance(tm.mid,uuid.UUID))
        self.assertEqual(tm.schedule, None)
        self.assertEqual(tm.f_params, {'param':'param'})
        f=tm(func)
        self.assertEqual(f,func)
        self.assertEqual(tm._func_params.keys(), {'param':'param'}.keys())
        self.assertNotEqual(tm.schedule, None)
        self.assertTrue(isinstance(tm.schedule, schedules.OnUpdateSchedule))
        self.assertEqual(tm.schedule.activation_metrics, [])
        self.assertEqual(tm.schedule.exec_on_load, False)
        self.assertIsNotNone(getattr(tm,'f',None))
        self.assertTrue(asyncio.iscoroutinefunction(tm.f))
        tm_info = tmIndex.get_tm_info(tm.mid)
        self.assertEqual(tm_info['enabled'], False)
        self.assertEqual(tm_info['tm'], tm)

def test_transfermethod_success_registering_decorated_transfermethod_CronSchedule(self):
        '''transfermethod object should be able to register a decorated transfer_method with CronSchedule '''
        def func(param):
            pass
        tm=transfer_methods.transfermethod(f_params={'param':'param'}, schedule=schedules.CronSchedule())
        self.assertTrue(isinstance(tm.mid,uuid.UUID))
        self.assertNotEqual(tm.schedule, None)
        self.assertTrue(isinstance(tm.schedule, schedules.CronSchedule))
        self.assertEqual(tm.f_params, {'param':'param'})
        f=tm(func)
        self.assertEqual(f,func)
        self.assertEqual(tm._func_params.keys(), {'param':'param'}.keys())
        self.assertNotEqual(tm.schedule, None)
        self.assertTrue(isinstance(tm.schedule, schedules.CronSchedule))
        self.assertEqual(tm.schedule.activation_metrics, [])
        self.assertEqual(tm.schedule.exec_on_load, False)
        self.assertIsNotNone(getattr(tm,'f',None))
        self.assertTrue(asyncio.iscoroutinefunction(tm.f))
        tm_info = tmIndex.get_tm_info(tm.mid)
        self.assertEqual(tm_info['enabled'], False)
        self.assertEqual(tm_info['tm'], tm)

def test_transfermethod_success_registering_decorated_transfermethod_DummySchedule(self):
        '''transfermethod object should be able to register a decorated transfer_method with DummySchedule '''
        def func(param):
            pass
        tm=transfer_methods.transfermethod(f_params={'param':'param'}, schedule=schedules.DummySchedule(exec_on_load=True))
        self.assertTrue(isinstance(tm.mid,uuid.UUID))
        self.assertNotEqual(tm.schedule, None)
        self.assertTrue(isinstance(tm.schedule, schedules.DummySchedule))
        self.assertEqual(tm.schedule.exec_on_load, True)
        self.assertEqual(tm.f_params, {'param':'param'})
        f=tm(func)
        self.assertEqual(f,func)
        self.assertEqual(tm._func_params.keys(), {'param':'param'}.keys())
        self.assertNotEqual(tm.schedule, None)
        self.assertTrue(isinstance(tm.schedule, schedules.DummySchedule))
        self.assertEqual(tm.schedule.activation_metrics, [])
        self.assertEqual(tm.schedule.exec_on_load, True)
        self.assertIsNotNone(getattr(tm,'f',None))
        self.assertTrue(asyncio.iscoroutinefunction(tm.f))
        tm_info = tmIndex.get_tm_info(tm.mid)
        self.assertEqual(tm_info['enabled'], False)
        self.assertEqual(tm_info['tm'], tm)

def load_entry_points():
    importlib.reload(pkg_resources)
    for ep in pkg_resources.iter_entry_points(group=defaults.PACKAGES_ENTRY_POINT):
        logging.logger.info('loading entry_point: '+str(ep))
        try:
            f = ep.load()
            if asyncio.iscoroutinefunction(f):
                await f()
            elif isinstance(f, types.FunctionType):
                f()
        except (ModuleNotFoundError,SyntaxError):
            logging.logger.error('Error loading package entry point.')
            ex_info=traceback.format_exc().splitlines()
            for line in ex_info:
                logging.logger.error(line)
            return False
        else:
            logging.logger.info('entry_point loaded successfully: '+str(ep))
    return True

def maybe_async_cps(fn: Callable) -> MaybeCorouteine:

    def parted(context: Callable[[Maybe], Response]):
        @wraps(fn)
        @maybe_coroutine(fn)
        def _(*args, **kwargs):
            try:
                if iscoroutinefunction(fn):
                    result = yield from fn(*args, **kwargs)
                else:
                    result = fn(*args, **kwargs)
            except Exception as e:
                result = e
            return context(result)
        return _
    return parted

def post_validate_resource(self, data):
        # NOTE: The fields in *data* are ordered, such that children are
        #       listed before their parent.
        for key, (field_data, field_sp) in data.items():
            field = self.get_field(key)
            field.post_validate(self, field_data, field_sp)

            # Run custom post-validators for field
            validators = get_processors(self, Tag.VALIDATE, field, None)
            for validator, validator_kwargs in validators:
                if validator_kwargs['step'] is not Step.AFTER_DESERIALIZATION:
                    continue
                if validator_kwargs['on'] not in (Event.ALWAYS, self.ctx.event):
                    continue

                if asyncio.iscoroutinefunction(validator):
                    await validator(field, field_data, field_sp,
                                    context=self.ctx)
                else:
                    validator(field, field_data, field_sp, context=self.ctx)

def process_in(self, iprot):
        api, type, seqid = yield from iprot.read_message_begin()
        if api not in self._service.thrift_services:
            yield from iprot.skip(TType.STRUCT)
            yield from iprot.read_message_end()
            return api, seqid, TApplicationException(TApplicationException.UNKNOWN_METHOD), None

        args = getattr(self._service, api + "_args")()
        yield from iprot.read_struct(args)
        yield from iprot.read_message_end()
        result = getattr(self._service, api + "_result")()

        # convert kwargs to args
        api_args = [args.thrift_spec[k][1] for k in sorted(args.thrift_spec)]

        @asyncio.coroutine
        def call():
            f = getattr(self._handler, api)
            arguments = (args.__dict__[k] for k in api_args)
            if asyncio.iscoroutinefunction(f):
                rv = yield from f(*arguments)
                return rv
            return f(*arguments)

        return api, seqid, result, call

def isawaitable(obj):
    ''' Return True if the object is an awaitable or is a function that
    returns an awaitable.

    This function is used internally by aiotesting.
    '''
    if PY35:
        result = inspect.iscoroutinefunction(obj) or inspect.isawaitable(obj)

    elif PY34:
        result = (isinstance(obj, asyncio.Future) or
                  asyncio.iscoroutine(obj) or
                  hasattr(obj, '__await__'))
    else:
        raise Exception(
            'isawaitable is not supported on Python {}'.format(
                sys.version_info))
    return result

def _tearDown(self):
        ''' Destroy the event loop '''
        if asyncio.iscoroutinefunction(self.tearDown):
            self.loop.run_until_complete(self.tearDown())
        else:
            self.tearDown()

        if not isinstance(self.loop, asyncio.AbstractEventLoop):
            raise Exception('Invalid event loop: ', self.loop)
        if self.loop.is_running():
            self.loop.stop()
        self.loop.close()
        del self.loop
        asyncio.set_event_loop_policy(None)
        asyncio.set_event_loop(None)

        # By explicitly forcing a garbage collection here,
        # the event loop will report any remaining sockets
        # and coroutines left in the event loop which indicates
        # that further cleanup actions should be implemented
        # in the code under test.
        gc.collect()

def __setattr__(self, attr, val):
        if attr.startswith('_'):
            super().__setattr__(attr, val)
        else:
            if isinstance(val, list):
                self._dct[attr] = ListProxy(val)
            elif isinstance(val, dict):
                self._dct[attr] = DictProxy(val)
            elif asyncio.iscoroutine(val) or asyncio.iscoroutinefunction(val) or isinstance(val, asyncio.Future):
                val = asyncio.async(val)

                def set_later(future_val, attr=attr):
                    setattr(self, attr, future_val.result())

                val.add_done_callback(set_later)
            else:
                self._dct[attr] = val

def __setattr__(self, attr, val):
        if attr.startswith('_'):
            super().__setattr__(attr, val)
        else:
            if isinstance(val, list):
                self._dct[attr] = ListProxy(val)
            elif isinstance(val, dict):
                self._dct[attr] = DictProxy(val)
            elif asyncio.iscoroutine(val) or asyncio.iscoroutinefunction(val) or isinstance(val, asyncio.Future):
                val = asyncio.async(val)

                def set_later(future_val, attr=attr):
                    setattr(self, attr, future_val.result())

                val.add_done_callback(set_later)
            else:
                self._dct[attr] = val

def maybe_resume(func):

    if asyncio.iscoroutinefunction(func):
        @asyncio.coroutine
        @functools.wraps(func)
        def wrapper(self, *args, **kw):
            result = yield from func(self, *args, **kw)
            self._check_buffer_size()
            return result
    else:
        @functools.wraps(func)
        def wrapper(self, *args, **kw):
            result = func(self, *args, **kw)
            self._check_buffer_size()
            return result

    return wrapper

def request_with_callback(self, request: _Request, callback=None):
        if not callback:
            callback = request.callback
        if callable(callback):
            try:
                async with self.session.request(request.method, request.url) as resp:
                    '''
                    if callback is a coroutine-function, the await is necessary.
                    if not, call_soon_threadsafe is better.
                    But why not coroutine ?
                    '''
                    if asyncio.iscoroutinefunction(callback):
                        await callback(resp)
                    else:
                        self.loop.call_soon_threadsafe(callback, resp)
                    self.log(logging.INFO, "Request [{method}] `{url}` finishend.(There are still {num})".format(
                        method=request.method, url=request.url, num=self.pending.qsize()))
            except Exception as e:
                self.log(logging.ERROR, "Error happened in request [{method}] `{url}`, Request is ignored.\n{error}".format(
                    error=traceback.format_exc(), url=request.url, method=request.method))
        else:
            self.log(logging.WARNING, "Callback for request [{method}] `{url}` is not callable. Request is ignored.".format(
                url=request.url, method=request.method))

def add_route(app, fn):
    """
    ???????????WEB APP???
    :param app: WEB APP??
    :param fn: ????
    """
    # ?????????????
    method = getattr(fn, '__method__', None)
    path = getattr(fn, '__route__', None)
    if path is None or method is None:
        return

    # ????????????
    if not asyncio.iscoroutinefunction(fn) and not inspect.isgeneratorfunction(fn):
        fn = asyncio.coroutine(fn)

    logging.info('add route function: %s(%s), method(%s), path(%s)' %
                 (fn.__name__, ', '.join(inspect.signature(fn).parameters.keys()), method, path, ))

    app.router.add_route(method, path, fn)

def __getattr__(self, attr):
        rv = self._wrap_cache.get(attr)
        if rv is not None:
            # noinspection PyArgumentList
            return types.MethodType(rv, self)
        cls_val = getattr(self._conn_cls, attr)
        if asyncio.iscoroutinefunction(cls_val):
            @functools.wraps(cls_val)
            async def wrapper(_, *args, **kwargs):
                conn = await self.get_connection()
                return await getattr(conn, attr)(*args, **kwargs)
            self._wrap_cache[attr] = wrapper
            # noinspection PyArgumentList
            return types.MethodType(wrapper, self)
        if self._conn is None:
            raise InterfaceError(
                'Connection is not ready yet, or has been released')
        return getattr(self._conn, attr)

def map(source, func, *more_sources):
    """Apply a given function to the elements of one or several
    asynchronous sequences.

    Each element is used as a positional argument, using the same order as
    their respective sources. The generation continues until the shortest
    sequence is exhausted. The function can either be synchronous or
    asynchronous.

    Note: the different sequences are awaited in parallel, so that their
    waiting times don't add up.
    """
    iscorofunc = asyncio.iscoroutinefunction(func)
    if more_sources:
        source = zip(source, *more_sources)
    async with streamcontext(source) as streamer:
        async for item in streamer:
            if not more_sources:
                item = (item,)
            result = func(*item)
            if iscorofunc:
                result = await result
            yield result

def takewhile(source, func):
    """Forward an asynchronous sequence while a condition is met.

    The given function takes the item as an argument and returns a boolean
    corresponding to the condition to meet. The function can either be
    synchronous or asynchronous.
    """
    iscorofunc = asyncio.iscoroutinefunction(func)
    async with streamcontext(source) as streamer:
        async for item in streamer:
            result = func(item)
            if iscorofunc:
                result = await result
            if not result:
                return
            yield item

def add_route(app, fn):
    '''
    ??????URL????
    '''
    method = getattr(fn, '__method__', None)
    path = getattr(fn, '__route__', None)
    if path is None or method is None:
        raise ValueError('@get or @post not defined in %s.' % str(fn))
    if not asyncio.iscoroutinefunction(fn) and not inspect.isgeneratorfunction(fn):
        fn = asyncio.coroutine(fn)
    logging.info('add route %s %s => %s(%s)' % (method, path, fn.__name__, ', '.join(inspect.signature(fn).parameters.keys())))
    app.router.add_route(method, path, RequestHandler(app, fn))


# ???????
# ???handler???????????????