"""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()