(self, loop, *, task_class, eager)
| 361 | |
| 362 | class AsyncTaskCounter: |
| 363 | def __init__(self, loop, *, task_class, eager): |
| 364 | self.suspense_count = 0 |
| 365 | self.task_count = 0 |
| 366 | |
| 367 | def CountingTask(*args, eager_start=False, **kwargs): |
| 368 | if not eager_start: |
| 369 | self.task_count += 1 |
| 370 | kwargs["eager_start"] = eager_start |
| 371 | return task_class(*args, **kwargs) |
| 372 | |
| 373 | if eager: |
| 374 | factory = asyncio.create_eager_task_factory(CountingTask) |
| 375 | else: |
| 376 | def factory(loop, coro, **kwargs): |
| 377 | return CountingTask(coro, loop=loop, **kwargs) |
| 378 | loop.set_task_factory(factory) |
| 379 | |
| 380 | def get(self): |
| 381 | return self.task_count |
nothing calls this directly
no test coverage detected