Feature: dynamic update intervals (#49)
* Updated workers_manager with ability to dynamically set update_interval * Updated README file with documentation for dynamic update intervals
This commit is contained in:
@@ -100,6 +100,16 @@ Use mosquitto_sub to print all messages
|
|||||||
mosquitto_sub -h localhost -d -t # command also help for me to test MQTT messages
|
mosquitto_sub -h localhost -d -t # command also help for me to test MQTT messages
|
||||||
```
|
```
|
||||||
|
|
||||||
|
**Dynamically Changing the Update Interval**
|
||||||
|
To dynamically change the `update_interval` of a worker, publish a message containing the new interval in seconds at the `update_interval` topic. Note that the `update_interval` will revert back to the value in `config.yaml` when the gateway is restarted.
|
||||||
|
I.E:
|
||||||
|
```
|
||||||
|
# Set a new update interval of 3 minutes
|
||||||
|
mosquitto_pub -h localhost -t 'miflora/update_interval' -m '150'
|
||||||
|
# Set a new update interval of 30 seconds
|
||||||
|
mosquitto_pub -h localhost -t 'mithermometer/update_interval' -m '30'
|
||||||
|
```
|
||||||
|
|
||||||
## Custom worker development
|
## Custom worker development
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
+19
-2
@@ -50,10 +50,15 @@ class WorkersManager:
|
|||||||
self._update_commands.append(command)
|
self._update_commands.append(command)
|
||||||
|
|
||||||
if 'update_interval' in worker_config:
|
if 'update_interval' in worker_config:
|
||||||
self._scheduler.add_job(
|
job_id = '{}_interval_job'.format(worker_name)
|
||||||
|
interval_job = self._scheduler.add_job(
|
||||||
partial(self._queue_command, command), 'interval',
|
partial(self._queue_command, command), 'interval',
|
||||||
seconds=worker_config['update_interval'],
|
seconds=worker_config['update_interval'], id=job_id
|
||||||
)
|
)
|
||||||
|
self._mqtt_callbacks.append((
|
||||||
|
worker_obj.format_topic('update_interval'),
|
||||||
|
partial(self._update_interval_wrapper, command, job_id)
|
||||||
|
))
|
||||||
elif hasattr(worker_obj, 'run'):
|
elif hasattr(worker_obj, 'run'):
|
||||||
_LOGGER.debug("Registered: %s as daemon" % (worker_name))
|
_LOGGER.debug("Registered: %s as daemon" % (worker_name))
|
||||||
self._daemons.append(worker_obj)
|
self._daemons.append(worker_obj)
|
||||||
@@ -100,6 +105,18 @@ class WorkersManager:
|
|||||||
for package in package_names:
|
for package in package_names:
|
||||||
pip_main(['install', '-q', package])
|
pip_main(['install', '-q', package])
|
||||||
|
|
||||||
|
def _update_interval_wrapper(self, command, job_id, client, userdata, c):
|
||||||
|
_LOGGER.debug("Recieved updated interval for %s with: %s", c.topic, c.payload)
|
||||||
|
try:
|
||||||
|
new_interval = int(c.payload)
|
||||||
|
self._scheduler.remove_job(job_id)
|
||||||
|
self._scheduler.add_job(
|
||||||
|
partial(self._queue_command, command), 'interval',
|
||||||
|
seconds=new_interval, id=job_id
|
||||||
|
)
|
||||||
|
except ValueError:
|
||||||
|
_LOGGER.info("New interval invalid, recieved: %s", c.payload)
|
||||||
|
|
||||||
def _on_command_wrapper(self, worker_obj, client, userdata, c):
|
def _on_command_wrapper(self, worker_obj, client, userdata, c):
|
||||||
_LOGGER.debug("on command wrapper for with %s: %s", c.topic, c.payload)
|
_LOGGER.debug("on command wrapper for with %s: %s", c.topic, c.payload)
|
||||||
global_topic_prefix = userdata['global_topic_prefix']
|
global_topic_prefix = userdata['global_topic_prefix']
|
||||||
|
|||||||
Reference in New Issue
Block a user