diff --git a/src/core/__init__.py b/src/core/__init__.py index 2174568..0835f7c 100644 --- a/src/core/__init__.py +++ b/src/core/__init__.py @@ -11,11 +11,11 @@ from .exceptions import * from .get_logger import get_logger from .structs import Addr -from .util import Immutable +from .util import Immutable, uninterruptible, process_redis_url __all__ = ["Addr", "StageEnum", "Immutable", "RCError", "AssignmentError", "NetworkError", "PartialResponseError", "PartialRequestError", "ConnectionCountError", "IS_CLI", "STAGE", "TLS_ENFORCED", "MAX_CONNECTIONS", "FILE_HANDLER", "STDOUT_HANDLER", "STDERR_HANDLER", - "get_logger"] + "get_logger", "uninterruptible", "process_redis_url"] diff --git a/src/core/config.py b/src/core/config.py index fdfa84e..9027eda 100644 --- a/src/core/config.py +++ b/src/core/config.py @@ -10,7 +10,7 @@ import sys from .constants import StageEnum -from .util import LogCompressor +from .log_compressor import LogCompressor __all__ = ["IS_CLI", "STAGE", "TLS_ENFORCED", "MAX_CONNECTIONS", "FILE_HANDLER", "STDOUT_HANDLER", "STDERR_HANDLER"] diff --git a/src/core/log_compressor.py b/src/core/log_compressor.py new file mode 100644 index 0000000..093d4ea --- /dev/null +++ b/src/core/log_compressor.py @@ -0,0 +1,60 @@ +from logging import Formatter + +class LogCompressor(Formatter): + """ + Custom formatter that truncates messages longer than a specified byte limit. + """ + + DEFAULT_MAX_BYTES: int = 256 + """ + Default maximum byte limit for message truncation. + """ + TRUNCATION_SUFFIX: bytes = "...".encode() + """ + Suffix added to truncated messages. + """ + MINIMUM_REQUIRED_BYTES: int = 3 + """ + Minimum required byte limit for message truncation. + """ + + def __init__(self, fmt=None, datefmt=None, max_bytes: int = DEFAULT_MAX_BYTES): + """ + Initializes the formatter with the specified format, date format, and maximum byte limit. + + Args: + fmt: The format string for the formatter. + datefmt: The date format string for the formatter. + max_bytes: The maximum byte limit for message truncation. + + Raises: + ValueError: If max_bytes is less than the minimum required bytes. + """ + super().__init__(fmt, datefmt) + if max_bytes < LogCompressor.MINIMUM_REQUIRED_BYTES: + raise ValueError("Invalid number of maximum bytes; must be at least 3") + self.max_bytes = max_bytes + + def format(self, record) -> str: + """ + Formats the log record by truncating the message if it exceeds the byte limit. + + Args: + record: The log record to format. + + Returns: + str: The formatted log record. + """ + msg = record.getMessage() + msg_bytes = msg.encode() + + if len(msg_bytes) > self.max_bytes: + # Truncate and add indicator. + suffix_bytes = len(LogCompressor.TRUNCATION_SUFFIX) + truncated_bytes = msg_bytes[:self.max_bytes - suffix_bytes] + LogCompressor.TRUNCATION_SUFFIX + truncated_msg = truncated_bytes.decode() + record.msg = truncated_msg + # Clear args so getMessage() doesn't try to re-format. + record.args = () + + return super().format(record) diff --git a/src/core/util.py b/src/core/util.py index b301908..8aa2154 100644 --- a/src/core/util.py +++ b/src/core/util.py @@ -1,66 +1,13 @@ -from logging import Formatter -from typing import Any +from typing import Any, Callable +from functools import wraps +from urllib.parse import urlparse +from .config import TLS_ENFORCED +from .constants import EMPTY_STR, SCHEME_LIST from .exceptions import AssignmentError +from .get_logger import get_logger -class LogCompressor(Formatter): - """ - Custom formatter that truncates messages longer than a specified byte limit. - """ - - DEFAULT_MAX_BYTES: int = 256 - """ - Default maximum byte limit for message truncation. - """ - TRUNCATION_SUFFIX: bytes = "...".encode() - """ - Suffix added to truncated messages. - """ - MINIMUM_REQUIRED_BYTES: int = 3 - """ - Minimum required byte limit for message truncation. - """ - - def __init__(self, fmt=None, datefmt=None, max_bytes: int = DEFAULT_MAX_BYTES): - """ - Initializes the formatter with the specified format, date format, and maximum byte limit. - - Args: - fmt: The format string for the formatter. - datefmt: The date format string for the formatter. - max_bytes: The maximum byte limit for message truncation. - - Raises: - ValueError: If max_bytes is less than the minimum required bytes. - """ - super().__init__(fmt, datefmt) - if max_bytes < LogCompressor.MINIMUM_REQUIRED_BYTES: - raise ValueError("Invalid number of maximum bytes; must be at least 3") - self.max_bytes = max_bytes - - def format(self, record) -> str: - """ - Formats the log record by truncating the message if it exceeds the byte limit. - - Args: - record: The log record to format. - - Returns: - str: The formatted log record. - """ - msg = record.getMessage() - msg_bytes = msg.encode() - - if len(msg_bytes) > self.max_bytes: - # Truncate and add indicator. - suffix_bytes = len(LogCompressor.TRUNCATION_SUFFIX) - truncated_bytes = msg_bytes[:self.max_bytes - suffix_bytes] + LogCompressor.TRUNCATION_SUFFIX - truncated_msg = truncated_bytes.decode() - record.msg = truncated_msg - # Clear args so getMessage() doesn't try to re-format. - record.args = () - - return super().format(record) +logger = get_logger(__name__) class Immutable: """ @@ -104,3 +51,71 @@ def __setattr__(self, name: str, value: Any) -> None: # If __getattribute__ raises AttributeError, the attribute is missing. # Therefore allow the construction of the object. super().__setattr__(name, value) + +def uninterruptible(callback: Callable) -> Callable: + """ + Decorator to make a function uninterruptible. + Catches system signals (like KeyboardInterrupt) and retries the execution, + ensuring the operation completes unless a standard Exception occurs. + """ + @wraps(callback) + def wrapper(*args, **kwargs): + while True: + try: + return callback(*args, **kwargs) + except Exception: + # Allow standard exceptions (ValueError, TypeError, etc.) to propagate. + raise + except BaseException as e: + # Catch signals (KeyboardInterrupt, SystemExit) and retry. + logger.error(f"func: {callback.__name__} interrupted by {e!r}. Retrying...") + return wrapper + +def process_redis_url(url: str, tls_enforced: bool = TLS_ENFORCED) -> tuple[str, str, str, str, str]: + """ + Processes and extracts data from a Redis URL string. + + Format: redis[s]://[[username][:password]@][host][:port][/db-number] + See more: https://docs.python.org/3/library/urllib.parse.html#url-parsing + + Args: + url (str): A string representing the Redis connection URL. + + Returns: + arr: A five-string tuple contaning the connection information. + + Raises: + ValueError: For invalid URL scheme or malformed URL. + ConnecterError: If the connection to the server fails. + """ + logger.debug(f"Processing Redis URL: {url}.") + parsed = urlparse(url) + + if parsed.scheme not in SCHEME_LIST: + raise ValueError(f"Invalid url scheme: '{parsed.scheme}'") + if tls_enforced and parsed.scheme == SCHEME_LIST[0]: + raise ValueError("TLS is enforced, and an invalid scheme was provided.") + + host = parsed.hostname if parsed.hostname else EMPTY_STR + port = str(parsed.port) if parsed.port else EMPTY_STR + user = parsed.username if parsed.username else EMPTY_STR + pasw = parsed.password if parsed.password else EMPTY_STR + + # Extraction of logical database index (the path segment). + # Redis URLs typically use /0, /1, etc. + db_idx = EMPTY_STR + if parsed.path: + # Strip leading slash, and check if the integer is valid. + db_idx = parsed.path.lstrip('/') + + if db_idx != EMPTY_STR: + try: + int(db_idx) + except ValueError: + raise ValueError( + f"Invalid database index: '{parsed.path}'; must be an integer" + ) + + # Do not log the password, sensitive data. + logger.debug(f"Url connection details: host={host}, port={port}, user={user}, db={db_idx}") + return (host, port, user, pasw, db_idx) diff --git a/src/main.py b/src/main.py index a275272..a904877 100644 --- a/src/main.py +++ b/src/main.py @@ -4,58 +4,54 @@ import core from frontend import Layout +from reactor import Reactor, ReactorClient from multiplexing import loop_multiplexing -from util import uninterruptible +from util import shutdown_app, add_conn -logger = core.get_logger(__name__) -@uninterruptible -async def close_page(multiplexing_event: Event, page: ft.Page) -> None: - logger.info("Closing application...") - try: - multiplexing_event.clear() - except BaseException as e: - logger.error(f"Application failed: {e}.", exc_info=True) - finally: - page.window.prevent_close = False - page.window.on_event = None - await page.window.destroy() - await page.window.close() +logger = core.get_logger(__name__) -@uninterruptible +@core.uninterruptible def build_page(page: ft.Page) -> None: + """ + Builds the main page. + """ try: + reactor = Reactor(page) multiplexing_event = Event() multiplexing_event.set() # Run the multiplexing loop in a background thread managed by Flet. - page.run_thread(loop_multiplexing, multiplexing_event) + page.run_thread(loop_multiplexing, multiplexing_event, reactor) logger.info("Multiplexing thread started.") - async def handle_close(event: ft.WindowEvent | None = None) -> None: - if event and event.data != "close" and event.type != ft.WindowEventType.CLOSE: - return - await close_page(multiplexing_event, page) + async def handle_close(event: ft.WindowEvent | None) -> None: + if event is None or event.data == "close" or event.type == ft.WindowEventType.CLOSE: + await shutdown_app(multiplexing_event, page) page.window.on_event = handle_close page.window.prevent_close = True + page.window.width = 800 + page.window.height = 600 + page.window.resizable = False page.theme_mode = ft.ThemeMode.DARK page.title = "RC-application" - + # Handles OS intrusions gracefully. # Similar to a regular container. - safe_area = ft.SafeArea(Layout(), expand=True) + reactor_client = ReactorClient(reactor) + layout = Layout(lambda conn_data: add_conn(conn_data, reactor_client, layout)) + safe_area = ft.SafeArea(layout, expand=True) page.add(safe_area) logger.info("Flet app window initialized.") except Exception as e: logger.error(f"Application failed: {e}.", exc_info=True) - page.window.close() + page.run_task(page.window.close) if __name__ == "__main__": if core.IS_CLI: raise NotImplementedError("CLI mode is not implemented yet") - logger.info("CLI mode enabled.") else: logger.info("GUI mode enabled.") ft.run(build_page, assets_dir="src/frontend/assets") diff --git a/src/multiplexing.py b/src/multiplexing.py index 07da18f..96a14a9 100644 --- a/src/multiplexing.py +++ b/src/multiplexing.py @@ -7,19 +7,17 @@ from selectors import EVENT_READ, EVENT_WRITE from threading import Event from time import sleep -from typing import Callable import core +from frontend import Chat from network import Connection import transmission -import reactor -from util import uninterruptible +from reactor import Reactor logger = core.get_logger(__name__) -@uninterruptible -def loop_multiplexing(stay_alive: Event) -> None: +def loop_multiplexing(stay_alive: Event, reactor: Reactor, timeout: float | None = None) -> None: """ The socket selection loop. @@ -27,121 +25,159 @@ def loop_multiplexing(stay_alive: Event) -> None: then selects ready sockets dispatching them to their corresponding handlers. Args: - stay_alive (obj): The event to signal the loop to continue/stop. + stay_alive (event): The event to signal the loop to continue/stop. + reactor (obj): The reactor instance to use. + timeout (float): The timeout for the selector. """ - while stay_alive.is_set(): - try: - _handle_connection_queues() - _sel_and_dispatch() - except Exception as e: - logger.critical(f"Multiplexing loop error: {e}.", exc_info=True) - # The application prepares to completely shutdown. - # Handle any remaining connections. - _handle_connection_queues() - reactor.close_resources() - -def _handle_connection_queues() -> None: - """ - Handles the interaction (add/removal) between the selector and enqueued connections. - """ - while reactor._connections_to_add: - connection, on_response = reactor._connections_to_add.popleft() - reactor.add_connection(connection, on_response) - - while reactor._connections_to_rem: - connection = reactor._connections_to_rem.popleft() - reactor.rem_connection(connection) + _Multiplexer(stay_alive, reactor, timeout) -_DEFAULT_TIMEOUT: float = 1 -""" -The default timeout for the event loop. -""" +class _Multiplexer: + """ + The main multiplexing loop class. -def _sel_and_dispatch(timeout: float = _DEFAULT_TIMEOUT) -> None: + It manages the event loop, handling connection queues and dispatching I/O events. """ - Polls for I/O events and dispatches them to the registered handlers. - It also manages connection related issues. - - Args: - timeout: The maximum time to wait for events. + DEFAULT_TIMEOUT: float = 1 """ - # No need to run `select()` if there are no connections. - if not reactor._selector.get_map(): - # Wait for a second, hoping that a client will enqueue a connection. - # This was arbitrarily chosen. - sleep(_DEFAULT_TIMEOUT) - return - - events = reactor._selector.select(timeout) - for key, mask in events: - try: - connection = key.fileobj - assert isinstance(connection, Connection) - response_lambda = reactor._response_lambdas[connection] + The default timeout for the event loop. + """ + + def __init__(self, stay_alive: Event, reactor: Reactor, timeout: float | None = None): + self.stay_alive = stay_alive + self.reactor = reactor + self.timeout = timeout if timeout is not None else _Multiplexer.DEFAULT_TIMEOUT + self.loop() + + @core.uninterruptible + def loop(self) -> None: + """ + Starts the event loop. + """ + while self.stay_alive.is_set(): + try: + self.handle_conn_queues() + self.sel_and_dispatch() + except Exception as e: + logger.critical(f"Multiplexing loop error: {e}.", exc_info=True) + # The event flag was cleared. + # The application prepares to completely shutdown. + self.reactor.close_resources() + + def handle_conn_queues(self) -> None: + """ + Handles the interaction (add/removal) between the selector and enqueued connections. + """ + while self.reactor.conns_to_reg: + conn = self.reactor.conns_to_reg.popleft() + self.reactor.add_conn(conn) - if mask & EVENT_READ: - _sel_readable(connection, response_lambda) - if mask & EVENT_WRITE: - _sel_writable(connection, response_lambda) + while self.reactor.conns_to_rem: + conn = self.reactor.conns_to_rem.popleft() + self.reactor.rem_conn(conn) + + def sel_and_dispatch(self) -> None: + """ + Polls for I/O events and dispatches them to the registered handlers. - except ConnectionError as e: - logger.warning(f"The connection {connection.addr} was closed by peer: {e}.") - logger.info("Removing the connection.") - reactor.rem_connection(connection) - continue - except Exception as e: - logger.error(f"Failed to handle event for connection {connection.addr}: {e}.", exc_info=True) - continue - -def _sel_readable(connection: Connection, response_lambda: Callable[[str], None]) -> None: - """ - Handles and processes readable sockets. + It also manages connection related issues. + """ + # No need to run `select()` if there are no connections. + if not self.reactor.selector.get_map(): + # Wait for a second, hoping that a client will enqueue a connection. + # This was arbitrarily chosen. + sleep(_Multiplexer.DEFAULT_TIMEOUT) + return - It also manages partial operations results, - and re-negotiates the protocol if necessary and possible (from RESP3 to RESP2) - in case the server does not support the newer version. - - Args: - connection (obj): The connection to handle. - response_lambda (lambda): The lambda function to forward the response to the client. - """ - try: - output_str = transmission.handle_read( - connection.addr, - connection.receiver, - connection.synchronizer.last_raw_input, - connection.synchronizer.all_sent) - response_lambda(output_str) - connection.synchronizer.all_recv = True + events = self.reactor.selector.select(self.timeout) + for key, mask in events: + + conn = key.fileobj + assert isinstance(conn, Connection) + + try: + if mask & EVENT_READ: + self.sel_readable(conn) + if mask & EVENT_WRITE: + self.sel_writable(conn) + + except ConnectionError as e: + logger.warning(f"The connection {conn.addr} was closed by peer: {e}.") + logger.info("Removing the connection.") + self.reactor.rem_conn(conn) + continue + + except Exception as e: + logger.error(f"Failed to handle event for conn {conn.addr}: {e}.", exc_info=True) + continue + + def sel_readable(self, conn: Connection) -> None: + """ + Handles and processes readable sockets. + + It also manages partial operations results, + and re-negotiates the protocol if necessary and possible (from RESP3 to RESP2) + in case the server does not support the newer version. + + Args: + conn (Connection): The connection to handle. + chat (Chat | None): The chat associated with the connection. + """ + try: + assert conn.synchronizer.last_raw_input is not None + output_str = transmission.handle_read( + conn.addr, + conn.receiver, + conn.synchronizer.last_raw_input, + conn.synchronizer.all_sent) + + self.cond_render(conn, output_str) + conn.synchronizer.all_recv = True - except core.PartialResponseError: - logger.debug("The response is not completely received.") - except core.PartialRequestError: - logger.debug("The request is not completely sent.") - # todo what if it is an auth error? - except transmission.Resp3NotSupportedError: - logger.warning("RESP3 not supported; retrying with RESP2.") - connection.say_hello(connection.initial_user, - connection.initial_pasw, - core.RespVer.RESP2) - -def _sel_writable(connection: Connection, response_lambda: Callable[[str], None]) -> None: - """ - Handles and processes writable sockets and manages invalid input and partial response issues. + except core.PartialResponseError: + logger.debug("The response is not completely received.") + except core.PartialRequestError: + logger.debug("The request is not completely sent.") + # todo what if it is an auth error? + except transmission.Resp3NotSupportedError: + logger.warning("RESP3 not supported; retrying with RESP2.") + conn.say_hello(conn.initial_user, + conn.initial_pasw, + core.RespVer.RESP2) + + def sel_writable(self, conn: Connection) -> None: + """ + Handles and processes writable sockets and manages invalid input and partial response issues. - Args: - connection (obj): The connection to handle. - response_lambda (lambda): The lambda function to forward potential input errors to the client. - """ - try: - transmission.handle_write( - connection.addr, - connection.sender, - connection.synchronizer) - except core.PartialResponseError: - logger.debug("The last result was not completely received.") - except ValueError as e: - # If the user makes an error, the error is both logged and printed on his screen as a response. - response_lambda(str(e)) - logger.error(f"Error when encoding data to {connection.addr}: {e}.", exc_info=True) + Args: + conn (Connection): The connection to handle. + chat (Chat | None): The chat associated with the connection. + """ + try: + transmission.handle_write( + conn.addr, + conn.sender, + conn.synchronizer) + except core.PartialResponseError: + logger.debug("The last result was not completely received.") + except ValueError as e: + # If the user makes an error, the error is both logged and printed on his screen as a response. + logger.error(f"Error when encoding data to {conn.addr}: {e}.", exc_info=True) + self.cond_render(conn, str(e)) + + def cond_render(self, conn: Connection, res: str) -> None: + """ + If the application was lauched in the GUI mode, it renders the response on the GUI. + + Args: + conn (obj): The connection object. + res (str): The response from either a network call or a client error. + """ + if core.IS_CLI: + return + + chat = self.reactor.conn_chat_dict[conn] + assert chat is not None + + assert self.reactor.page is not None + self.reactor.page.run_task(chat.add_res, res) diff --git a/src/network/connection.py b/src/network/connection.py index 9cd1fa4..243cbd4 100644 --- a/src/network/connection.py +++ b/src/network/connection.py @@ -30,3 +30,6 @@ def __init__(self, host: str, port: str, user: str, pasw: str, db_idx: str) -> N def close(self) -> None: self.sock.close() Connection.count -= 1 + + def __str__(self) -> str: + return str(self.addr) diff --git a/src/protocol/cmds/patterns/__init__.py b/src/protocol/cmds/patterns/__init__.py index 4318bf1..a7b96a8 100644 --- a/src/protocol/cmds/patterns/__init__.py +++ b/src/protocol/cmds/patterns/__init__.py @@ -1,6 +1,6 @@ from frozendict import frozendict -from .core.constants import * +from .constants import * from .interfaces import * from .sections import * from .types import * diff --git a/src/reactor.py b/src/reactor.py index ba2f2f6..01a36f5 100644 --- a/src/reactor.py +++ b/src/reactor.py @@ -11,124 +11,175 @@ The client interface threads enques operations to be performed by the common selector. The multiplexing loop thread dequeues and processes these operations. """ +import flet as ft from collections import deque import selectors -from typing import Callable import core +from frontend import Chat from network import Connection -from util import uninterruptible logger = core.get_logger(__name__) -# Client modules should only call these functions. -def enque_new_connection(connection: Connection, on_response: Callable[[str], None]) -> None: - """ - Enqueues a new connection to be added to the selector. - """ - logger.info(f"Enqueuing connection {connection.addr} to be added.") - _connections_to_add.append((connection, on_response)) - -def enque_close_connection(connection: Connection) -> None: - """ - Enqueues a connection to be removed from the selector. - """ - logger.info(f"Enqueuing connection {connection.addr} to be removed.") - _connections_to_rem.append(connection) - -# Calling these functions in the main thread provokes race-conditions. +# Directly using instances of Reactor provokes race-conditions. # Should only be used by the multiplexing thread. -@uninterruptible -def close_resources() -> None: +class Reactor: """ - Removes any active connections and the selector. + Attributes: + page (obj): The frontend page, if the application was launched in GUI mode. + It servers as a context for running async tasks i.e. printing output. + conn_chat_dict (dict): A dictionary of connections and their corresponding chats. - This function should be called right before exiting the application. + selector (obj): The selector used by the application. + conns_to_reg (deque): A queue of connections to be added to the selector. + conns_to_rem (deque): A queue of connections to be removed from the selector. + + In GUI mode, the chats are mandatory. + They provide the possibility to print to the user's screen the network outputs. """ - logger.info("Closing resources...") - - try: - leftover_connections = set(_response_lambdas.keys()) - if leftover_connections: - logger.warning("Removing leftover active connections.") - logger.debug(f"Leftover connections: {leftover_connections}.") - for connection in leftover_connections: - rem_connection(connection) - connections_to_add_len = len(_connections_to_add) - if connections_to_add_len: - logger.warning(f"Ignoring and closing {connections_to_add_len} connections enqueued to be added.") - logger.debug(f"Leftover connections to be added: {_connections_to_add}.") - for connection, _ in _connections_to_add: - connection.close() + def __init__(self, page: ft.Page | None = None) -> None: + """ + Args: + page (obj): The frontend page, if the application was launched in GUI mode. + + Raises: + TypeError: if the application was launched in GUI mode and the page was not provided. + """ + # Both must be of the same value. + # i.e. CLI and no page / GUI and page. + if core.IS_CLI != page is None: + raise TypeError("The page must be provided (only) in GUI mode") + + self.page = page + + self.conn_chat_dict: dict[Connection, Chat | None] = {} - connections_to_rem_len = len(_connections_to_rem) - if connections_to_rem_len: - logger.warning(f"Removing {connections_to_rem_len} connections enqueued to be removed.") - logger.debug(f"Leftover connections to be removed: {_connections_to_rem}.") - for connection in _connections_to_rem: - rem_connection(connection) + self.selector = selectors.DefaultSelector() + self.conns_to_reg: deque[Connection] = deque() + self.conns_to_rem: deque[Connection] = deque() + + def add_conn(self, conn: Connection) -> None: + """ + Registers a connection to the selector. + https://docs.python.org/3/library/selectors.html#selectors.BaseSelector.register + + Closes the connection if it fails to be registered. - _selector.close() - logger.info("Resources closed.") + Args: + conn (obj): The connection to register. + """ + try: + self.selector.register(conn, selectors.EVENT_READ | selectors.EVENT_WRITE) + except KeyError as e: + logger.error(f"Failed to register connection {conn.addr}: {e}.") + conn.close() + logger.info(f"Closed connection {conn.addr}.") + else: + logger.info(f"Added connection {conn.addr} to selector.") - except Exception as e: - logger.critical(f"Failed to close resources: {e}.", exc_info=True) + def rem_conn(self, conn: Connection) -> None: + """ + Removes a connection from the selector. + "A file object shall be unregistered prior to being closed." + https://docs.python.org/3/library/selectors.html#selectors.BaseSelector.unregister + + Args: + conn (obj): The connection to remove. + """ + try: + self.selector.unregister(conn) + self.conn_chat_dict.pop(conn) + except (KeyError, ValueError) as e: + logger.error(f"Failed to remove connection {conn.addr}: {e}.") + else: + logger.info(f"Removed connection {conn.addr} from selector.") + finally: + conn.close() + logger.info(f"Closed connection {conn.addr}.") -def add_connection(connection: Connection, on_response: Callable) -> None: - """ - Registers a connection to the selector. - https://docs.python.org/3/library/selectors.html#selectors.BaseSelector.register + # See main.py --- `shutdown_app()`. + # Doing `multiplexing_event.clear()` sends a signal to the multiplexing thread. + # It will elegantly finish its last iteration before running the below function. + # After all resources are cleared then the page is closed and destroyed. + @core.uninterruptible + def close_resources(self) -> None: + """ + Removes any active connections and the selector. + + This function should be called right before exiting the application. + """ + logger.info("Closing resources...") + + try: + leftover_conns = set(self.conn_chat_dict.keys()) + if leftover_conns: + logger.warning("Removing leftover active connections.") + for conn in leftover_conns: + logger.debug(f"Leftover connection: {conn}.") + self.rem_conn(conn) - Closes the connection if it fails to be registered. + conns_to_add_len = len(self.conns_to_reg) + if conns_to_add_len > 0: + logger.warning(f"Ignoring and closing {conns_to_add_len} conns enqueued to be added.") + for conn in self.conns_to_reg: + logger.debug(f"Ignoring connection: {conn}.") + conn.close() - Args: - connection (obj): The connection to register. - on_response (lambda): The callback function to be called when a response is received. - """ - try: - _selector.register(connection, selectors.EVENT_READ | selectors.EVENT_WRITE) - _response_lambdas[connection] = on_response - except KeyError as e: - logger.error(f"Failed to register connection {connection.addr}: {e}.") - connection.close() - logger.info(f"Closed connection {connection.addr}.") - else: - logger.info(f"Added connection {connection.addr} to selector.") - -def rem_connection(connection: Connection) -> None: + conns_to_rem_len = len(self.conns_to_rem) + if conns_to_rem_len > 0: + logger.warning(f"Removing {conns_to_rem_len} conns enqueued to be removed.") + for conn in self.conns_to_rem: + logger.debug(f"Removing connection: {conn}.") + self.rem_conn(conn) + + self.selector.close() + logger.info("Resources closed.") + + except Exception as e: + logger.critical(f"Failed to close resources: {e}.", exc_info=True) + +class ReactorClient: """ - Removes a connection from the selector. - "A file object shall be unregistered prior to being closed." - https://docs.python.org/3/library/selectors.html#selectors.BaseSelector.unregister + Client interface for the Reactor. - Args: - connection (obj): The connection to remove. + This class provides a thread-safe way for the frontend to interact with the reactor (e.g. enqueue connections/commands). """ - try: - _selector.unregister(connection) - _response_lambdas.pop(connection) - except (KeyError, ValueError) as e: - logger.error(f"Failed to remove connection {connection.addr}: {e}.") - else: - logger.info(f"Removed connection {connection.addr} from selector.") - finally: - connection.close() - logger.info(f"Closed connection {connection.addr}.") - -_selector = selectors.DefaultSelector() -""" -The unique selector used by the application. -""" -_response_lambdas: dict[Connection, Callable[[str], None]] = {} -""" -Lambda functions for each connection to be called when a full response is received. -""" -_connections_to_add: deque[tuple[Connection, Callable[[str], None]]] = deque() -""" -A queue of connections to be added to the selector. -""" -_connections_to_rem: deque[Connection] = deque() -""" -A queue of connections to be removed from the selector. -""" + + def __init__(self, reactor: Reactor): + self._reactor = reactor + + def enqueue_new_conn(self, conn_data: tuple) -> Connection: + """ + Enqueues a new connection to be added to the selector. + + Raises: + ConnectionCountError: If the maximum number of connections is reached. + """ + conn = Connection(*conn_data) + logger.info(f"Enqueuing connection {conn.addr} to be added.") + self._reactor.conns_to_reg.append(conn) + return conn + + # The below two methods are separated. + # The Chat can not be created before the connection is, + # and the bind is performed after the chat is created. + def enqueue_cmd(self, conn: Connection, cmd: str) -> None: + """ + Enqueues a command to be sent over the connection. + """ + conn.sender.add_pending(cmd) + + def bind_chat(self, conn: Connection, chat: Chat) -> None: + """ + Binds a Chat instance to a Connection. + """ + self._reactor.conn_chat_dict[conn] = chat + + def enqueue_close_conn(self, conn: Connection) -> None: + """ + Enqueues a connection to be closed and removed from the selector. + """ + logger.info(f"Enqueuing connection {conn.addr} to be removed.") + assert conn in self._reactor.conn_chat_dict.keys() + self._reactor.conns_to_rem.append(conn) diff --git a/src/transmission/handle_read.py b/src/transmission/handle_read.py index 9de0522..0a32cc7 100644 --- a/src/transmission/handle_read.py +++ b/src/transmission/handle_read.py @@ -5,7 +5,7 @@ logger = core.get_logger(__name__) -def handle_read(addr: core.Addr, receiver: Receiver, last_raw_cmd: str, all_sent: bool) -> str: +def handle_read(addr: core.Addr, receiver: Receiver, last_raw_cmd: str, all_sent: bool | None) -> str: """ Reads from the socket, decodes data, and updates history. @@ -31,8 +31,8 @@ def handle_read(addr: core.Addr, receiver: Receiver, last_raw_cmd: str, all_sent if all_sent != True: raise core.PartialRequestError("The request is not completely sent") + initial_buf_idx = receiver._idx try: - initial_buf_idx = receiver._idx output = process_output(receiver) if is_init_command(last_raw_cmd): validate_init_cmd_output(last_raw_cmd, output) @@ -57,5 +57,5 @@ def _handle_recv(receiver: Receiver, addr: core.Addr) -> None: except BlockingIOError: logger.warning("Receiving would block.") except ConnectionError as e: - logger.error(f"Error receiving data from {connection.addr}: {e}.") + logger.error(f"Error receiving data from {addr}: {e}.") raise diff --git a/src/transmission/handle_write.py b/src/transmission/handle_write.py index 6831638..4800cfc 100644 --- a/src/transmission/handle_write.py +++ b/src/transmission/handle_write.py @@ -45,7 +45,8 @@ def handle_write(addr: core.Addr, sender: Sender, synchronizer: Synchronizer) -> try: encoded = process_input(pending) except ValueError as e: - logger.debug(f"Unsyncing and removing the input from pending commands to send.") + logger.error(f"Input processing failed: {e}.") + logger.info("Unsyncing and removing the input from pending commands to send.") synchronizer.unsync() sender.rem_first_pending() raise diff --git a/src/transmission/processor.py b/src/transmission/processor.py index b250f90..f09ce11 100644 --- a/src/transmission/processor.py +++ b/src/transmission/processor.py @@ -34,7 +34,7 @@ def process_input(input_str: str) -> bytes: encoded = encoder(cmd, argv) return encoded.encode(core.ASCII_ENC) -def process_output(receiver: Receiver) -> str: +def process_output(receiver: Receiver) -> Output: """ Processes the output string by decoding it and formatting it. diff --git a/src/util.py b/src/util.py index d2a1c7d..cf17e77 100644 --- a/src/util.py +++ b/src/util.py @@ -1,75 +1,83 @@ -from functools import wraps -from typing import Callable -from urllib.parse import urlparse +import flet as ft +from threading import Event import core +from frontend import Chat, ConnBox, Layout +from network import Connection +from reactor import ReactorClient logger = core.get_logger(__name__) -def process_redis_url(url: str, tls_enforced: bool = core.TLS_ENFORCED) -> tuple[str, str, str, str, str]: +@core.uninterruptible +async def shutdown_app(multiplexing_event: Event, page: ft.Page | None = None) -> None: """ - Processes and extracts data from a Redis URL string. + Sends an event signal to the multiplexing loop to stop, and closes the page. - Format: redis[s]://[[username][:password]@][host][:port][/db-number] - See more: https://docs.python.org/3/library/urllib.parse.html#url-parsing - Args: - url (str): A string representing the Redis connection URL. + multiplexing_event (obj): The signalling event for the multiplexing loop. + page (obj): The page to destroy and close. - Returns: - arr: A five-string tuple contaning the connection information. - Raises: - ValueError: For invalid URL scheme or malformed URL. - ConnecterError: If the connection to the server fails. + AssertionError: If the page is not provided in GUI mode. """ - logger.debug(f"Processing Redis URL: {url}.") - parsed = urlparse(url) + logger.info("Closing application...") + multiplexing_event.clear() + + if core.IS_CLI: + return + + assert page is not None + page.window.prevent_close = False + page.window.on_event = None + await page.window.destroy() + await page.window.close() - if parsed.scheme not in core.SCHEME_LIST: - raise ValueError(f"Invalid url scheme: '{parsed.scheme}'") - if tls_enforced and parsed.scheme == core.SCHEME_LIST[0]: - raise ValueError("TLS is enforced, and an invalid scheme was provided.") +def add_conn(conn_data: tuple, reactor_client: ReactorClient, layout: Layout) -> None: + """ + Receives the connection details and creates a new Connection. + Sets up UI components (Chat, ConnBox), and enqueues the connection to the reactor. + + Args: + conn_data (arr): A tuple containing connection arguments (host, port, user, pass, db). + reactor_client (obj): The enqueuer of the connection creation. + layout (obj): The application layout. + """ + try: + # We get the connection object but it remains in read mode. + conn = reactor_client.enqueue_new_conn(conn_data) + except core.ConnectionCountError: + logger.error( + "Can not add a new connection.\n" + "Remove old connections or restart the application with a new \".env\" configuration.") + return + + conn_host = conn_data[0] + rem_conn_callback = lambda: rem_conn(conn, reactor_client, layout) - host = parsed.hostname if parsed.hostname else core.EMPTY_STR - port = str(parsed.port) if parsed.port else core.EMPTY_STR - user = parsed.username if parsed.username else core.EMPTY_STR - pasw = parsed.password if parsed.password else core.EMPTY_STR + conn_box = ConnBox( + text=conn_host.split(".")[0], + on_click=lambda: layout.chat_frame.sel_chat(chat), + on_close=rem_conn_callback) + layout.agenda_frame.add_box(conn_box) + layout.conn_boxes_dict[conn] = conn_box - # Extraction of logical database index (the path segment). - # Redis URLs typically use /0, /1, etc. - db_idx = core.EMPTY_STR - if parsed.path: - # Strip leading slash, and check if the integer is valid. - db_idx = parsed.path.lstrip('/') - - if db_idx != core.EMPTY_STR: - try: - int(db_idx) - except ValueError: - raise ValueError( - f"Invalid database index: '{parsed.path}'; must be an integer" - ) - - # Do not log the password, sensitive data. - logger.debug(f"Url connection details: host={host}, port={port}, user={user}, db={db_idx}") - return (host, port, user, pasw, db_idx) + chat = Chat( + text=conn_host, + on_enter=lambda cmd: reactor_client.enqueue_cmd(conn, cmd), + exit_cmd_callback=rem_conn_callback) + layout.chat_frame.sel_chat(chat) + reactor_client.bind_chat(conn, chat) -def uninterruptible(func: Callable) -> Callable: +def rem_conn(conn: Connection, reactor_client: ReactorClient, layout: Layout) -> None: """ - Decorator to make a function uninterruptible. - Catches system signals (like KeyboardInterrupt) and retries the execution, - ensuring the operation completes unless a standard Exception occurs. + Enqueues a connection to be unregistered, closed, and removed from the UI. + + Args: + conn (obj): The connection to be removed. + reactor_client (obj): The enqueuer of the connection creation. + layout (obj): The application layout. """ - @wraps(func) - def wrapper(*args, **kwargs): - while True: - try: - return func(*args, **kwargs) - except Exception: - # Allow standard exceptions (ValueError, TypeError, etc.) to propagate. - raise - except BaseException as e: - # Catch signals (KeyboardInterrupt, SystemExit) and retry. - logger.error(f"func: {func.__name__} interrupted by {e!r}. Retrying...") - return wrapper + reactor_client.enqueue_close_conn(conn) + conn_box = layout.conn_boxes_dict[conn] + layout.agenda_frame.rem_box(conn_box) + layout.chat_frame.rem_chat() diff --git a/tst/core/test_log_compressor.py b/tst/core/test_log_compressor.py new file mode 100644 index 0000000..a78d8b4 --- /dev/null +++ b/tst/core/test_log_compressor.py @@ -0,0 +1,42 @@ +from unittest import TestCase +import logging + +from src.core.log_compressor import LogCompressor + +class TestLogCompressor(TestCase): + + def test_valid_log_compressor_init(self): + compressor = LogCompressor(max_bytes=10) + self.assertEqual(compressor.max_bytes, 10) + + def test_invalid_log_compressor_init(self): + with self.assertRaises(ValueError): + LogCompressor(max_bytes=2) + + def test_log_compressor_format(self): + """ + Test LogCompressor formatting and truncation. + """ + + # Case 1: Message shorter than max_bytes. + compressor = LogCompressor(max_bytes=10) + record = logging.LogRecord("name", logging.INFO, "pathname", 1, "Short", (), None) + compressor.format(record) + self.assertEqual(record.msg, "Short") + + # Case 2: Message longer than max_bytes. + # "Hello World" is 11 bytes. max_bytes=10. Suffix is "...". + # Expect: "Hello W..." (7 bytes prefix + 3 bytes suffix = 10 bytes). + compressor = LogCompressor(max_bytes=10) + record = logging.LogRecord("name", logging.INFO, "pathname", 1, "Hello World", (), None) + compressor.format(record) + + self.assertEqual(len(record.msg), 10) + self.assertTrue(record.msg.endswith("...")) + self.assertEqual(record.msg, "Hello W...") + self.assertEqual(record.args, ()) + + # Case 3: Message exactly max_bytes. + record = logging.LogRecord("name", logging.INFO, "pathname", 1, "1234567890", (), None) + compressor.format(record) + self.assertEqual(record.msg, "1234567890") diff --git a/tst/core/test_util.py b/tst/core/test_util.py index 448a3d2..93bf63e 100644 --- a/tst/core/test_util.py +++ b/tst/core/test_util.py @@ -1,49 +1,10 @@ from unittest import TestCase -import logging +from unittest.mock import MagicMock from src.core.exceptions import AssignmentError -from src.core.util import LogCompressor, Immutable +from src.core.util import Immutable, uninterruptible, process_redis_url class TestUtil(TestCase): - - # ------------------------------ - # ---- LogCompressor class ----- - # ------------------------------ - - def test_valid_log_compressor_init(self): - compressor = LogCompressor(max_bytes=10) - self.assertEqual(compressor.max_bytes, 10) - - def test_invalid_log_compressor_init(self): - with self.assertRaises(ValueError): - LogCompressor(max_bytes=2) - - def test_log_compressor_format(self): - """Test LogCompressor formatting and truncation.""" - - # Case 1: Message shorter than max_bytes - compressor = LogCompressor(max_bytes=10) - record = logging.LogRecord("name", logging.INFO, "pathname", 1, "Short", (), None) - compressor.format(record) - self.assertEqual(record.msg, "Short") - - # Case 2: Message longer than max_bytes. - # "Hello World" is 11 bytes. max_bytes=10. Suffix is "...". - # Expect: "Hello W..." (7 bytes prefix + 3 bytes suffix = 10 bytes). - compressor = LogCompressor(max_bytes=10) - record = logging.LogRecord("name", logging.INFO, "pathname", 1, "Hello World", (), None) - compressor.format(record) - - self.assertEqual(len(record.msg), 10) - self.assertTrue(record.msg.endswith("...")) - self.assertEqual(record.msg, "Hello W...") - self.assertEqual(record.args, ()) - - # Case 3: Message exactly max_bytes - # "1234567890" (10 bytes) - record = logging.LogRecord("name", logging.INFO, "pathname", 1, "1234567890", (), None) - compressor.format(record) - self.assertEqual(record.msg, "1234567890") # ------------------------------ # ------ Immutable class ------- @@ -65,3 +26,63 @@ def test_assignment_failure(self): obj.key = "cool" with self.assertRaises(AssignmentError): obj.value = 1 + + # ------------------------------ + # - uninterruptible decorator -- + # ------------------------------ + + def test_uninterruptible_success(self): + func = MagicMock(return_value="OK") + decorated = uninterruptible(func) + result = decorated("arg") + self.assertEqual(result, "OK") + func.assert_called_with("arg") + + def test_uninterruptible_retry_on_interrupt(self): + func = MagicMock(side_effect=[KeyboardInterrupt, SystemExit, GeneratorExit, "OK"]) + func.__name__ = "mock_func" + + decorated = uninterruptible(func) + result = decorated("arg") + + self.assertEqual(result, "OK") + self.assertEqual(func.call_count, 4) + + def test_uninterruptible_exception(self): + func = MagicMock(side_effect=ValueError("fail")) + decorated = uninterruptible(func) + with self.assertRaises(ValueError): + decorated("arg") + self.assertEqual(func.call_count, 1) + + # ------------------------------ + # -- process_redis_url method -- + # ------------------------------ + + # format: redis[s]://[[username][:password]@][host][:port][/db-number] + def test_process_simple_url(self): + expected = ("localhost", "6379", "", "", "") + actual = process_redis_url("redis://localhost:6379") + self.assertEqual(actual, expected) + + def test_process_complete_url(self): + expected = ("localhost", "6379", "petru", "cool_petru", "4") + actual = process_redis_url("rediss://petru:cool_petru@localhost:6379/4") + self.assertEqual(actual, expected) + + def test_process_empty_url(self): + expected = ("", "", "", "", "") + actual = process_redis_url("redis://") + self.assertEqual(actual, expected) + + def test_process_url_invalid_scheme(self): + with self.assertRaises(ValueError): + process_redis_url("red://localhost:6379") + + def test_raise_when_tls_enforced(self): + with self.assertRaises(ValueError): + process_redis_url("redis://localhost:6379", tls_enforced=True) + + def test_process_url_invalid_db(self): + with self.assertRaises(ValueError): + process_redis_url("redis://localhost:6379/my_db") diff --git a/tst/test_util.py b/tst/test_util.py index f18830d..e952234 100644 --- a/tst/test_util.py +++ b/tst/test_util.py @@ -1,67 +1,99 @@ +import asyncio from unittest import TestCase -from unittest.mock import MagicMock +from unittest.mock import MagicMock, AsyncMock, patch -from src.util import process_redis_url, uninterruptible +from src.util import close_page, add_conn, rem_conn, core class TestUtil(TestCase): - # ------------------------------ - # -- process_redis_url method -- - # ------------------------------ + chat_patch = patch("src.util.Chat") + box_patch = patch("src.util.ConnBox") - # format: redis[s]://[[username][:password]@][host][:port][/db-number] - def test_process_simple_url(self): - expected = ("localhost", "6379", "", "", "") - actual = process_redis_url("redis://localhost:6379") - self.assertEqual(actual, expected) + def setUp(self): + self.mock_chat_cls = self.chat_patch.start() + self.mock_box_cls = self.box_patch.start() + + self.mock_reactor = MagicMock() + self.mock_layout = MagicMock() + self.mock_layout.conn_boxes_dict = {} - def test_process_complete_url(self): - expected = ("localhost", "6379", "petru", "cool_petru", "4") - actual = process_redis_url("rediss://petru:cool_petru@localhost:6379/4") - self.assertEqual(actual, expected) + def tearDown(self): + self.chat_patch.stop() + self.box_patch.stop() - def test_process_empty_url(self): - expected = ("", "", "", "", "") - actual = process_redis_url("redis://") - self.assertEqual(actual, expected) - - def test_process_url_invalid_scheme(self): - with self.assertRaises(ValueError): - process_redis_url("red://localhost:6379") - - def test_raise_when_tls_enforced(self): - with self.assertRaises(ValueError): - process_redis_url("redis://localhost:6379", tls_enforced=True) - - def test_process_url_invalid_db(self): - with self.assertRaises(ValueError): - process_redis_url("redis://localhost:6379/my_db") + def test_close_page(self): + mock_event = MagicMock() + mock_page = MagicMock() + mock_page.window.destroy = AsyncMock() + mock_page.window.close = AsyncMock() + + asyncio.run(close_page(mock_event, mock_page)) + + mock_event.clear.assert_called() + mock_page.window.destroy.assert_called() + mock_page.window.close.assert_called() - # ------------------------------ - # - uninterruptible decorator -- - # ------------------------------ + def test_add_conn_success(self): + mock_conn = MagicMock() + self.mock_reactor.enqueue_new_conn.return_value = mock_conn + + conn_data = ("localhost", "6379", "user", "pass") + add_conn(conn_data, self.mock_reactor, self.mock_layout) + + self.mock_reactor.enqueue_new_conn.assert_called_with(conn_data) + self.mock_box_cls.assert_called() + self.mock_layout.agenda_frame.add_box.assert_called() + self.mock_chat_cls.assert_called() + self.mock_layout.chat_frame.sel_chat.assert_called() + self.mock_reactor.bind_chat.assert_called_with(mock_conn, self.mock_chat_cls.return_value) + + self.assertIn(mock_conn, self.mock_layout.conn_boxes_dict) - def test_uninterruptible_success(self): - func = MagicMock(return_value="OK") - decorated = uninterruptible(func) - result = decorated("arg") - self.assertEqual(result, "OK") - func.assert_called_with("arg") + @patch("src.util.logger") + def test_add_conn_failure(self, mock_logger): + self.mock_reactor.enqueue_new_conn.side_effect = core.ConnectionCountError("Full") + + conn_data = ("localhost", "6379") + add_conn(conn_data, self.mock_reactor, self.mock_layout) + + self.mock_reactor.enqueue_new_conn.assert_called_with(conn_data) + mock_logger.error.assert_called() + + self.mock_box_cls.assert_not_called() + self.mock_layout.agenda_frame.add_box.assert_not_called() + self.mock_chat_cls.assert_not_called() + self.mock_layout.chat_frame.sel_chat.assert_not_called() + self.mock_reactor.bind_chat.assert_not_called() + + self.assertEqual(len(self.mock_layout.conn_boxes_dict), 0) - def test_uninterruptible_retry_on_interrupt(self): - func = MagicMock(side_effect=[KeyboardInterrupt, SystemExit, GeneratorExit, "OK"]) - func.__name__ = "mock_func" - - decorated = uninterruptible(func) - result = decorated("arg") - - self.assertEqual(result, "OK") - self.assertEqual(func.call_count, 4) - - def test_uninterruptible_exception(self): - func = MagicMock(side_effect=ValueError("fail")) - decorated = uninterruptible(func) - with self.assertRaises(ValueError): - decorated("arg") - self.assertEqual(func.call_count, 1) + def test_add_conn_callbacks(self): + """ + Verify lambda callbacks are wired correctly. + """ + mock_conn = MagicMock() + self.mock_reactor.enqueue_new_conn.return_value = mock_conn + + conn_data = ("host.com", "6379") + add_conn(conn_data, self.mock_reactor, self.mock_layout) + + box_kwargs = self.mock_box_cls.call_args[1] + on_click = box_kwargs['on_click'] + on_close = box_kwargs['on_close'] + + on_click() + self.mock_layout.chat_frame.sel_chat.assert_called() + + on_close() + self.mock_reactor.enqueue_close_conn.assert_called() + def test_rem_conn(self): + mock_conn = MagicMock() + mock_box = MagicMock() + + self.mock_layout.conn_boxes_dict = {mock_conn: mock_box} + rem_conn(mock_conn, self.mock_reactor, self.mock_layout) + + self.mock_reactor.enqueue_close_conn.assert_called_with(mock_conn) + self.mock_layout.agenda_frame.rem_box.assert_called_with(mock_box) + self.mock_layout.chat_frame.rem_chat.assert_called()