Use Notifications instead of Requests

Author: BenediktBurger

pyleco/utils/extended_data_publisher.py

@@ -23,8 +23,11 @@
 #
 
 from __future__ import annotations
+from json import JSONDecodeError
 from typing import Any, Callable, Generator, Union
 
+from ..json_utils.errors import JSONRPCError, NODE_UNKNOWN, RECEIVER_UNKNOWN
+from ..json_utils.json_objects import Notification
 from ..core.message import Message, MessageTypes
 from ..core.data_message import DataMessage
 from ..json_utils.rpc_generator import RPCGenerator
@@ -57,11 +60,10 @@ def convert_data_message_to_messages(
         self, data_message: DataMessage, receivers: Union[set[Union[bytes, str]], set[bytes]],
     ) -> Generator[Message, Any, Any]:
         cid = data_message.conversation_id
-        data = self.rpc_generator.build_request_str(method="add_subscription_message")
         for receiver in receivers:
             yield Message(
                 receiver=receiver,
-                data=data,
+                data=Notification("add_subscription_message"),
                 conversation_id=cid,
                 additional_payload=data_message.payload,
                 message_type=MessageTypes.JSON,
@@ -72,3 +74,32 @@ def send_message(self, message: DataMessage) -> None:
         for msg in self.convert_data_message_to_messages(message, self.subscribers):
             # ideas: change to ask and check, whether it succeeded, otherwise remove subscriber
             self.send_control_message(msg)
+
+    # TODO should unregister subscribers to which a message could not be forwarded
+    def handle_json_error(self, message: Message) -> None:
+        """Unregister unavailable subscribers.
+        
+        Call this method from the message handler, for example.
+        """
+        try:
+            data: dict[str, Any] = message.data  # type: ignore
+        except JSONDecodeError as exc:
+            self.log.exception(f"Could not decode json message {message}", exc_info=exc)
+            return
+        try:
+            self.rpc_generator.get_result_from_response(data)
+        except JSONRPCError as exc:
+            error_code = exc.rpc_error.code
+            try:
+                error_data = exc.rpc_error.data
+            except AttributeError:
+                return
+            if error_code == RECEIVER_UNKNOWN:
+                self.unregister_subscriber(error_data)
+            if error_code == NODE_UNKNOWN:
+                if isinstance(error_data, str):
+                    error_data = error_data.encode()
+                for subscriber in self.subscribers:
+                    if subscriber.startswith(error_data):
+                        self.unregister_subscriber(subscriber)
+

tests/utils/test_extended_data_publisher.py

@@ -27,6 +27,7 @@
 from pyleco.test import FakeContext
 from pyleco.core.message import Message, MessageTypes
 from pyleco.core.data_message import DataMessage
+from pyleco.json_utils.json_objects import Notification
 from pyleco.utils.extended_data_publisher import ExtendedDataPublisher
 
 
@@ -86,11 +87,10 @@ def test_unregister_subscribers(publisher: ExtendedDataPublisher):
 @pytest.mark.parametrize("receivers", (set(), {b"abc"}, {b"abc", b"def"}, {"string"}))
 def test_convert(publisher: ExtendedDataPublisher, receivers, data_message: DataMessage):
     msgs = publisher.convert_data_message_to_messages(data_message, receivers=receivers)
-    assert len(receivers) == len(list(msgs)), "The lengths of receivers and messages do not match."
-    for rec, msg in zip(receivers, msgs):
+    for rec, msg in zip(receivers, msgs, strict=True):
         assert msg == Message(
             receiver=rec,
-            data={"id": 1, "method": "add_subscription_message", "jsonrpc": "2.0"},
+            data=Notification(method="add_subscription_message"),
             conversation_id=CID,
             message_type=MessageTypes.JSON,
             additional_payload=data_message.payload,
@@ -109,7 +109,7 @@ def test_send_message(publisher: ExtendedDataPublisher, data_message: DataMessag
     assert messages == [
         Message(
             "abc",
-            data={"id": 1, "method": "add_subscription_message", "jsonrpc": "2.0"},
+            data=Notification(method="add_subscription_message"),
             conversation_id=CID,
             message_type=MessageTypes.JSON,
             additional_payload=data_message.payload,