(self, scheduler: CompletionScheduler)
| 15462 | |
| 15463 | class CompletionService: |
| 15464 | def __init__(self, scheduler: CompletionScheduler) -> None: |
| 15465 | self.scheduler = scheduler |
| 15466 | self.formatter = scheduler.formatter |
| 15467 | self.condition = threading.Condition() |
| 15468 | self.commands: Deque[Callable[[], None]] = deque() |
| 15469 | self.closed = False |
| 15470 | self.worker = threading.Thread( |
| 15471 | target=self.run_loop, |
| 15472 | name="completion-service", |
| 15473 | daemon=True, |
| 15474 | ) |
| 15475 | self.worker.start() |
| 15476 | |
| 15477 | def close(self) -> None: |
| 15478 | with self.condition: |
nothing calls this directly
no outgoing calls
no test coverage detected