DbgMuxConnHandler: new abstract class for connection handlers

sedbgmux/ch/__init__.py

@@ -0,0 +1,2 @@
+from .base import DbgMuxConnHandler
+from .base import DbgMuxConnState

sedbgmux/ch/base.py

@@ -0,0 +1,105 @@
+# This file is a part of sedbgmux, an open source DebugMux client.
+# Copyright (c) 2023  Vadim Yanitskiy <fixeria@osmocom.org>
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+import logging as log
+import queue
+import enum
+import abc
+
+from typing import Any, Optional
+
+from .. import DbgMuxFrame
+
+
+class DbgMuxConnState(enum.Enum):
+    ''' Connection state for DbgMuxConnHandler '''
+    NotEstablished      = enum.auto()
+    Establishing        = enum.auto()
+    Established         = enum.auto()
+
+
+class DbgMuxConnHandler(abc.ABC):
+    ''' Abstract DebugMux connection handler '''
+
+    def __init__(self):
+        self.conn_state: DbgMuxConnState = DbgMuxConnState.NotEstablished
+        self._rx_data_queue: queue.Queue = queue.Queue()
+        self._tx_queue: Optional[queue.Queue] = None
+        self.ConnRef: int = 0xffff
+        self.DPRef: int = 0xffff
+
+    @abc.abstractmethod
+    def _conn_established(self) -> None:
+        ''' Called when a connection has been established '''
+
+    @abc.abstractmethod
+    def _conn_terminated(self) -> None:
+        ''' Called when a connection has been terminated '''
+
+    def send_msg(self, msg_type: DbgMuxFrame.MsgType, msg: Any = b'') -> None:
+        ''' Send a DebugMux message to the target '''
+        assert self._tx_queue is not None
+        self._tx_queue.put((msg_type, msg))
+
+    def send_data(self, data: bytes) -> None:
+        ''' Send connection data to the target '''
+        assert self.conn_state == DbgMuxConnState.Established
+        msg = dict(ConnRef=self.ConnRef, Data=data)
+        self.send_msg(DbgMuxFrame.MsgType.ConnData, msg)
+
+    def establish(self, DPRef: int, txq: queue.Queue) -> None:
+        ''' Establish connection with a DataProvider '''
+        assert self.conn_state == DbgMuxConnState.NotEstablished
+        log.info('Establishing connection with DPRef=0x%04x', DPRef)
+        self.DPRef = DPRef
+        self._tx_queue = txq
+        self.conn_state = DbgMuxConnState.Establishing
+        self.send_msg(DbgMuxFrame.MsgType.ConnEstablish, dict(DPRef=DPRef))
+
+    def terminate(self) -> None:
+        ''' Terminate connection with a DataProvider '''
+        assert self.conn_state == DbgMuxConnState.Established
+        log.info('Terminating connection ConnRef=0x%04x with DPRef=0x%04x',
+                 self.ConnRef, self.DPRef)
+        self.send_msg(DbgMuxFrame.MsgType.ConnTerminate, dict(ConnRef=self.ConnRef))
+
+    def _handle_established(self, ConnRef: int, DataBlockLimit: int) -> None:
+        ''' Called on connection establishment '''
+        assert self.conn_state == DbgMuxConnState.Establishing
+        log.info('Connection established: DPRef=0x%04x, ConnRef=0x%04x',
+                 self.DPRef, ConnRef)
+        self.conn_state = DbgMuxConnState.Established
+        self.ConnRef = ConnRef
+        self._conn_established()
+
+    def _handle_terminated(self) -> None:
+        ''' Called on connection termination '''
+        assert self.conn_state == DbgMuxConnState.Established
+        log.info('Connection terminated: DPRef=0x%04x, ConnRef=0x%04x',
+                 self.DPRef, self.ConnRef)
+        self.conn_state = DbgMuxConnState.NotEstablished
+        # TODO: reset the internal state?
+        self._conn_terminated()
+
+    def _handle_data(self, data: bytes) -> None:
+        ''' Called on reciept of connection data '''
+        self._rx_data_queue.put(data)
+
+    def _handle_flow_control(self, DataBlockLimit: int):
+        ''' Called on reciept of FlowControl message '''
+        # TODO: purpose is not yet understood