Skip to content

server

dev_tool.launcher.ipc.server

log = logging.getLogger(__name__) module-attribute

IPCServer

A server for inter-process communication.

This class provides methods for handling commands from IPC clients.

The constructor for the IPCServer class.

Parameters:

  • ipc_port (int) –

    The port number for the IPC server.

  • shutdown_callback –

    The callback function to call when a shutdown command is received.

  • host (str, default: '127.0.0.1' ) –

    The host address for the IPC server.

Source code in dev_tool/launcher/ipc/server.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(
    self,
    ipc_port: int,
    callback: Callable[[], None],
    host: str = '127.0.0.1'
) -> None:
    """
    The constructor for the IPCServer class.

    :param ipc_port: The port number for the IPC server.
    :param shutdown_callback: The callback function to call when a shutdown command is received.
    :param host: The host address for the IPC server.
    """

    self._stop_event = threading.Event()
    self._thread: threading.Thread | None = None

    self.callback = callback
    self.host = host
    self.ipc_port = ipc_port
    self.server: socket.socket | None = None

callback = callback instance-attribute

host = host instance-attribute

ipc_port = ipc_port instance-attribute

server = None instance-attribute

start

A method that starts the IPC server.

Source code in dev_tool/launcher/ipc/server.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def start(self) -> None:
    """A method that starts the IPC server."""

    self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    socket_address = (self.host, self.ipc_port)
    self.server.bind(socket_address)

    self.server.listen(1)

    self._thread = threading.Thread(
        target=self._handle_connection,
        daemon=True
    )

    self._thread.start()

stop

A method that stops the IPC server.

Source code in dev_tool/launcher/ipc/server.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def stop(self) -> None:
    """A method that stops the IPC server."""

    self._stop_event.set()

    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as handle:
            command = IPCCommand.CLOSE.encode('utf-8')

            socket_address = (self.host, self.ipc_port)
            handle.connect(socket_address)

            handle.sendall(command)
    except Exception as exception:
        message = f'Error sending message to IPC server: {exception}'
        CONTEXT.notification.error_banner(message)

        log.exception(message)

    if self._thread:
        self._thread.join(timeout=5)
        self._thread = None

    if self.server is not None:
        try:
            self.server.close()
        except Exception as exception:
            message = f'Error closing IPC server socket: {exception}'
            CONTEXT.notification.error_banner(message)

            log.exception(message)