diff --git a/libervia/backend/bridge/bridge_constructor/bridge_template.ini b/libervia/backend/bridge/bridge_constructor/bridge_template.ini index dfd74a8b..586c05dc 100644 --- a/libervia/backend/bridge/bridge_constructor/bridge_template.ini +++ b/libervia/backend/bridge/bridge_constructor/bridge_template.ini @@ -61,6 +61,14 @@ doc_param_7=extra: extra message information, can have data added by plugins and - info_type: subtype for info messages doc_param_8=%(doc_profile)s +[messages_new] +type=signal +category=core +sig_in=a(sdssa{ss}a{ss}sss)s +doc=Batch of messages received (for MAM history sync) +doc_param_0=messages: list of messages, each as for message_new (uid, timestamp, from_jid, to_jid, message, subject, mess_type, extra) +doc_param_1=%(doc_profile)s + [message_update] type=signal category=core diff --git a/libervia/backend/bridge/dbus_bridge.py b/libervia/backend/bridge/dbus_bridge.py index a8eb4990..3cf1857f 100644 --- a/libervia/backend/bridge/dbus_bridge.py +++ b/libervia/backend/bridge/dbus_bridge.py @@ -175,6 +175,7 @@ class DBusObject(objects.DBusObject): Signal("message_encryption_stopped", "sa{ss}s"), Signal("message_new", "sdssa{ss}a{ss}sss"), Signal("message_update", "ssss"), + Signal("messages_new", "a(sdssa{ss}a{ss}sss)s"), Signal("notification_deleted", "ss"), Signal("notification_new", "sdssssbidss"), Signal("param_update", "ssss"), @@ -627,6 +628,9 @@ class bridge: def message_update(self, uid, message_type, message_data, profile): self._obj.emitSignal("message_update", uid, message_type, message_data, profile) + def messages_new(self, messages, profile): + self._obj.emitSignal("messages_new", messages, profile) + def notification_deleted(self, id, profile): self._obj.emitSignal("notification_deleted", id, profile) diff --git a/libervia/backend/bridge/pb.py b/libervia/backend/bridge/pb.py index efffb8ed..fddb281e 100644 --- a/libervia/backend/bridge/pb.py +++ b/libervia/backend/bridge/pb.py @@ -236,6 +236,9 @@ class bridge: def message_update(self, uid, message_type, message_data, profile): self.send_signal("message_update", uid, message_type, message_data, profile) + def messages_new(self, messages, profile): + self.send_signal("messages_new", messages, profile) + def notification_deleted(self, id, profile): self.send_signal("notification_deleted", id, profile) diff --git a/libervia/backend/core/xmpp.py b/libervia/backend/core/xmpp.py index 339be7b8..6ee4dec6 100644 --- a/libervia/backend/core/xmpp.py +++ b/libervia/backend/core/xmpp.py @@ -460,9 +460,7 @@ class SatXMPPEntity(core_types.SatXMPPEntity): f" Plugins total={total_plugins:.2f}s real={total_real:.2f}s\n" ) - list_d.addCallback( - log_plugin_results - ) + list_d.addCallback(log_plugin_results) # FIXME: we should have a timeout here, and a way to know if a plugin freeze await list_d @@ -1568,6 +1566,7 @@ class SatMessageProtocol(xmppim.MessageProtocol): def __init__(self): xmppim.MessageProtocol.__init__(self) self.queue_manager = MessagesQueueManager(self.process_message) + self.archive_batch: list[dict] = [] def setHandlerParent(self, parent): super().setHandlerParent(parent) @@ -1806,6 +1805,12 @@ class SatMessageProtocol(xmppim.MessageProtocol): ) return data + def flush_archive_batch(self) -> None: + """Flush the archive message batch to the bridge.""" + if self.archive_batch: + G.host.bridge.messages_new(self.archive_batch, self.parent.profile) + self.archive_batch.clear() + def bridge_signal(self, data: MessageData) -> MessageData: """Send signal to frontends for the given message""" try: @@ -1817,17 +1822,34 @@ class SatMessageProtocol(xmppim.MessageProtocol): data["extra"]["encrypted"] = True if data is not None: if self.parent.is_message_printable(data): - G.host.bridge.message_new( - data["uid"], - data["timestamp"], - data["from"].full(), - data["to"].full(), - data["message"], - data["subject"], - data["type"], - data_format.serialise(data["extra"]), - profile=self.parent.profile, - ) + if data["extra"].get("is_from_archive"): + # Archives are sent in batches to avoid flooding the bridge. + self.archive_batch.append( + { + "uid": data["uid"], + "timestamp": data["timestamp"], + "from_jid": data["from"].full(), + "to_jid": data["to"].full(), + "message": data["message"], + "subject": data["subject"], + "mess_type": data["type"], + "extra": data_format.serialise(data["extra"]), + } + ) + if len(self.archive_batch) >= 50: + self.flush_archive_batch() + else: + G.host.bridge.message_new( + data["uid"], + data["timestamp"], + data["from"].full(), + data["to"].full(), + data["message"], + data["subject"], + data["type"], + data_format.serialise(data["extra"]), + profile=self.parent.profile, + ) else: log.debug( "Discarding bridge signal for empty message: {data}".format(data=data) diff --git a/libervia/backend/plugins/plugin_xep_0045.py b/libervia/backend/plugins/plugin_xep_0045.py index 465108e9..482edc95 100644 --- a/libervia/backend/plugins/plugin_xep_0045.py +++ b/libervia/backend/plugins/plugin_xep_0045.py @@ -1018,6 +1018,8 @@ class XEP_0045: except Exception as e: log.exception(f"MAM history sync failed for {room_jid}") + finally: + client.messageProt.flush_archive_batch() def pop_rooms(self, client: SatXMPPEntity) -> list[tuple]: """Remove rooms and return data needed to re-join them.