diff --git a/packages/helpermodules/broker.py b/packages/helpermodules/broker.py index ecd5d2d30c..a828f45355 100644 --- a/packages/helpermodules/broker.py +++ b/packages/helpermodules/broker.py @@ -1,11 +1,21 @@ import datetime +import errno import logging import paho.mqtt.client as mqtt +import threading import time -from typing import Callable +from typing import Any, Callable, Dict, List, Tuple + +from helpermodules.utils.topic_parser import decode_payload log = logging.getLogger(__name__) +# Zeitfenster nach dem ersten Verbindungsversuch, in dem "noch nicht verbunden" nicht als Fehler +# gilt: connect_async()/loop_start() verbinden im Hintergrund, is_connected() ist direkt danach +# so gut wie immer noch False. Ohne Kulanzfrist würde der allererste publish() an einen neuen Host +# nach jedem Neustart praktisch garantiert einen falschen Fehler auslösen. +INITIAL_CONNECT_GRACE = 5.0 + def get_name_suffix() -> str: with open('/proc/cpuinfo', 'r') as f: @@ -45,6 +55,122 @@ def disconnect(self) -> None: log.info(f"Verbindung von Client {self.name} geschlossen.") +class PersistentBrokerClient: + """Lang lebende, automatisch wiederverbindende MQTT-Verbindung zu einem Host (typischerweise ein + Secondary-openWB), die über mehrere Regelzyklen hinweg wiederverwendet wird. + + Ersetzt für publish/subscribe zu externen Hosts das bisherige Muster "pro Aufruf neu verbinden, + kurz warten, wieder trennen" (paho publish.single() bzw. BrokerClient.start_finite_loop()): jeder + dieser Auf- und Abbauten ist ein eigener TCP+MQTT-Handshake, und mehrere davon pro Regelzyklus + (z.B. 5x publish + 1x subscribe für einen externen Ladepunkt) mussten bisher gemeinsam in das enge + Timeout von control_interval/3 passen - schon normaler Netzwerk-Jitter zur Gegenstelle konnte das + reißen. Mit einer dauerhaft offenen, im Hintergrund automatisch wiederverbundenen Verbindung fällt + dieser Auf-/Abbau pro Zyklus komplett weg. + """ + + def __init__(self, host: str, port: int) -> None: + self.host = host + self.port = port + self._lock = threading.Lock() + self._subscribed_topics: set = set() + self._received_topics: Dict[str, Any] = {} + # eigenes Event je wartendem Aufrufer statt eines gemeinsamen: bei zwei Ladepunkten am selben + # Host (duo_num 0/1, gemeinsame Verbindung) laufen mehrere wait_for_fresh_data()-Aufrufe + # gleichzeitig auf demselben PersistentBrokerClient - ein gemeinsames Event hätte dazu geführt, + # dass ein Aufrufer das Event im eigenen clear() dem anderen wegnimmt (verpasster Wakeup). + self._waiters: List[threading.Event] = [] + self._ever_connected = False + self._created_at = time.monotonic() + self.client = mqtt.Client(f"openWB-persistent-{host}-{port}-{get_name_suffix()}") + self.client.on_connect = self._on_connect + self.client.on_message = self._on_message + self.client.reconnect_delay_set(min_delay=1, max_delay=30) + try: + self.client.connect_async(host, port) + self.client.loop_start() + except Exception: + log.exception(f"Fehler beim Verbindungsaufbau zu {host}:{port}") + + def _on_connect(self, client, userdata, flags, rc) -> None: + self._ever_connected = True + with self._lock: + topics = list(self._subscribed_topics) + for topic in topics: + client.subscribe(topic) + + def _on_message(self, client, userdata, message) -> None: + with self._lock: + self._received_topics[message.topic] = decode_payload(message.payload) + waiters = list(self._waiters) + for waiter in waiters: + waiter.set() + + def publish(self, topic: str, payload, qos: int = 0, retain: bool = True) -> None: + # anders als paho publish.single() wirft Client.publish() bei fehlender Verbindung keine Exception, + # sondern liefert nur einen Fehlercode zurueck - hier bewusst in eine Exception uebersetzt, damit + # Aufrufer (insbesondere ErrorTimerContext/client_error_context) eine nicht erreichbare Gegenstelle + # weiterhin wie bisher als Fehler erkennen und nach 60s die Sicherheitsabschaltung greifen kann. + if not self.client.is_connected(): + if not self._ever_connected and time.monotonic() - self._created_at < INITIAL_CONNECT_GRACE: + # Erstverbindung läuft im Hintergrund noch (siehe INITIAL_CONNECT_GRACE weiter oben) - + # das ist kein Fehlerfall, Nachricht wird verworfen, naechster Zyklus greift wieder. + return + # errno.EHOSTUNREACH statt nur einer Nachricht, damit der bestehende OSError-Handler + # (helpermodules/exceptions/os.py) die schon vorhandene, uebersetzte Meldung "Die Verbindung + # zum Host ist fehlgeschlagen..." verwendet, statt auf "Unbekannter Fehler" zurueckzufallen - + # der greift nur ueber e.errno/e.strerror, eine reine Textnachricht wuerde er ignorieren. + raise ConnectionError(errno.EHOSTUNREACH, f"Keine Verbindung zu {self.host}:{self.port}") + self.client.publish(topic, payload, qos=qos, retain=retain) + + def ensure_subscribed(self, topic_filter: str) -> None: + with self._lock: + if topic_filter in self._subscribed_topics: + return + self._subscribed_topics.add(topic_filter) + self.client.subscribe(topic_filter) + + def wait_for_fresh_data(self, max_wait: float = 2.0, settle: float = 0.1) -> Dict[str, Any]: + """Wartet auf mindestens eine neue Nachricht und danach auf eine kurze Ruhephase ohne weitere + Nachrichten (Sammeln des restlichen Nachrichten-Schwalls, z.B. alle retained Topics nach einem + Reconnect), insgesamt begrenzt auf `max_wait`. Liefert bei mindestens einer neuen Nachricht den + gesamten aktuellen Stand aller je empfangenen Topics zurück (nicht nur die seit dem Aufruf neu + eingetroffenen) - sonst ein leeres Dict, damit Aufrufer "keine aktuellen Daten diesen Zyklus" + weiterhin wie bisher von "Daten vorhanden" unterscheiden können (z.B. für Fault-State-Meldungen).""" + waiter = threading.Event() + with self._lock: + self._waiters.append(waiter) + try: + deadline = time.monotonic() + max_wait + if not waiter.wait(timeout=max(0.0, deadline - time.monotonic())): + return {} + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + waiter.clear() + if not waiter.wait(timeout=min(settle, remaining)): + break + with self._lock: + return dict(self._received_topics) + finally: + with self._lock: + self._waiters.remove(waiter) + + +_persistent_clients: Dict[Tuple[str, int], PersistentBrokerClient] = {} +_persistent_clients_lock = threading.Lock() + + +def get_persistent_broker_client(host: str, port: int) -> PersistentBrokerClient: + key = (host, port) + with _persistent_clients_lock: + client = _persistent_clients.get(key) + if client is None: + client = PersistentBrokerClient(host, port) + _persistent_clients[key] = client + return client + + class InternalBrokerPublisher: def __init__(self) -> None: try: diff --git a/packages/helpermodules/broker_test.py b/packages/helpermodules/broker_test.py new file mode 100644 index 0000000000..fdf3652ab1 --- /dev/null +++ b/packages/helpermodules/broker_test.py @@ -0,0 +1,214 @@ +import errno +import threading +import time +from unittest.mock import MagicMock, patch + +import pytest + +from helpermodules import broker +from helpermodules.exceptions.os import handle_os_error +from helpermodules.broker import PersistentBrokerClient, get_persistent_broker_client + + +@pytest.fixture +def mock_mqtt_client(): + with patch("helpermodules.broker.mqtt.Client") as mock_client_cls: + instance = MagicMock() + instance.is_connected.return_value = True + mock_client_cls.return_value = instance + yield instance + + +class TestPersistentBrokerClient: + def test_publish_does_not_raise_during_initial_connect_grace_period(self, mock_mqtt_client): + """connect_async()/loop_start() verbinden im Hintergrund - direkt nach dem Konstruktor ist + is_connected() so gut wie immer noch False, ohne dass das ein echter Fehler waere (siehe + Review-Kommentar zu PR #3996: der allererste publish() an einen neuen Host loeste sonst nach + jedem Neustart praktisch garantiert einen falschen Fehler aus).""" + mock_mqtt_client.is_connected.return_value = False + client = PersistentBrokerClient("10.0.0.5", 1883) + + client.publish("some/topic", "1") + + mock_mqtt_client.publish.assert_not_called() + + def test_publish_raises_once_initial_connect_grace_period_expires(self, mock_mqtt_client): + mock_mqtt_client.is_connected.return_value = False + client = PersistentBrokerClient("10.0.0.5", 1883) + client._created_at -= broker.INITIAL_CONNECT_GRACE + 1 + + with pytest.raises(ConnectionError) as excinfo: + client.publish("some/topic", "1") + + mock_mqtt_client.publish.assert_not_called() + # errno muss gesetzt sein, sonst faellt der bestehende OSError-Handler + # (helpermodules/exceptions/os.py) auf die nichtssagende "Unbekannter Fehler"-Meldung zurueck, + # statt die schon vorhandene, uebersetzte "Verbindung zum Host fehlgeschlagen"-Meldung zu nutzen. + assert excinfo.value.errno == errno.EHOSTUNREACH + assert handle_os_error(excinfo.value) == ( + "Die Verbindung zum Host ist fehlgeschlagen. Überprüfe Adresse und Netzwerk.") + + def test_publish_raises_immediately_once_connection_has_dropped(self, mock_mqtt_client): + """Eine Verbindung, die schon einmal stand und dann abbricht, ist ein echter Ausfall - dafuer + gibt es keine Kulanzfrist, das muss weiterhin sofort erkannt werden.""" + client = PersistentBrokerClient("10.0.0.5", 1883) + client._on_connect(mock_mqtt_client, None, None, 0) + mock_mqtt_client.is_connected.return_value = False + + with pytest.raises(ConnectionError): + client.publish("some/topic", "1") + + def test_publish_forwards_when_connected(self, mock_mqtt_client): + client = PersistentBrokerClient("10.0.0.5", 1883) + + client.publish("some/topic", "1", qos=0, retain=True) + + mock_mqtt_client.publish.assert_called_once_with("some/topic", "1", qos=0, retain=True) + + def test_ensure_subscribed_is_idempotent(self, mock_mqtt_client): + client = PersistentBrokerClient("10.0.0.5", 1883) + + client.ensure_subscribed("openWB/internal_chargepoint/0/get/#") + client.ensure_subscribed("openWB/internal_chargepoint/0/get/#") + client.ensure_subscribed("openWB/internal_chargepoint/1/get/#") + + assert mock_mqtt_client.subscribe.call_count == 2 + + def test_on_connect_resubscribes_all_known_topics(self, mock_mqtt_client): + client = PersistentBrokerClient("10.0.0.5", 1883) + client.ensure_subscribed("topic/a") + client.ensure_subscribed("topic/b") + mock_mqtt_client.subscribe.reset_mock() + + client._on_connect(mock_mqtt_client, None, None, 0) + + assert mock_mqtt_client.subscribe.call_count == 2 + + def test_wait_for_fresh_data_returns_empty_dict_when_nothing_arrives(self, mock_mqtt_client): + client = PersistentBrokerClient("10.0.0.5", 1883) + + result = client.wait_for_fresh_data(max_wait=0.05, settle=0.01) + + assert result == {} + + def test_wait_for_fresh_data_returns_accumulated_state_once_a_message_arrives(self, mock_mqtt_client): + client = PersistentBrokerClient("10.0.0.5", 1883) + + def deliver_message(): + message = MagicMock() + message.topic = "openWB/internal_chargepoint/0/get/power" + message.payload = b"1500" + client._on_message(mock_mqtt_client, None, message) + + threading.Timer(0.02, deliver_message).start() + + result = client.wait_for_fresh_data(max_wait=1, settle=0.05) + + assert result == {"openWB/internal_chargepoint/0/get/power": 1500} + + def test_wait_for_fresh_data_keeps_previously_received_topics(self, mock_mqtt_client): + """Ein Zyklus mit nur teilweise neuen Daten darf frueher empfangene Topics nicht verlieren - + genau das war die Ursache des Bugs, den PR #3976 zu beheben versuchte.""" + client = PersistentBrokerClient("10.0.0.5", 1883) + first_message = MagicMock() + first_message.topic = "openWB/internal_chargepoint/0/get/power" + first_message.payload = b"1500" + client._on_message(mock_mqtt_client, None, first_message) + + def deliver_second_message(): + message = MagicMock() + message.topic = "openWB/internal_chargepoint/0/get/currents" + message.payload = b"[1.0, 1.0, 1.0]" + client._on_message(mock_mqtt_client, None, message) + + threading.Timer(0.02, deliver_second_message).start() + + result = client.wait_for_fresh_data(max_wait=1, settle=0.05) + + assert result == { + "openWB/internal_chargepoint/0/get/power": 1500, + "openWB/internal_chargepoint/0/get/currents": [1.0, 1.0, 1.0], + } + + def test_wait_for_fresh_data_registers_and_deregisters_its_own_waiter(self, mock_mqtt_client): + client = PersistentBrokerClient("10.0.0.5", 1883) + + client.wait_for_fresh_data(max_wait=0.02, settle=0.01) + + assert client._waiters == [] + + def test_on_message_notifies_all_concurrent_waiters(self, mock_mqtt_client): + """Bei zwei Ladepunkten an einem Secondary (duo_num 0/1, gemeinsame Verbindung) laufen + mehrere wait_for_fresh_data()-Aufrufe gleichzeitig auf demselben PersistentBrokerClient - + eine eingehende Nachricht muss alle wecken, nicht nur den zuerst registrierten.""" + client = PersistentBrokerClient("10.0.0.5", 1883) + waiter_a = threading.Event() + waiter_b = threading.Event() + client._waiters.extend([waiter_a, waiter_b]) + + message = MagicMock() + message.topic = "some/topic" + message.payload = b"1" + client._on_message(mock_mqtt_client, None, message) + + assert waiter_a.is_set() + assert waiter_b.is_set() + + def test_wait_for_fresh_data_both_duo_waiters_receive_their_data(self, mock_mqtt_client): + """Reproduziert das im Review zu PR #3996 beschriebene Szenario: zwei Ladepunkte am selben + Secondary (duo_num 0/1) rufen wait_for_fresh_data() auf demselben PersistentBrokerClient + nahezu gleichzeitig auf. Mit einem einzigen gemeinsamen Event konnte ein Aufrufer per eigenem + clear() dem anderen den Wakeup wegnehmen (verpasste Daten trotz eingetroffener Nachricht) - + mit einem Event je Aufrufer kann das nicht mehr passieren.""" + client = PersistentBrokerClient("10.0.0.5", 1883) + results = {} + + def waiter(name): + results[name] = client.wait_for_fresh_data(max_wait=1, settle=0.05) + + t1 = threading.Thread(target=waiter, args=("cp0",)) + t2 = threading.Thread(target=waiter, args=("cp1",)) + t1.start() + t2.start() + + deadline = time.monotonic() + 1 + while len(client._waiters) < 2 and time.monotonic() < deadline: + time.sleep(0.005) + assert len(client._waiters) == 2 + + for topic, payload in [ + ("openWB/internal_chargepoint/0/get/power", b"1500"), + ("openWB/internal_chargepoint/1/get/power", b"2500"), + ]: + message = MagicMock() + message.topic = topic + message.payload = payload + client._on_message(mock_mqtt_client, None, message) + + t1.join(timeout=2) + t2.join(timeout=2) + + expected = { + "openWB/internal_chargepoint/0/get/power": 1500, + "openWB/internal_chargepoint/1/get/power": 2500, + } + assert results["cp0"] == expected + assert results["cp1"] == expected + + +class TestGetPersistentBrokerClient: + def test_returns_same_instance_for_same_host_and_port(self, mock_mqtt_client): + broker._persistent_clients.clear() + + first = get_persistent_broker_client("10.0.0.5", 1883) + second = get_persistent_broker_client("10.0.0.5", 1883) + + assert first is second + + def test_returns_different_instance_for_different_host(self, mock_mqtt_client): + broker._persistent_clients.clear() + + first = get_persistent_broker_client("10.0.0.5", 1883) + second = get_persistent_broker_client("10.0.0.6", 1883) + + assert first is not second diff --git a/packages/helpermodules/pub.py b/packages/helpermodules/pub.py index 1286da7ad9..5e5fcf7b42 100644 --- a/packages/helpermodules/pub.py +++ b/packages/helpermodules/pub.py @@ -1,8 +1,7 @@ import json import logging -import paho.mqtt.publish as publish -from helpermodules.broker import InternalBrokerPublisher +from helpermodules.broker import InternalBrokerPublisher, get_persistent_broker_client log = logging.getLogger(__name__) @@ -50,4 +49,5 @@ def pub_single(topic: str, payload, hostname: str = "localhost", port: int = 188 Pub().pub(topic, payload, qos=0, no_json=no_json, retain=retain) return - publish.single(topic, payload if no_json else json.dumps(payload), hostname=hostname, port=port, retain=retain) + get_persistent_broker_client(hostname, port).publish( + topic, payload if no_json else json.dumps(payload), qos=0, retain=retain) diff --git a/packages/modules/chargepoints/external_openwb/chargepoint_module.py b/packages/modules/chargepoints/external_openwb/chargepoint_module.py index db60fb8d5b..db16055d0c 100644 --- a/packages/modules/chargepoints/external_openwb/chargepoint_module.py +++ b/packages/modules/chargepoints/external_openwb/chargepoint_module.py @@ -3,10 +3,9 @@ from control import data from helpermodules import pub, timecheck -from helpermodules.broker import BrokerClient +from helpermodules.broker import get_persistent_broker_client from helpermodules.utils import get_default from helpermodules.utils.error_handling import CP_ERROR, ErrorTimerContext -from helpermodules.utils.topic_parser import decode_payload from modules.chargepoints.external_openwb.config import OpenWBSeries from modules.common.abstract_chargepoint import AbstractChargepoint from modules.common.abstract_device import DeviceDescriptor @@ -59,22 +58,11 @@ def parse_received_topics(value: str): pub.pub_single(f"openWB/set/isss/parentCPlp{self.config.configuration.duo_num + 1}", str(num), hostname=ip_address) - def on_connect(client, userdata, flags, rc): - client.subscribe(f"openWB/internal_chargepoint/{self.config.configuration.duo_num}/get/#") - - def on_message(client, userdata, message): - received_topics.update({message.topic: decode_payload(message.payload)}) - - received_topics = {} - # Höheres Timeout als der Default: muss zuverlässig alle ~20 retained Topics über eine - # echte Netzwerkverbindung zu einer ggf. gerade stark ausgelasteten Gegenstelle einsammeln - # (zB während einer laufenden Ladung), nicht nur ein oder zwei lokale Werte. - BrokerClient(f"subscribeSeriesChargepoint{self.config.id}", - on_connect, - on_message, - host=self.config.configuration.ip_address, - port=1886 if self.config.configuration.ip_address == "localhost" else 1883 - ).start_finite_loop(timeout=2) + broker_client = get_persistent_broker_client( + ip_address, 1886 if ip_address == "localhost" else 1883) + broker_client.ensure_subscribed( + f"openWB/internal_chargepoint/{self.config.configuration.duo_num}/get/#") + received_topics = broker_client.wait_for_fresh_data(max_wait=2) if received_topics: log.debug(f"Empfange MQTT Daten für Ladepunkt {self.config.id}: {received_topics}")