94 lines
3.4 KiB
Python
94 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
|
|
from threading import Thread
|
|
from typing import Any, Self, Optional
|
|
from Interfaces.OpoTestsInterface import OpoTestsInterface
|
|
from socket import socket as Socket
|
|
from socket import AF_INET as ADDRESS_FAMILY_IPV4
|
|
from socket import SOCK_STREAM as SOCKET_STREAM
|
|
from socket import SOL_SOCKET as SOCKET_LAYER
|
|
from socket import SO_REUSEADDR as SOCKET_REUSE_ADDRESS
|
|
from Utils.Patterns import RE
|
|
|
|
|
|
class HTTPSocketDriver:
|
|
|
|
def __init__(self:Self, ot:OpoTestsInterface, inputs:Optional[dict[str, Any|None]|tuple[Any|None]|list] = None) -> None:
|
|
self.ot:OpoTestsInterface = ot
|
|
self.__socket:Socket|None = None
|
|
self.__address:str = self.ot.settings.get(("address", "http_address"), inputs, "localhost")
|
|
self.__port:int = self.ot.settings.get(("port", "http_port"), inputs, 8080)
|
|
self.__maximum_connections:int = self.ot.settings.get(("maximum_connections", "http_maximum_connections"), inputs, 5)
|
|
self.__cache_size:int = self.ot.settings.get(("cache_size", "http_cache_size"), inputs, 1024)
|
|
self.__public_path:str = self.ot.settings.get(("public_path", "http_public_path"), inputs, "/Public")
|
|
self.__working:bool = False
|
|
|
|
def __listen(self:Self) -> None:
|
|
while self.__working:
|
|
try:
|
|
|
|
client:Socket
|
|
address:str
|
|
port:int
|
|
data:bytes = b""
|
|
|
|
client, (address, port) = self.__socket.accept()
|
|
|
|
while True:
|
|
|
|
buffer:bytes = client.recv(self.__cache_size)
|
|
|
|
data += buffer
|
|
if len(buffer) < self.__cache_size:
|
|
break
|
|
|
|
file_data:bytes|None
|
|
|
|
for index in ("", "/index.html", "/index.htm"):
|
|
|
|
path:str = self.__public_path + RE.LINE.match(data.decode("utf-8")).group(1).split(" ")[1] + index
|
|
|
|
file_data = self.ot.files.load(path)
|
|
|
|
if file_data is not None:
|
|
break
|
|
|
|
if not file_data:
|
|
client.sendall(b"HTTP/1.1 404 Not Found\r\nContent-Type: text/plain\r\n\r\n404 Not Found")
|
|
else:
|
|
client.sendall(b"HTTP/1.1 200 OK\r\nContent-Type: " + self.ot.files.get_mime(path).encode("utf-8") + b"\r\n\r\n" + file_data)
|
|
client.close()
|
|
|
|
except Exception as exception:
|
|
self.ot.exception(exception, "opo_tests_http_error_listen", {
|
|
"address" : self.__address,
|
|
"port" : self.__port
|
|
})
|
|
|
|
def start(self:Self) -> None:
|
|
|
|
self.__working = True
|
|
self.__socket = Socket(ADDRESS_FAMILY_IPV4, SOCKET_STREAM)
|
|
|
|
try:
|
|
|
|
self.__socket.setsockopt(SOCKET_LAYER, SOCKET_REUSE_ADDRESS, 1)
|
|
self.__socket.bind((self.__address, self.__port))
|
|
self.__socket.listen(self.__maximum_connections)
|
|
|
|
Thread(target=self.__listen).start()
|
|
|
|
self.ot.print("info", "opo_tests_http_started", {
|
|
"address" : self.__address,
|
|
"port" : self.__port,
|
|
}, "HTTP server started on '{address}:{port}'.")
|
|
|
|
except Exception as exception:
|
|
self.close()
|
|
self.ot.exception(exception, "opo_tests_http_error_start", {
|
|
"address" : self.__address,
|
|
"port" : self.__port
|
|
})
|
|
|
|
def close(self:Self) -> None:
|
|
self.__working = False |