python/meshtastic/tcp_interface.py

"""TCPInterface class for interfacing with http endpoint
"""
# pylint: disable=R0917
import contextlib
import logging
import socket
import threading
import time
from typing import Optional

from meshtastic.stream_interface import StreamInterface

DEFAULT_TCP_PORT = 4403

# How long close() gives the device to consume what we last wrote before forcing
# the connection down. See close() for why this exists.
GRACEFUL_CLOSE_TIMEOUT = 0.25

logger = logging.getLogger(__name__)


class TCPInterface(StreamInterface):
    """Interface class for meshtastic devices over a TCP link"""

    def __init__(
        self,
        hostname: str,
        debugOut=None,
        noProto: bool = False,
        connectNow: bool = True,
        portNumber: int = DEFAULT_TCP_PORT,
        noNodes: bool = False,
        timeout: int = 300,
    ):
        """Constructor, opens a connection to a specified IP address/hostname

        Keyword Arguments:
            hostname {string} -- Hostname/IP address of the device to connect to
            timeout -- How long to wait for replies (default: 300 seconds)
        """
        self.hostname: str = hostname
        self.portNumber: int = portNumber

        self.socket: Optional[socket.socket] = None
        self.reconnectLock = threading.Lock()

        super().__init__(
            debugOut=debugOut,
            noProto=noProto,
            connectNow=connectNow,
            noNodes=noNodes,
            timeout=timeout,
        )

    def __repr__(self):
        rep = f"TCPInterface({self.hostname!r}"
        if self.debugOut is not None:
            rep += f", debugOut={self.debugOut!r}"
        if self.noProto:
            rep += ", noProto=True"
        if self.socket is None:
            rep += ", connectNow=False"
        if self.portNumber != DEFAULT_TCP_PORT:
            rep += f", portNumber={self.portNumber!r}"
        if self.noNodes:
            rep += ", noNodes=True"
        rep += ")"
        return rep

    def _socket_shutdown(self) -> None:
        """Shutdown the socket.
        Note: Broke out this line so the exception could be unit tested.
        """
        if self.socket is not None:
            self.socket.shutdown(socket.SHUT_RDWR)

    def connect(self) -> None:
        """Connect the interface"""
        self.myConnect()
        super().connect()

    def myConnect(self) -> None:
        """Connect to socket (without attempting to start the interface's receive thread)"""
        logger.debug(f"Connecting to {self.hostname}")  # type: ignore[str-bytes-safe]
        server_address = (self.hostname, self.portNumber)
        self.socket = socket.create_connection(server_address)

    def _wait_for_reader_exit(self, timeout: float) -> None:
        """Wait briefly for the reader thread to drain and exit after a half-close.

        Returns as soon as it exits, or after timeout: the device is not obliged
        to close just because we did.
        """
        rx = getattr(self, "_rxThread", None)
        if rx is None or rx is threading.current_thread():
            return
        with contextlib.suppress(Exception):
            rx.join(timeout)

    def close(self) -> None:
        """Close a connection to the device."""
        logger.debug("Closing TCP stream")
        # Sometimes the socket read might be blocked in the reader thread.
        # Therefore force a shutdown first to unblock reader thread reads.
        self._wantExit = True
        if self.socket is not None:
            # Half-close first. shutdown(SHUT_WR) sends FIN, which tells the
            # device we are done writing and lets it consume what we last wrote
            # before the connection goes away -- typically the admin message a
            # one-shot command such as `--set` just sent.
            #
            # Going straight to shutdown(SHUT_RDWR) + close() while either side
            # still has unread data makes the stack send RST instead. Winsock
            # then discards data the peer had already received but not yet read,
            # so the write is lost; Linux delivers it before reporting
            # ECONNRESET, which is why this only bites on Windows.
            with contextlib.suppress(Exception):
                self.socket.shutdown(socket.SHUT_WR)
            self._wait_for_reader_exit(GRACEFUL_CLOSE_TIMEOUT)

            with contextlib.suppress(
                Exception
            ):  # Ignore errors in shutdown, because we might have a race with the server
                self._socket_shutdown()
            with contextlib.suppress(Exception):
                self.socket.close()

        self.socket = None
        super().close()

    def _writeBytes(self, b: bytes) -> None:
        """Write an array of bytes to our stream"""
        if self.socket is not None:
            try:
                self.socket.sendall(b)
            except OSError as e:
                logger.error(f"Socket send error, reconnecting: {e}")
                if not self._wantExit:
                    self._reconnect()
                raise

    def _readBytes(self, length) -> Optional[bytes]:
        """Read an array of bytes from our stream"""
        if self.socket is not None:
            data = self.socket.recv(length)
            # empty byte indicates a disconnected socket,
            # we need to handle it to avoid an infinite loop reading from null socket
            if data == b"":
                logger.debug("Closed socket, re-connecting")
                if not self._wantExit:
                    self._reconnect()
            return data

        # no socket, break reader thread
        self._wantExit = True
        return None

    def _reconnect(self) -> None:
        """Reconnect to the socket"""
        # Save the socket reference before attempting to acquire the lock.
        sock = self.socket
        start_config = False
        with self.reconnectLock:
            if self._wantExit:
                return
            # Don't reconnect: someone else already did it.
            if sock is not self.socket:
                return

            with contextlib.suppress(Exception):
                self._socket_shutdown()
            if self.socket is not None:
                self.socket.close()
            self.socket = None
            time.sleep(1)
            self.myConnect()
            start_config = True

        if start_config and not self._wantExit and self.socket is not None:
            self._startConfig()