The actor offers publishing via control protocol (+test)

Author: BenediktBurger

pyleco/actors/actor.py

@@ -29,7 +29,7 @@
 import zmq
 
 from ..utils.message_handler import MessageHandler
-from ..utils.data_publisher import DataPublisher
+from ..utils.extended_data_publisher import ExtendedDataPublisher as DataPublisher
 from ..utils.timers import RepeatingTimer
 
 
@@ -101,7 +101,9 @@ def __init__(
         self.pipeL.connect(f"inproc://listenerPipe:{pipe_port}")
 
         self.timer = RepeatingTimer(interval=periodic_reading, function=self.queue_readout)
-        self.publisher = DataPublisher(full_name=name, log=self.root_logger)
+        self.publisher = DataPublisher(
+            full_name=name, log=self.root_logger, send_message_method=self.send_message
+        )
 
         if auto_connect:
             self.connect(**auto_connect)
@@ -118,6 +120,8 @@ def register_rpc_methods(self) -> None:
         self.register_rpc_method(self.set_polling_interval)
         self.register_rpc_method(self.connect)
         self.register_rpc_method(self.disconnect)
+        self.register_rpc_method(self.register_subscriber)
+        self.register_rpc_method(self.unregister_subscriber)
 
     def register_device_method(self, method: Callable) -> None:
         """Make a device method available via RPC. The method name is prefixed with `device.`."""
@@ -271,3 +275,9 @@ def call_action(self, action: str, args: Optional[Sequence] = None,
         for attr in path[:-1]:
             obj = getattr(obj, attr)
         return getattr(obj, path[-1])(*args, **kwargs)
+
+    def register_subscriber(self):
+        self.publisher.register_subscriber(self.current_message.sender)
+
+    def unregister_subscriber(self):
+        self.publisher.unregister_subscriber(self.current_message.sender)

tests/acceptance_tests/test_director_actor.py

@@ -29,6 +29,7 @@
 
 import pytest
 
+from pyleco.core.message import MessageTypes
 from pyleco.coordinators.coordinator import Coordinator
 from pyleco.actors.actor import Actor
 from pyleco.directors.director import Director
@@ -83,10 +84,14 @@ def binary_method_created(additional_payload: list[bytes]) -> tuple[None, list[b
         """Receive binary data and return it. Create binary method by registering it."""
         return None, [additional_payload[0] * 2]
 
+    def publish():
+        actor.publisher.send_data("super content")
+
     actor.register_rpc_method(binary_method_manually)
     actor.register_binary_rpc_method(
         binary_method_created, accept_binary_input=True, return_binary_output=True
     )
+    actor.register_rpc_method(publish)
     actor.connect()
     actor.rpc.method()(actor.device.triple)
     actor.register_device_method(actor.device.triple)
@@ -165,3 +170,23 @@ def test_binary_data_transfer_created(director: Director):
     assert director.ask_rpc(
         method="binary_method_created", additional_payload=[b"123"], extract_additional_payload=True
     ) == (None, [b"123123"])
+
+
+def test_data_via_control_protocol(director: Director):
+    # act
+    director.ask_rpc("register_subscriber")
+    director.ask_rpc("publish")
+
+    msg = director.communicator.read_message()
+    director.communicator.send(
+        receiver=director.actor,  # type: ignore
+        data={"jsonrpc": "2.0", "id": 1, "result": None},
+        conversation_id=msg.conversation_id,
+        message_type=MessageTypes.JSON,
+    )
+
+    # teardown
+    director.ask_rpc("unregister_subscriber")
+
+    assert msg.data == {"jsonrpc": "2.0", "id": 1, "method": "set_subscription_message"}
+    assert msg.payload[1:] == [b'super content']