(vm: &VirtualMachine)
| 2351 | |
| 2352 | #[pyfunction] |
| 2353 | fn get_event_loop(vm: &VirtualMachine) -> PyResult { |
| 2354 | if let Some(loop_) = vm.asyncio_running_loop.borrow().clone() { |
| 2355 | return Ok(loop_); |
| 2356 | } |
| 2357 | |
| 2358 | let asyncio_events = vm.import("asyncio.events", 0)?; |
| 2359 | let get_event_loop_policy = vm |
| 2360 | .get_attribute_opt(asyncio_events, vm.ctx.intern_str("get_event_loop_policy"))? |
| 2361 | .ok_or_else(|| vm.new_attribute_error("get_event_loop_policy"))?; |
| 2362 | let policy = get_event_loop_policy.call((), vm)?; |
| 2363 | let get_event_loop = vm |
| 2364 | .get_attribute_opt(policy, vm.ctx.intern_str("get_event_loop"))? |
| 2365 | .ok_or_else(|| vm.new_attribute_error("get_event_loop"))?; |
| 2366 | get_event_loop.call((), vm) |
| 2367 | } |
| 2368 | |
| 2369 | #[pyfunction] |
| 2370 | fn current_task(args: LoopArg, vm: &VirtualMachine) -> PyResult { |
no test coverage detected