Server loop for accepting connections.
(self)
| 90 | self._thread = None |
| 91 | |
| 92 | def _server_loop(self) -> None: |
| 93 | """Server loop for accepting connections.""" |
| 94 | while self._running: |
| 95 | if self.protocol == "tcp": |
| 96 | try: |
| 97 | client, addr = self._socket.accept() |
| 98 | if self._client_socket: |
| 99 | try: |
| 100 | self._client_socket.close() |
| 101 | except Exception: |
| 102 | pass |
| 103 | self._client_socket = client |
| 104 | self._monitor_client(client) |
| 105 | except socket.timeout: |
| 106 | continue |
| 107 | except Exception: |
| 108 | break |
| 109 | else: |
| 110 | break |
| 111 | |
| 112 | def _monitor_client(self, client: socket.socket) -> None: |
| 113 | """Spawn a thread that clears _client_socket when the remote side closes.""" |
nothing calls this directly
no test coverage detected