43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import socket
|
|
from typing import Optional
|
|
|
|
|
|
class ClientConnection:
|
|
def __init__(self, host: str, port: int, timeout: float = 30.0) -> None:
|
|
self.host = host
|
|
self.port = port
|
|
self.timeout = timeout
|
|
self.sock: Optional[socket.socket] = None
|
|
self._buffer = b""
|
|
|
|
def connect(self) -> None:
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.connect((self.host, self.port))
|
|
self.sock = s
|
|
|
|
def send_line(self, line: str) -> None:
|
|
if self.sock is None:
|
|
raise RuntimeError("Not connected")
|
|
data = (line + "\n").encode("utf-8")
|
|
self.sock.sendall(data)
|
|
|
|
def read_line(self) -> Optional[str]:
|
|
if self.sock is None:
|
|
raise RuntimeError("Not connected")
|
|
while True:
|
|
if b"\n" in self._buffer:
|
|
line, self._buffer = self._buffer.split(b"\n", 1)
|
|
return line.decode("utf-8", errors="replace").rstrip("\r")
|
|
chunk = self.sock.recv(4096)
|
|
if not chunk:
|
|
return None
|
|
self._buffer += chunk
|
|
|
|
def close(self) -> None:
|
|
if self.sock is not None:
|
|
try:
|
|
self.sock.close()
|
|
finally:
|
|
self.sock = None
|