(self)
| 3292 | self.raise_on_sigint = False |
| 3293 | |
| 3294 | def cmdloop(self): |
| 3295 | with ( |
| 3296 | self._sigint_handler(), |
| 3297 | self.readline_completion(self.complete), |
| 3298 | ): |
| 3299 | while not self.write_failed: |
| 3300 | try: |
| 3301 | if not (payload_bytes := self._readline()): |
| 3302 | break |
| 3303 | except KeyboardInterrupt: |
| 3304 | self.send_interrupt() |
| 3305 | continue |
| 3306 | |
| 3307 | try: |
| 3308 | payload = json.loads(payload_bytes) |
| 3309 | except json.JSONDecodeError: |
| 3310 | print( |
| 3311 | f"*** Invalid JSON from remote: {payload_bytes!r}", |
| 3312 | flush=True, |
| 3313 | ) |
| 3314 | continue |
| 3315 | |
| 3316 | self.process_payload(payload) |
| 3317 | |
| 3318 | def send_interrupt(self): |
| 3319 | if self.interrupt_sock is not None: |
nothing calls this directly
no test coverage detected