From 4624987292550e582c76b20bde8ecb18d77c9426 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Wed, 2 Sep 2026 11:49:17 +0200 Subject: [PATCH 1/2] fix(server): dispatch callbacks and queue events --- snap7/server/__init__.py | 154 +++++++++++++++++++++++++++++++-------- tests/test_server.py | 74 ++++++++++++++++--- 2 files changed, 188 insertions(+), 40 deletions(-) diff --git a/snap7/server/__init__.py b/snap7/server/__init__.py index 1a652cc7..e3cbe55a 100644 --- a/snap7/server/__init__.py +++ b/snap7/server/__init__.py @@ -6,25 +6,37 @@ S7CommPlus clients. """ +import logging import socket import struct import sys import threading import time -import logging -from typing import Dict, Optional, List, Callable, Any, Tuple, Type, Union -from types import TracebackType -from enum import IntEnum from ctypes import Array, c_char +from enum import IntEnum +from types import TracebackType +from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union -from ..s7protocol import S7Protocol, S7Function, S7PDUType, S7UserDataGroup, S7UserDataSubfunction from ..datatypes import S7Area, S7WordLen from ..error import S7ConnectionError, S7ProtocolError -from ..type import SrvArea, SrvEvent, Parameter +from ..s7protocol import S7Function, S7PDUType, S7Protocol, S7UserDataGroup, S7UserDataSubfunction +from ..type import Parameter, SrvArea, SrvEvent logger = logging.getLogger(__name__) +# Event codes from the Snap7 server API. +EVC_SERVER_STARTED = 0x00000001 +EVC_SERVER_STOPPED = 0x00000002 +EVC_LISTENER_CANNOT_START = 0x00000004 +EVC_CLIENT_ADDED = 0x00000008 +EVC_CLIENT_NO_ROOM = 0x00000020 +EVC_CLIENT_EXCEPTION = 0x00000040 +EVC_CLIENT_DISCONNECTED = 0x00000080 +EVC_DATA_READ = 0x00020000 +EVC_DATA_WRITE = 0x00040000 + + class ServerState(IntEnum): """S7 server states.""" @@ -97,6 +109,7 @@ def __init__(self, log: bool = True, max_clients: int = 64, **kwargs: object) -> # Event queue for pick_event self._event_queue: List[SrvEvent] = [] + self._event_lock = threading.Lock() # Logging self._log_enabled = log @@ -144,10 +157,7 @@ def start(self, tcp_port: int = 102) -> int: self.server_thread = threading.Thread(target=self._server_loop, daemon=True) self.server_thread.start() - # Add startup event to queue - startup_event = SrvEvent() - startup_event.EvtCode = 0x00010000 # Server started - self._event_queue.append(startup_event) + self._emit_event(EVC_SERVER_STARTED) logger.info(f"S7 Server started on {self.host}:{self.port}") return 0 @@ -158,6 +168,7 @@ def start(self, tcp_port: int = 102) -> int: if self.server_socket: self.server_socket.close() self.server_socket = None + self._emit_event(EVC_LISTENER_CANNOT_START) raise S7ConnectionError(f"Failed to start server: {e}") def stop(self) -> int: @@ -195,6 +206,7 @@ def stop(self) -> int: self.clients.clear() self.client_count = 0 + self._emit_event(EVC_SERVER_STOPPED) logger.info("S7 Server stopped") return 0 @@ -408,10 +420,15 @@ def event_text(self, event: SrvEvent) -> str: Event description string """ event_texts = { - 0x00004000: "Read operation completed", - 0x00004001: "Write operation completed", - 0x00008000: "Client connected", - 0x00008001: "Client disconnected", + EVC_SERVER_STARTED: "Server started", + EVC_SERVER_STOPPED: "Server stopped", + EVC_LISTENER_CANNOT_START: "Listener cannot start", + EVC_CLIENT_ADDED: "Client connected", + EVC_CLIENT_NO_ROOM: "Client rejected: no room", + EVC_CLIENT_EXCEPTION: "Client exception", + EVC_CLIENT_DISCONNECTED: "Client disconnected", + EVC_DATA_READ: "Read operation completed", + EVC_DATA_WRITE: "Write operation completed", } return event_texts.get(event.EvtCode, f"Event code: {event.EvtCode:#08x}") @@ -556,8 +573,9 @@ def pick_event(self) -> Union[SrvEvent, bool]: Returns: Server event if available, False if no events """ - if self._event_queue: - return self._event_queue.pop(0) + with self._event_lock: + if self._event_queue: + return self._event_queue.pop(0) return False def clear_events(self) -> int: @@ -567,9 +585,65 @@ def clear_events(self) -> int: Returns: 0 on success """ - self._event_queue.clear() + with self._event_lock: + self._event_queue.clear() return 0 + def _emit_event( + self, + code: int, + ret_code: int = 0, + param1: int = 0, + param2: int = 0, + param3: int = 0, + param4: int = 0, + *, + sender: int = 0, + notify_read_callback: bool = False, + ) -> None: + """Queue a server event and notify the configured callbacks.""" + event = SrvEvent() + event.EvtTime = int(time.time()) + event.EvtSender = sender + event.EvtCode = code + event.EvtRetCode = ret_code + event.EvtParam1 = param1 + event.EvtParam2 = param2 + event.EvtParam3 = param3 + event.EvtParam4 = param4 + + with self._event_lock: + self._event_queue.append(event) + + if notify_read_callback and self.read_callback: + try: + self.read_callback(event) + except Exception as e: # noqa: BLE001 -- user callbacks must not terminate the server + logger.error(f"Error in read callback: {e}") + + if self.event_callback: + try: + self.event_callback(event) + except Exception as e: # noqa: BLE001 -- user callbacks must not terminate the server + logger.error(f"Error in event callback: {e}") + + @staticmethod + def _event_area(area: S7Area) -> int: + """Convert an S7 wire-area value to the area code used by SrvEvent.""" + return { + S7Area.PE: SrvArea.PE, + S7Area.PA: SrvArea.PA, + S7Area.MK: SrvArea.MK, + S7Area.CT: SrvArea.CT, + S7Area.TM: SrvArea.TM, + S7Area.DB: SrvArea.DB, + }[area] + + @staticmethod + def _event_sender(address: Tuple[str, int]) -> int: + """Encode a client's IPv4 address like the native Snap7 event sender.""" + return int.from_bytes(socket.inet_aton(address[0]), "big") + def _set_log_callback(self) -> None: """Set up default logging callback.""" @@ -597,11 +671,13 @@ def _server_loop(self) -> None: with self.client_lock: if self.client_count >= self.max_clients: logger.warning(f"Rejecting client {address}: maximum of {self.max_clients} clients reached") + self._emit_event(EVC_CLIENT_NO_ROOM, sender=self._event_sender(address)) client_socket.close() continue self.clients.append(client_thread) self.client_count += 1 + self._emit_event(EVC_CLIENT_ADDED, sender=self._event_sender(address)) client_thread.start() except socket.timeout: @@ -649,6 +725,7 @@ def _handle_client(self, client_socket: socket.socket, address: Tuple[str, int]) break except Exception as e: logger.error(f"Error handling client {address}: {e}") + self._emit_event(EVC_CLIENT_EXCEPTION, sender=self._event_sender(address)) break except Exception as e: @@ -670,6 +747,7 @@ def _handle_client(self, client_socket: socket.socket, address: Tuple[str, int]) if hasattr(self, "_upload_contexts"): self._upload_contexts.pop(address, None) + self._emit_event(EVC_CLIENT_DISCONNECTED, sender=self._event_sender(address)) logger.info(f"Client {address} handler finished") def _process_request(self, request_data: bytes, client_address: Tuple[str, int]) -> Optional[bytes]: @@ -800,20 +878,15 @@ def _handle_read_area(self, request: Dict[str, Any], client_address: Tuple[str, data_section = struct.pack(">BBH", 0xFF, 0x04, len(read_data) * 8) + read_data - if self.read_callback: - event = SrvEvent() - event.EvtTime = int(time.time()) - event.EvtSender = 0 - event.EvtCode = 0x00004000 - event.EvtRetCode = 0 - event.EvtParam1 = 1 - event.EvtParam2 = 0 - event.EvtParam3 = len(read_data) - event.EvtParam4 = 0 - try: - self.read_callback(event) - except Exception as e: - logger.error(f"Error in read callback: {e}") + self._emit_event( + EVC_DATA_READ, + param1=self._event_area(area), + param2=db_number, + param3=start, + param4=len(read_data), + sender=self._event_sender(client_address), + notify_read_callback=True, + ) return header + parameters + data_section @@ -861,6 +934,16 @@ def _handle_multi_read_area(self, request: Dict[str, Any], client_address: Tuple if i < item_count - 1 and len(read_data) % 2 != 0: data_parts.append(0x00) + self._emit_event( + EVC_DATA_READ, + param1=self._event_area(area), + param2=db_number, + param3=start, + param4=byte_count, + sender=self._event_sender(client_address), + notify_read_callback=True, + ) + data_len = len(data_parts) header = struct.pack( @@ -1006,6 +1089,15 @@ def _handle_write_area(self, request: Dict[str, Any], client_address: Tuple[str, # Data section (write response) data_section = b"\xff" # Success return code + self._emit_event( + EVC_DATA_WRITE, + param1=self._event_area(area), + param2=db_number, + param3=start, + param4=len(write_data), + sender=self._event_sender(client_address), + ) + return header + parameters + data_section except Exception as e: diff --git a/tests/test_server.py b/tests/test_server.py index d6a6e4e3..aa6f7581 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,18 +1,19 @@ -from ctypes import c_char import logging import socket import time -from datetime import datetime - -import pytest import unittest +from ctypes import c_char +from datetime import datetime from threading import Thread from unittest.mock import MagicMock +import pytest + from snap7.client import Client -from snap7.error import server_errors, error_text, S7ConnectionError -from snap7.server import Server, ServerISOConnection -from snap7.type import SrvEvent, mkEvent, mkLog, SrvArea, Parameter, Block +from snap7.datatypes import S7Area, S7WordLen +from snap7.error import S7ConnectionError, error_text, server_errors +from snap7.server import EVC_DATA_READ, EVC_DATA_WRITE, EVC_SERVER_STARTED, EVC_SERVER_STOPPED, Server, ServerISOConnection +from snap7.type import Block, Parameter, SrvArea, SrvEvent, mkEvent, mkLog logging.basicConfig(level=logging.WARNING) @@ -98,16 +99,71 @@ def test_unregister_area(self) -> None: self.server.unregister_area(area_code, index) def test_events_callback(self) -> None: + events: list[SrvEvent] = [] + def event_call_back(event: SrvEvent) -> None: - logging.debug(event) + events.append(event) self.server.set_events_callback(event_call_back) + self.server.clear_events() + + self.server.register_area(SrvArea.DB, 1, bytearray(8)) + address = ("127.0.0.1", 102) + + read_pdu = self.server.protocol.build_read_request(S7Area.DB, 1, 2, S7WordLen.BYTE, 3) + self.server._handle_read_area(self.server._parse_request(read_pdu), address) + + write_pdu = self.server.protocol.build_write_request(S7Area.DB, 1, 4, S7WordLen.BYTE, b"\x01\x02") + self.server._handle_write_area(self.server._parse_request(write_pdu), address) + + self.assertEqual([event.EvtCode for event in events], [EVC_DATA_READ, EVC_DATA_WRITE]) + self.assertEqual( + [(event.EvtParam1, event.EvtParam2, event.EvtParam3, event.EvtParam4) for event in events], + [(SrvArea.DB, 1, 2, 3), (SrvArea.DB, 1, 4, 2)], + ) + read_event = self.server.pick_event() + write_event = self.server.pick_event() + self.assertIsInstance(read_event, SrvEvent) + self.assertIsInstance(write_event, SrvEvent) + assert isinstance(read_event, SrvEvent) + assert isinstance(write_event, SrvEvent) + self.assertEqual(read_event.EvtCode, EVC_DATA_READ) + self.assertEqual(write_event.EvtCode, EVC_DATA_WRITE) + self.assertFalse(self.server.pick_event()) def test_read_events_callback(self) -> None: + events: list[SrvEvent] = [] + def read_events_call_back(event: SrvEvent) -> None: - logging.debug(event) + events.append(event) self.server.set_read_events_callback(read_events_call_back) + self.server.register_area(SrvArea.DB, 1, bytearray(4)) + + read_pdu = self.server.protocol.build_read_request(S7Area.DB, 1, 0, S7WordLen.BYTE, 4) + self.server._handle_read_area(self.server._parse_request(read_pdu), ("127.0.0.1", 102)) + + self.assertEqual(len(events), 1) + self.assertEqual(events[0].EvtCode, EVC_DATA_READ) + + def test_lifecycle_events_reach_callback_and_queue(self) -> None: + server = Server(log=False) + events: list[SrvEvent] = [] + server.set_events_callback(events.append) + + try: + server.start(tcp_port=0) + server.stop() + finally: + server.destroy() + + self.assertEqual([event.EvtCode for event in events], [EVC_SERVER_STARTED, EVC_SERVER_STOPPED]) + started_event = server.pick_event() + stopped_event = server.pick_event() + assert isinstance(started_event, SrvEvent) + assert isinstance(stopped_event, SrvEvent) + self.assertEqual(started_event.EvtCode, EVC_SERVER_STARTED) + self.assertEqual(stopped_event.EvtCode, EVC_SERVER_STOPPED) def test_pick_event(self) -> None: event = self.server.pick_event() From 0de7891e17b11beaa5c5f894b3704142d82a385b Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Sat, 5 Sep 2026 10:22:29 +0200 Subject: [PATCH 2/2] fix(server): bound event queue and document polling --- snap7/server/__init__.py | 42 +++++++++++++++++++++++++++++----------- tests/test_server.py | 18 +++++++++++++++-- 2 files changed, 47 insertions(+), 13 deletions(-) diff --git a/snap7/server/__init__.py b/snap7/server/__init__.py index d9d31276..c45518e7 100644 --- a/snap7/server/__init__.py +++ b/snap7/server/__init__.py @@ -7,6 +7,7 @@ """ import logging +import queue import socket import struct import sys @@ -36,6 +37,8 @@ EVC_DATA_READ = 0x00020000 EVC_DATA_WRITE = 0x00040000 +_EVENT_QUEUE_CAPACITY = 1024 + class ServerState(IntEnum): """S7 server states.""" @@ -108,8 +111,7 @@ def __init__(self, log: bool = True, max_clients: int = 64, **kwargs: object) -> self.max_clients = max_clients # Event queue for pick_event - self._event_queue: List[SrvEvent] = [] - self._event_lock = threading.Lock() + self._event_queue: queue.Queue[SrvEvent] = queue.Queue(maxsize=_EVENT_QUEUE_CAPACITY) # Logging self._log_enabled = log @@ -568,15 +570,19 @@ def set_cpu_status(self, status: int) -> int: def pick_event(self) -> Union[SrvEvent, bool]: """ - Pick an event from the queue. + Return the oldest queued event without waiting. + + Poll this method when callbacks are not convenient. Applications that + need immediate delivery should use :meth:`set_events_callback` instead; + callbacks do not need to call ``pick_event()`` themselves. Returns: Server event if available, False if no events """ - with self._event_lock: - if self._event_queue: - return self._event_queue.pop(0) - return False + try: + return self._event_queue.get_nowait() + except queue.Empty: + return False def clear_events(self) -> int: """ @@ -585,8 +591,11 @@ def clear_events(self) -> int: Returns: 0 on success """ - with self._event_lock: - self._event_queue.clear() + while True: + try: + self._event_queue.get_nowait() + except queue.Empty: + break return 0 def _emit_event( @@ -612,8 +621,19 @@ def _emit_event( event.EvtParam3 = param3 event.EvtParam4 = param4 - with self._event_lock: - self._event_queue.append(event) + try: + self._event_queue.put_nowait(event) + except queue.Full: + # Never block protocol handling on an application that is not + # draining events. Retain the most recent bounded history. + try: + self._event_queue.get_nowait() + except queue.Empty: + pass + try: + self._event_queue.put_nowait(event) + except queue.Full: + logger.warning("Server event queue is full; dropping event %#x", event.EvtCode) if notify_read_callback and self.read_callback: try: diff --git a/tests/test_server.py b/tests/test_server.py index a12c9a4c..9fec91f7 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -123,14 +123,28 @@ def event_call_back(event: SrvEvent) -> None: ) read_event = self.server.pick_event() write_event = self.server.pick_event() - self.assertIsInstance(read_event, SrvEvent) - self.assertIsInstance(write_event, SrvEvent) assert isinstance(read_event, SrvEvent) assert isinstance(write_event, SrvEvent) + self.assertEqual((read_event, write_event), tuple(events)) self.assertEqual(read_event.EvtCode, EVC_DATA_READ) self.assertEqual(write_event.EvtCode, EVC_DATA_WRITE) self.assertFalse(self.server.pick_event()) + def test_event_queue_is_bounded(self) -> None: + self.server.clear_events() + + for param in range(1025): + self.server._emit_event(EVC_DATA_READ, param1=param) + + first_event = self.server.pick_event() + assert isinstance(first_event, SrvEvent) + self.assertEqual(first_event.EvtParam1, 1) + + events = [first_event] + while event := self.server.pick_event(): + events.append(event) + self.assertEqual(len(events), 1024) + def test_read_events_callback(self) -> None: events: list[SrvEvent] = []