Merge pull request #53 from GabyUnalaq/connection_problems_fix

Connection problems fix

Author: inf17101

ankaios_sdk/_components/control_interface.py

@@ -42,7 +42,7 @@
 - Change the state of the control interface.
     .. code-block:: python
 
-        ci.change_state(ControlInterfaceState.CONNECTED)
+        ci.change_state(ControlInterfaceState.TERMINATED)
 """
 
 
@@ -60,7 +60,7 @@
 
 from .._protos import _control_api
 from .request import Request
-from .response import Response, ResponseException
+from .response import Response, ResponseException, ResponseType
 from ..exceptions import ControlInterfaceException, ConnectionClosedException
 from ..utils import DEFAULT_CONTROL_INTERFACE_PATH, get_logger, ANKAIOS_VERSION
 
@@ -208,18 +208,24 @@ def _cleanup(self) -> None:
             self._input_file = None
         self._logger.debug("Cleanup happened")
 
-    def change_state(self, state: ControlInterfaceState) -> None:
+    def change_state(
+            self, state: ControlInterfaceState, info: str = None
+            ) -> None:
         """
         Change the state of the control interface.
 
         Args:
             state (ControlInterfaceState): The new state.
+            info (str): Additional information about the state change.
         """
         if state == self._state:
             self._logger.debug("State is already %s.", state)
             return
+        if self._state == ControlInterfaceState.CONNECTION_CLOSED:
+            self._logger.debug("State CONNECTION_CLOSED is unrecoverable.")
+            return
         self._state = state
-        self._state_changed_callback(state)
+        self._state_changed_callback(state, info)
 
     # pylint: disable=too-many-statements, too-many-branches
     def _read_from_control_interface(self) -> None:
@@ -298,12 +304,15 @@ def _read_from_control_interface(self) -> None:
                 except ResponseException as e:  # pragma: no cover
                     self._logger.error("Error while reading: %s", e)
                     continue
-                except ConnectionClosedException as e:  # pragma: no cover
-                    self._logger.error("Connection closed: %s", e)
-                    self.change_state(ControlInterfaceState.CONNECTION_CLOSED)
-                    break
 
                 self._add_response_callback(response)
+
+                if response.content_type == ResponseType.CONNECTION_CLOSED:
+                    self.change_state(
+                        ControlInterfaceState.CONNECTION_CLOSED,
+                        response.content
+                        )
+                    raise ConnectionClosedException(response.content)
         except Exception as e:  # pylint: disable=broad-exception-caught
             self._logger.error("Error while reading fifo file: %s", e)
         finally:
@@ -317,15 +326,15 @@ def _agent_gone_routine(self) -> None:
         It will attempt to write the hello message to the agent
         until the agent is connected.
         """
-        AGENT_RECONNECT_INTERVAL = 1  # seconds
+        agent_reconnect_interval = 1  # seconds
         while self.state == ControlInterfaceState.AGENT_DISCONNECTED:
             try:
                 self._send_initial_hello()
             except BrokenPipeError as _:
                 self._logger.warning(
                     "Waiting for the agent.."
                     )
-                time.sleep(AGENT_RECONNECT_INTERVAL)
+                time.sleep(agent_reconnect_interval)
             else:
                 self.change_state(ControlInterfaceState.INITIALIZED)
                 break
@@ -364,7 +373,12 @@ def write_request(self, request: Request) -> None:
 
         Raises:
             ControlInterfaceException: If not connected.
+            ConnectionClosedException: If the connection is closed.
         """
+        if self._state == ControlInterfaceState.CONNECTION_CLOSED:
+            raise ConnectionClosedException(
+                "Could not write to pipe, connection closed."
+            )
         if not self._state == ControlInterfaceState.INITIALIZED:
             raise ControlInterfaceException(
                 "Could not write to pipe, not connected.")

ankaios_sdk/_components/response.py

@@ -13,16 +13,14 @@
 # SPDX-License-Identifier: Apache-2.0
 
 """
-This script defines the Response and ResponseEvent classes,
+This script defines the Response and UpdateStateSuccess classes,
 used for receiving messages from the control interface.
 
 Classes
 --------
 
 - Response:
     Represents a response from the control interface.
-- ResponseEvent:
-    Represents an event used to wait for a response.
 - UpdateStateSuccess:
     Represents a response for a successful update state request.
 
@@ -31,7 +29,7 @@
 
 - ResponseType:
     Enumeration for the different types of response. It includes
-    ERROR, COMPLETE_STATE, and UPDATE_STATE_SUCCESS.
+    ERROR, COMPLETE_STATE, and UPDATE_STATE_SUCCESS and CONNECTION_CLOSED.
 
 Usage
 ------
@@ -56,13 +54,12 @@
         update_state_success.to_dict()
 """
 
-__all__ = ["Response", "ResponseType", "ResponseEvent", "UpdateStateSuccess"]
+__all__ = ["Response", "ResponseType", "UpdateStateSuccess"]
 
 from typing import Union
-from threading import Event
 from enum import Enum
 from .._protos import _control_api
-from ..exceptions import ResponseException, ConnectionClosedException
+from ..exceptions import ResponseException
 from ..utils import get_logger
 from .complete_state import CompleteState
 from .workload_state import WorkloadInstanceName
@@ -95,7 +92,6 @@ def __init__(self, message_buffer: bytes) -> None:
         self.content = None
 
         self._parse_response()
-        self._from_proto()
 
     def _parse_response(self) -> None:
         """
@@ -115,12 +111,13 @@ def _parse_response(self) -> None:
             raise ResponseException(f"Parsing error: '{e}'") from e
         if from_ankaios.HasField("response"):
             self._response = from_ankaios.response
+            self._from_proto()
+        elif from_ankaios.HasField("connectionClosed"):
+            self.content_type = ResponseType.CONNECTION_CLOSED
+            self.content = from_ankaios.connectionClosed.reason
         else:
-            logger.error(
-                "Connection closed by the server."
-            )
-            raise ConnectionClosedException(
-                from_ankaios.connectionClosed.reason)
+            raise ResponseException(  # pragma: no cover
+                "Invalid response type.")
 
     def _from_proto(self) -> None:
         """
@@ -168,6 +165,8 @@ def get_request_id(self) -> str:
         Returns:
             str: The request id of the response.
         """
+        if self.content_type == ResponseType.CONNECTION_CLOSED:
+            return None
         return self._response.requestId
 
     def get_content(self) -> \
@@ -193,6 +192,8 @@ class ResponseType(Enum):
     "(int): Got the complete state."
     UPDATE_STATE_SUCCESS = 3
     "(int): Got a successful update state response."
+    CONNECTION_CLOSED = 4
+    "(int): Connection closed by the server."
 
     def __str__(self) -> str:
         """
@@ -204,60 +205,6 @@ def __str__(self) -> str:
         return self.name.lower()
 
 
-class ResponseEvent(Event):
-    """
-    Represents an event that holds a Response object.
-    """
-    def __init__(self, response: Response = None) -> None:
-        """
-        Initializes the ResponseEvent with an optional Response object.
-
-        Args:
-            response Optional(Response): The response to associate with
-                the event. Defaults to None.
-        """
-        super().__init__()
-        self._response = response
-
-    def set_response(self, response: Response) -> None:
-        """
-        Sets the response and triggers the event.
-
-        Args:
-            response (Response): The response to set.
-        """
-        self._response = response
-        self.set()
-
-    def get_response(self) -> Response:
-        """
-        Gets the response associated with the event.
-
-        Returns:
-            Response: The response associated with the event.
-        """
-        return self._response
-
-    def wait_for_response(self, timeout: int) -> Response:
-        """
-        Waits for the response to be set, with a specified timeout.
-
-        Args:
-            timeout (int): The maximum time to wait for the response,
-                in seconds.
-
-        Returns:
-            Response: The response associated with the event.
-
-        Raises:
-            TimeoutError: If the response is not set within the
-                specified timeout.
-        """
-        if not self.wait(timeout):
-            raise TimeoutError("Timeout while waiting for the response.")
-        return self.get_response()
-
-
 class UpdateStateSuccess:
     """
     Represents an object that holds the added and deleted workloads.