blob: 7da90c983a8842f7ee1b250d24c3bfc48702d69f (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
|
import socket
import threading
class MockSDS1000XEServer:
IP = "0.0.0.0"
PORT = 5025
MOTD = "Hello".encode()
def __init__(self):
self.serverSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.serverSocket.bind((MockSDS1000XEServer.IP, MockSDS1000XEServer.PORT))
self.serverSocket.listen(1)
print("server socket is bound")
self.main = threading.Thread(target=self._mainLoop)
self.main.start()
def _mainLoop(self) -> None:
clientSocket, _ = self.serverSocket.accept()
clientSocket.settimeout(0.1)
clientSocket.send(MockSDS1000XEServer.MOTD)
try:
while True:
request = clientSocket.recv(4096).decode()
if request != "":
print(f"** Received request: {request}")
response = self._handleRequest(request)
clientSocket.send(response.encode())
except Exception as e:
print("Exception caught in main server loop")
print(e)
def _handleRequest(self, request: str) -> str:
return "NOP"
def stop(self):
print("Closing server socket")
self.serverSocket.close()
print("joining main")
self.main.join()
print("stopped")
|
