From de1154a35e73bd925f0750a6ceed515d159dd9fd Mon Sep 17 00:00:00 2001 From: Luis Fagundes Date: Fri, 16 Feb 2024 15:11:24 -0300 Subject: [PATCH] Support for dicts in StateChannel.broadcast_instance() This allows optimistic updates from external services. Our use case is that we have a service synchronizing state of IoT devices using MQTT, and we want to reduce latency by broadcasting serialized instances to connected clients before saving to the database. --- rxdjango/signal_handler.py | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/rxdjango/signal_handler.py b/rxdjango/signal_handler.py index 7bbd68a..321b22d 100644 --- a/rxdjango/signal_handler.py +++ b/rxdjango/signal_handler.py @@ -80,18 +80,27 @@ def prepare_save(sender, instance, **kwargs): instance.__old_parent = old_parent def _relay_instance(_layer, instance, tstamp, created): + # This method may receive instances as objectse or already + # serialized. broadcast_instance allows this so microservices + # can send optimistic updates + if not instance: return - if isinstance(instance, models.Model): - instances = [instance] - elif isinstance(instance, models.Manager): - instances = instance.all() + if isinstance(instance, models.Manager) or isinstance(instance, list): + instances = instance else: - raise ProgrammingError() + instances = [instance] for _instance in instances: - serialized = _layer.serialize_instance(_instance, tstamp) + if isinstance(_instance, models.Model): + serialized = _layer.serialize_instance(_instance, tstamp) + elif isinstance(_instance, dict): + serialized = _instance + else: + cls = _instance.__class__ + raise ProgrammingError(f'Cannot relay instance of type {cls}') + serialized['_operation'] = 'create' if created else 'update' self._schedule(serialized, _layer)