rest_api_child

drunc.controller.children_interface.rest_api_child

Classes

AppCommander(app_name, app_host, app_port, response_host, response_port, proxy_host=None, proxy_port=None)

Source code in drunc/controller/children_interface/rest_api_child.py

def __init__(
    self,
    app_name: str,
    app_host: str,
    app_port: int,
    response_host: str,
    response_port: int,
    proxy_host: str = None,
    proxy_port: int = None,
):
    self.app_host = app_host
    self.app_port = app_port
    self.response_host = response_host
    self.response_port = response_port
    self.proxy_host = proxy_host
    self.proxy_port = proxy_port

    self.app = app_name
    self.log = get_logger(f"controller.core.{self.app}-commander")
    self.app_url = f"http://{self.app_host}:{self.app_port}/command"

    self.response_queue = queue.Queue()
    self.sent_cmd = None

Methods:

check_response(timeout=0)

Check if a response is present in the queue

Parameters:

Name	Type	Description	Default
timeout	int	Timeout in seconds	0

Returns:

Name	Type	Description
dict	dict	Command response is json

Raises:

Type	Description
NoResponse	Description
ResponseTimeout	Description

Source code in drunc/controller/children_interface/rest_api_child.py

def check_response(self, timeout: int = 0) -> dict:
    """Check if a response is present in the queue

    Args:
        timeout (int, optional): Timeout in seconds

    Returns:
        dict: Command response is json

    Raises:
        NoResponse: Description
        ResponseTimeout: Description

    """
    try:
        # self.log.info(f"Checking for answers from {self.app} {self.sent_cmd}")
        r = self.response_queue.get(block=(timeout > 0), timeout=timeout)
        self.log.info(f"Received reply from {self.app} to {self.sent_cmd}")
        self.sent_cmd = None

    except queue.Empty:
        self.log.info(f"Queue empty! {self.app} to {self.sent_cmd}")
        if not timeout:
            raise NoResponse(
                f"No response available from {self.app} for command {self.sent_cmd}"
            )
        else:
            self.log.error(
                f"Timeout while waiting for a reply from {self.app} for command {self.sent_cmd}"
            )
            raise ResponseTimeout(
                f"Timeout while waiting for a reply from {self.app} for command {self.sent_cmd}"
            )
    return r

RESTAPIChildNode(name, configuration, uri, fsm_configuration, connectivity_service=None)

Bases: ChildNode

Source code in drunc/controller/children_interface/rest_api_child.py

def __init__(
    self,
    name: str,
    configuration: RESTAPIChildNodeConfHandler,
    uri: str,
    fsm_configuration: FSMConfHandler,
    connectivity_service=None,
):
    super().__init__(name, ControlType.REST_API)
    self._state = ClientSideState()
    self.configuration = configuration
    self.fsm_configuration = fsm_configuration
    self.connectivity_service = connectivity_service
    if fsm_configuration:
        fsmch = FSMConfHandler(fsm_configuration)
        self.fsm = FSM(conf=fsmch)

    response_listener_host = socket.gethostname()
    self.response_listener = ResponseListener.get()

    self.app_host, app_port = uri.split(":")
    self.app_port = int(app_port)
    if self.app_port == 0:
        raise DruncSetupException(
            f"Application {name} does not expose a control service in the configuration, or has not advertised itself to the application registry service, or the application registry service is not reachable."
        )

    proxy_host, proxy_port = getattr(self.configuration.data, "proxy", [None, None])
    proxy_port = int(proxy_port) if proxy_port is not None else None

    self.commander = AppCommander(
        app_name=self.name,
        app_host=self.app_host,
        app_port=self.app_port,
        response_host=response_listener_host,
        response_port=self.response_listener.get_port(),
        proxy_host=proxy_host,
        proxy_port=proxy_port,
    )

    self.response_listener.register(self.name, self.commander)

ResponseListener()

Source code in drunc/controller/children_interface/rest_api_child.py

def __init__(self):
    raise DruncSetupException("Call get() instead")

Methods:

register(app, handler) classmethod

Register a new notification handler

:param app: The application :type app: str :param handler: The handler :type handler: { type_description }

:rtype: None

:raises RuntimeError: { exception_description }

Source code in drunc/controller/children_interface/rest_api_child.py

@classmethod
def register(cls, app: str, handler):
    """Register a new notification handler

    :param      app:           The application
    :type       app:           str
    :param      handler:       The handler
    :type       handler:       { type_description }

    :rtype:     None

    :raises     RuntimeError:  { exception_description }
    """
    if app in cls.handlers:
        raise DruncSetupException(
            f"Handler already registered with notification listerner for app {app}"
        )

    cls.handlers[app] = handler

unregister(app) classmethod

De-register a notification handler

Parameters:

Name	Type	Description	Default
app	str	application name	required

Source code in drunc/controller/children_interface/rest_api_child.py

@classmethod
def unregister(cls, app: str):
    """De-register a notification handler

    Args:
        app (str): application name

    """
    if app not in cls.handlers:
        raise DruncException(f"No handler registered for app {app}")
    del cls.handlers[app]

Functions: