diff --git a/juturna/cli/commands/_juturna_remote_service.py b/juturna/cli/commands/_juturna_remote_service.py index 10416a67..cfbc9c7f 100644 --- a/juturna/cli/commands/_juturna_remote_service.py +++ b/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) diff --git a/juturna/components/_node.py b/juturna/components/_node.py index d3931c80..55c5fe0b 100644 --- a/juturna/components/_node.py +++ b/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 diff --git a/juturna/payloads/_control_signal.py b/juturna/payloads/_control_signal.py index 13093094..a61d9084 100644 --- a/juturna/payloads/_control_signal.py +++ b/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 diff --git a/juturna/payloads/_payloads.py b/juturna/payloads/_payloads.py index 730a57fd..91d31062 100644 --- a/juturna/payloads/_payloads.py +++ b/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) diff --git a/juturna/remotizer/utils.py b/juturna/remotizer/utils.py index 4765f711..af586eee 100644 --- a/juturna/remotizer/utils.py +++ b/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 diff --git a/plugins/nodes/proc/_passthrough_identity/passthrough_identity.py b/plugins/nodes/proc/_passthrough_identity/passthrough_identity.py index 86ffb71c..58892698 100644 --- a/plugins/nodes/proc/_passthrough_identity/passthrough_identity.py +++ b/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(