From 29b1b2060e9aaa5afbc97cbf00d9d541859d4f4e Mon Sep 17 00:00:00 2001 From: seaspotter Date: Wed, 23 Sep 2026 07:47:15 +0200 Subject: [PATCH 1/3] external_openwb: replace per-cycle MQTT reconnects with a persistent connection cp threads were timing out (loadvars.py's control_interval/3 budget) because get_values() opened 6 fresh TCP+MQTT connections per cycle to the remote host - already tight before #3976 raised the internal wait from 1s to 2s to fix missed retained topics, which shrank the margin further and made timeouts common on any network jitter. Replaces this with one long-lived, auto-reconnecting connection per remote host (PersistentBrokerClient), shared by all pub_single() calls to that host and kept continuously subscribed instead of resubscribing every cycle. Preserves existing safety semantics: publish() now raises when the socket is down (paho's Client.publish() fails silently otherwise, which would have broken the exception-driven client_error_context shutoff), and wait_for_fresh_data() still returns an empty dict - not stale cached data - when nothing new arrived this cycle, so the "no data received" fault_state warning still fires correctly. --- packages/helpermodules/broker.py | 98 ++++++++++++++- packages/helpermodules/broker_test.py | 117 ++++++++++++++++++ packages/helpermodules/pub.py | 6 +- .../external_openwb/chargepoint_module.py | 24 +--- 4 files changed, 223 insertions(+), 22 deletions(-) create mode 100644 packages/helpermodules/broker_test.py diff --git a/packages/helpermodules/broker.py b/packages/helpermodules/broker.py index ecd5d2d30c..f9281bba6c 100644 --- a/packages/helpermodules/broker.py +++ b/packages/helpermodules/broker.py @@ -1,8 +1,11 @@ import datetime import logging import paho.mqtt.client as mqtt +import threading import time -from typing import Callable +from typing import Any, Callable, Dict, Tuple + +from helpermodules.utils.topic_parser import decode_payload log = logging.getLogger(__name__) @@ -45,6 +48,99 @@ 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] = {} + self._message_event = threading.Event() + 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: + 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) + self._message_event.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(): + raise ConnectionError(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).""" + self._message_event.clear() + deadline = time.monotonic() + max_wait + if not self._message_event.wait(timeout=max(0.0, deadline - time.monotonic())): + return {} + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + self._message_event.clear() + if not self._message_event.wait(timeout=min(settle, remaining)): + break + with self._lock: + return dict(self._received_topics) + + +_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..ecdb20eef7 --- /dev/null +++ b/packages/helpermodules/broker_test.py @@ -0,0 +1,117 @@ +import threading +from unittest.mock import MagicMock, patch + +import pytest + +from helpermodules import broker +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_raises_when_not_connected(self, mock_mqtt_client): + mock_mqtt_client.is_connected.return_value = False + client = PersistentBrokerClient("10.0.0.5", 1883) + + with pytest.raises(ConnectionError): + client.publish("some/topic", "1") + + mock_mqtt_client.publish.assert_not_called() + + 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], + } + + +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}") From 320e386db4e69347054f70bebe1bfe5e4d62d5b6 Mon Sep 17 00:00:00 2001 From: seaspotter Date: Wed, 23 Sep 2026 08:59:39 +0200 Subject: [PATCH 2/3] external_openwb: give the connection-lost fault a real message instead of "Unbekannter Fehler" MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit publish() raised a plain ConnectionError(message), but the existing OSError fault-state translator (helpermodules/exceptions/os.py) only reads e.errno/e.strerror, so the message was silently discarded and fell back to its generic "OSError None: Unbekannter Fehler None" template - confirmed live during today's 3-minute network-outage test. Construct it as ConnectionError(errno.EHOSTUNREACH, message) instead, so it hits that translator's existing EHOSTUNREACH special case and produces the already-established "Die Verbindung zum Host ist fehlgeschlagen. Überprüfe Adresse und Netzwerk." message. --- packages/helpermodules/broker.py | 7 ++++++- packages/helpermodules/broker_test.py | 10 +++++++++- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/packages/helpermodules/broker.py b/packages/helpermodules/broker.py index f9281bba6c..fc7ff520d8 100644 --- a/packages/helpermodules/broker.py +++ b/packages/helpermodules/broker.py @@ -1,4 +1,5 @@ import datetime +import errno import logging import paho.mqtt.client as mqtt import threading @@ -95,7 +96,11 @@ def publish(self, topic: str, payload, qos: int = 0, retain: bool = True) -> Non # 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(): - raise ConnectionError(f"Keine Verbindung zu {self.host}:{self.port}") + # 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: diff --git a/packages/helpermodules/broker_test.py b/packages/helpermodules/broker_test.py index ecdb20eef7..c5f67ffa4e 100644 --- a/packages/helpermodules/broker_test.py +++ b/packages/helpermodules/broker_test.py @@ -1,9 +1,11 @@ +import errno import threading 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 @@ -21,10 +23,16 @@ def test_publish_raises_when_not_connected(self, mock_mqtt_client): mock_mqtt_client.is_connected.return_value = False client = PersistentBrokerClient("10.0.0.5", 1883) - with pytest.raises(ConnectionError): + 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_forwards_when_connected(self, mock_mqtt_client): client = PersistentBrokerClient("10.0.0.5", 1883) From f4e70bae321b7ef7ef8dd930c59e5a3131942bec Mon Sep 17 00:00:00 2001 From: seaspotter Date: Fri, 25 Sep 2026 18:28:21 +0200 Subject: [PATCH 3/3] broker: Verbindungsrace beim ersten Connect und Wakeup-Race bei Duo-Ladepunkten beheben (Review Copilot) publish() wertete is_connected() sofort nach dem Konstruktor aus, obwohl connect_async()/loop_start() asynchron verbinden - der allererste Aufruf an einen neuen Host loeste dadurch nach jedem Neustart praktisch garantiert einen falschen Fehler aus. Neu: 5s Kulanzfrist fuer die Erstverbindung, ein bereits gestandener und dann abgebrochener Connect loest weiterhin sofort aus. wait_for_fresh_data() nutzte ein einziges Event fuer alle Aufrufer. Bei zwei Ladepunkten an einem Secondary (duo_num 0/1, gemeinsame Verbindung) laufen mehrere Aufrufe gleichzeitig - ein Aufrufer konnte dem anderen per eigenem clear() den Wakeup wegnehmen. Jetzt bekommt jeder Aufruf ein eigenes Event. --- packages/helpermodules/broker.py | 55 +++++++++++----- packages/helpermodules/broker_test.py | 91 ++++++++++++++++++++++++++- 2 files changed, 130 insertions(+), 16 deletions(-) diff --git a/packages/helpermodules/broker.py b/packages/helpermodules/broker.py index fc7ff520d8..a828f45355 100644 --- a/packages/helpermodules/broker.py +++ b/packages/helpermodules/broker.py @@ -4,12 +4,18 @@ import paho.mqtt.client as mqtt import threading import time -from typing import Any, Callable, Dict, Tuple +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: @@ -68,7 +74,13 @@ def __init__(self, host: str, port: int) -> None: self._lock = threading.Lock() self._subscribed_topics: set = set() self._received_topics: Dict[str, Any] = {} - self._message_event = threading.Event() + # 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 @@ -80,6 +92,7 @@ def __init__(self, host: str, port: int) -> None: 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: @@ -88,7 +101,9 @@ def _on_connect(self, client, userdata, flags, rc) -> None: def _on_message(self, client, userdata, message) -> None: with self._lock: self._received_topics[message.topic] = decode_payload(message.payload) - self._message_event.set() + 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, @@ -96,6 +111,10 @@ def publish(self, topic: str, payload, qos: int = 0, retain: bool = True) -> Non # 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 - @@ -117,19 +136,25 @@ def wait_for_fresh_data(self, max_wait: float = 2.0, settle: float = 0.1) -> Dic 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).""" - self._message_event.clear() - deadline = time.monotonic() + max_wait - if not self._message_event.wait(timeout=max(0.0, deadline - time.monotonic())): - return {} - while True: - remaining = deadline - time.monotonic() - if remaining <= 0: - break - self._message_event.clear() - if not self._message_event.wait(timeout=min(settle, remaining)): - break + waiter = threading.Event() with self._lock: - return dict(self._received_topics) + 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] = {} diff --git a/packages/helpermodules/broker_test.py b/packages/helpermodules/broker_test.py index c5f67ffa4e..fdf3652ab1 100644 --- a/packages/helpermodules/broker_test.py +++ b/packages/helpermodules/broker_test.py @@ -1,5 +1,6 @@ import errno import threading +import time from unittest.mock import MagicMock, patch import pytest @@ -19,10 +20,23 @@ def mock_mqtt_client(): class TestPersistentBrokerClient: - def test_publish_raises_when_not_connected(self, mock_mqtt_client): + 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") @@ -34,6 +48,16 @@ def test_publish_raises_when_not_connected(self, mock_mqtt_client): 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) @@ -106,6 +130,71 @@ def deliver_second_message(): "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):