()
| 180 | """ |
| 181 | |
| 182 | async def callback_coro(): |
| 183 | while True: |
| 184 | try: |
| 185 | event = await self.next_client_event() |
| 186 | except SessionClosedException: |
| 187 | return |
| 188 | |
| 189 | assert event['event'] == 'callback' |
| 190 | coro = None |
| 191 | if iscoroutinefunction(callback): |
| 192 | coro = callback(event['data']) |
| 193 | elif isgeneratorfunction(callback): |
| 194 | coro = asyncio.coroutine(callback)(event['data']) |
| 195 | else: |
| 196 | try: |
| 197 | res = callback(event['data']) |
| 198 | if asyncio.iscoroutine(res): |
| 199 | coro = res |
| 200 | else: |
| 201 | del res # `res` maybe pywebio.io_ctrl.Output, so need release `res` |
| 202 | except Exception: |
| 203 | self.on_task_exception() |
| 204 | |
| 205 | if coro is not None: |
| 206 | if mutex_mode: |
| 207 | await coro |
| 208 | else: |
| 209 | self.run_async(coro) |
| 210 | |
| 211 | cls = type(self) |
| 212 | callback_task = Task(callback_coro(), cls.get_current_session()) |
nothing calls this directly
no test coverage detected