Add dynamic configuration management with CONFIGURE/RESET control signals

juturna/cli/commands/_juturna_remote_service.py

@@ -4,12 +4,14 @@
 import threading
 import time
 import itertools
-
 from concurrent import futures
 
+from juturna.components import Message, Node
+from juturna.payloads import ControlPayload, ControlSignal
+
+
 import grpc
 
-from juturna.components import Message, Node
 from juturna.remotizer._remote_context import RequestContext
 from juturna.remotizer._remote_builder import _standalone_builder
 
@@ -168,9 +170,18 @@ def SendAndReceive(self, request: ProtoEnvelope, context):
             request_message.id = tracking_id
 
             if envelope_dict.get('configuration'):
-                _configuration_to_be_applied = envelope_dict.get(
+                configuration_to_be_applied = envelope_dict.get(
                     'configuration', {}
                 )
+                configuration_message = Message(
+                    creator=self.remote_name,
+                    version=request_message.version,
+                    payload=ControlPayload(
+                        signal=ControlSignal.CONFIGURE,
+                        data=configuration_to_be_applied,
+                    ),
+                )
+                self.node.put(configuration_message)
 
             request_message._freeze()
             self.node.put(request_message)

juturna/components/_node.py

@@ -427,6 +427,10 @@ def configure(self): ...
 
     def update(self, message: Message[T_Input]): ...
 
+    def update_config(self, config: dict): ...
+
+    def reset_config(self): ...
+
     def set_on_config(self, prop: str, value: Any): ...
 
     def warmup(self): ...
@@ -527,6 +531,16 @@ def _control(self, message: Message):
                 self._suspended = False
                 self._logger.info('node resumed')
 
+                return
+            case ControlSignal.CONFIGURE:
+                self.update_config(message.payload.data)
+                self._logger.info('node reconfigured')
+
+                return
+            case ControlSignal.RESET:
+                self.reset_config()
+                self._logger.info('node reset')
+
                 return
             case None:
                 return

juturna/payloads/_control_signal.py

@@ -20,3 +20,5 @@ class ControlSignal(IntEnum):
     WARMUP: int = 2
     SUSPEND: int = 3
     RESUME: int = 4
+    CONFIGURE: int = 5
+    RESET: int = 6

juturna/payloads/_payloads.py

@@ -25,6 +25,7 @@ def serialize(obj):
 @dataclass(frozen=True)
 class ControlPayload(BasePayload):
     signal: ControlSignal = ControlSignal.STOP
+    data: dict = field(default_factory=dict)
 
 
 @dataclass(frozen=True)

juturna/remotizer/utils.py

@@ -366,6 +366,7 @@ def deserialize_envelope(envelope: ProtoEnvelope) -> dict[str, Any]:
         'ttl': envelope.ttl,
         'request_type': envelope.request_type,
         'response_type': envelope.response_type,
+        'configuration': dict(envelope.configuration),
         'message': message,
     }
     return envelope_dict

plugins/nodes/proc/_passthrough_identity/passthrough_identity.py

@@ -33,6 +33,14 @@ def __init__(self, delay: int, **kwargs):
         self._delay = delay
         self._transmitted = 0
 
+    def update_config(self, config: dict):
+        """Update the node configuration"""
+        if 'delay' in config:
+            self._delay = config['delay']
+            self.logger.info(
+                f'config updated: delay set to {self._delay} seconds'
+            )
+
     def update(self, message: Message[BasePayload]):
         """Receive a message from downstream, transmit a message upstream"""
         self.logger.info(