diff --git a/miniconf.py b/miniconf.py index 40f163e..7f88f65 100644 --- a/miniconf.py +++ b/miniconf.py @@ -14,8 +14,7 @@ import uuid from gmqtt import Client as MqttClient -logger = logging.getLogger(__name__) - +LOGGER = logging.getLogger(__name__) class Miniconf: """An asynchronous API for controlling Miniconf devices using MQTT.""" @@ -34,13 +33,13 @@ class Miniconf: client: A connected MQTT5 client. prefix: The MQTT toptic prefix of the device to control. """ - self.uuid = uuid.uuid1(prefix) + self.uuid = uuid.uuid1() self.request_id = 0 self.client = client self.prefix = prefix self.inflight = {} self.client.on_message = self._handle_response - self.client.subscribe(f'{prefix}/response/#') + self.client.subscribe(f'{prefix}/response/{self.uuid.hex}') def _handle_response(self, _client, _topic, payload, _qos, properties): """Callback function for when messages are received over MQTT. @@ -53,32 +52,13 @@ class Miniconf: properties: A dictionary of properties associated with the message. """ # Extract corrleation data from the properties - try: - correlation_data = json.loads(properties['correlation_data']) - except (json.decoder.JSONDecodeError, KeyError): - logger.warning('Ignoring message with invalid correlation data') - return + correlation_data = json.loads(properties['correlation_data'][0].decode('ascii')) - # Validate the correlation data. - try: - if correlation_data['id'] != self.uuid.hex: - logger.info('Ignoring correlation data for different ID') - return - pid = correlation_data['pid'] - except KeyError: - logger.warning('Ignoring unknown correlation data: %s', correlation_data) - return + # Get the request ID from the correlation data + request_id = correlation_data['request_id'] - if pid not in self.inflight: - logger.warning('Unexpected pid: %s', pid) - return - - try: - response = json.loads(payload) - self.inflight[pid].set_result((response['code'], response['msg'])) - del self.inflight[pid] - except json.decoder.JSONDecodeError: - logger.warning('Invalid response format: %s', payload) + self.inflight[request_id].set_result(json.loads(payload)) + del self.inflight[request_id] async def command(self, path, value): @@ -89,27 +69,25 @@ class Miniconf: value: The value to write to the path. Returns: - (code, msg) tuple as a response to the command. `code` is zero for success and `msg` is - a use-readable message indicating further information. + The response to the command as a dictionary. """ setting_topic = f'{self.prefix}/settings/{path}' - response_topic = f'{self.prefix}/response/{path}' + response_topic = f'{self.prefix}/response/{self.uuid.hex}' # Assign a unique identifier to this update request. - pid = self.request_id + request_id = self.request_id self.request_id += 1 - assert pid not in self.inflight, 'Invalid PID encountered' + assert request_id not in self.inflight, 'Invalid ID encountered' correlation_data = json.dumps({ - 'id': self.uuid.hex, - 'pid': pid, - }) + 'request_id': request_id, + }).encode('ascii') value = json.dumps(value) - logger.info('Sending %s to "%s"', value, setting_topic) + LOGGER.info('Sending %s to "%s"', value, setting_topic) fut = asyncio.get_running_loop().create_future() - self.inflight[pid] = fut + self.inflight[request_id] = fut self.client.publish(setting_topic, payload=value, qos=0, retain=True, response_topic=response_topic, correlation_data=correlation_data) @@ -146,10 +124,10 @@ def main(): interface = await Miniconf.create(args.prefix, args.broker) for key_value in args.settings: path, value = key_value.split("=", 1) - code, response = await interface.command(path, json.loads(value)) + response = await interface.command(path, json.loads(value)) print(f'{path}: {response}') - if code != 0: - return code + if response['code'] != 0: + return response['code'] return 0 sys.exit(loop.run_until_complete(configure_settings())) diff --git a/src/net/miniconf_client.rs b/src/net/miniconf_client.rs index 6e9cf06..e7fec73 100644 --- a/src/net/miniconf_client.rs +++ b/src/net/miniconf_client.rs @@ -75,6 +75,8 @@ where String::from(self.settings_prefix.as_str()); settings_topic.push_str("/#").unwrap(); + // We do not currently handle or process potential subscription failures. Instead, this + // failure will be logged through the stabilizer logging interface. self.mqtt.subscribe(&settings_topic, &[]).unwrap(); self.subscribed = true; }