Method register_callback

pywebio/session/coroutinebased.py:173–221

向Session注册一个回调函数，返回回调id :type callback: Callable or Coroutine :param callback: 回调函数. 函数签名为 ``callback(data)``. ``data`` 参数为回调事件的值 :param bool mutex_mode: 互斥模式。若为 ``True`` ，则在运行回调函数过程中，无法响应同一组件（callback_id相同）的新点击事件，仅当 ``callback`` 为协程函数时有效 :return str: 回调id.

(self, callback, mutex_mode=False)

        self._cleanup()

    def register_callback(self, callback, mutex_mode=False):
        """ 向Session注册一个回调函数，返回回调id

        :type callback: Callable or Coroutine
        :param callback: 回调函数. 函数签名为 ``callback(data)``. ``data`` 参数为回调事件的值
        :param bool mutex_mode: 互斥模式。若为 ``True`` ，则在运行回调函数过程中，无法响应同一组件（callback_id相同）的新点击事件，仅当 ``callback`` 为协程函数时有效
        :return str: 回调id.
        """

        async def callback_coro():
            while True:
                try:
                    event = await self.next_client_event()
                except SessionClosedException:
                    return

                assert event['event'] == 'callback'
                coro = None
                if iscoroutinefunction(callback):
                    coro = callback(event['data'])
                elif isgeneratorfunction(callback):
                    coro = asyncio.coroutine(callback)(event['data'])
                else:
                    try:
                        res = callback(event['data'])
                        if asyncio.iscoroutine(res):
                            coro = res
                        else:
                            del res  # `res` maybe pywebio.io_ctrl.Output, so need release `res`
                    except Exception:
                        self.on_task_exception()

                if coro is not None:
                    if mutex_mode:
                        await coro
                    else:
                        self.run_async(coro)

        cls = type(self)
        callback_task = Task(callback_coro(), cls.get_current_session())
        # Activate task
        # Don't callback.step(), it will result in recursive calls to step()
        # todo: integrate with inactive_coro_instances
        callback_task.coro.send(None)
        cls.get_current_session().coros[callback_task.coro_id] = callback_task

        self._need_keep_alive = True

        return callback_task.coro_id

    def run_async(self, coro_obj):
        """异步运行协程对象。可以在协程内调用 PyWebIO 交互函数