(self, _cmd="X")
| 185 | self.stop(_cmd="B") |
| 186 | |
| 187 | def stop(self, _cmd="X"): |
| 188 | # type: (str) -> None |
| 189 | try: |
| 190 | with self.command_lock: |
| 191 | if self.thread is not None: |
| 192 | self._write_cmd(_cmd) |
| 193 | self.thread.join() |
| 194 | try: |
| 195 | self.thread_lock.release() |
| 196 | except Exception: |
| 197 | pass |
| 198 | else: |
| 199 | log_runtime.debug("Pipe engine thread not running") |
| 200 | except KeyboardInterrupt: |
| 201 | print("Interrupted by user.") |
| 202 | |
| 203 | def add(self, *pipes): |
| 204 | # type: (*Pipe) -> None |
no test coverage detected