aboutsummaryrefslogtreecommitdiff
path: root/libdino/src/service/message_processor.vala
diff options
context:
space:
mode:
Diffstat (limited to 'libdino/src/service/message_processor.vala')
-rw-r--r--libdino/src/service/message_processor.vala156
1 files changed, 113 insertions, 43 deletions
diff --git a/libdino/src/service/message_processor.vala b/libdino/src/service/message_processor.vala
index 39f7f373..56ce2bfe 100644
--- a/libdino/src/service/message_processor.vala
+++ b/libdino/src/service/message_processor.vala
@@ -9,12 +9,13 @@ public class MessageProcessor : StreamInteractionModule, Object {
public static ModuleIdentity<MessageProcessor> IDENTITY = new ModuleIdentity<MessageProcessor>("message_processor");
public string id { get { return IDENTITY.id; } }
- public signal void pre_message_received(Entities.Message message, Xmpp.MessageStanza message_stanza, Conversation conversation);
public signal void message_received(Entities.Message message, Conversation conversation);
public signal void build_message_stanza(Entities.Message message, Xmpp.MessageStanza message_stanza, Conversation conversation);
public signal void pre_message_send(Entities.Message message, Xmpp.MessageStanza message_stanza, Conversation conversation);
public signal void message_sent(Entities.Message message, Conversation conversation);
+ public MessageListenerHolder received_pipeline = new MessageListenerHolder();
+
private StreamInteractor stream_interactor;
private Database db;
private Object lock_send_unsent;
@@ -31,6 +32,9 @@ public class MessageProcessor : StreamInteractionModule, Object {
stream_interactor.connection_manager.connection_state_changed.connect((account, state) => {
if (state == ConnectionManager.ConnectionState.CONNECTED) send_unsent_messages(account);
});
+ received_pipeline.connect(new DeduplicateMessageListener(db));
+ received_pipeline.connect(new StoreMessageListener(stream_interactor));
+ received_pipeline.connect(new MamMessageListener(stream_interactor));
}
public void send_message(string text, Conversation conversation) {
@@ -52,7 +56,7 @@ public class MessageProcessor : StreamInteractionModule, Object {
private void on_account_added(Account account) {
stream_interactor.module_manager.get_module(account, Xmpp.MessageModule.IDENTITY).received_message.connect( (stream, message) => {
- on_message_received(account, message);
+ on_message_received.begin(account, message);
});
stream_interactor.module_manager.get_module(account, Xmpp.Xep.MessageArchiveManagement.Module.IDENTITY).feature_available.connect( (stream) => {
DateTime start_time = account.mam_earliest_synced.to_unix() > 60 ? account.mam_earliest_synced.add_minutes(-1) : account.mam_earliest_synced;
@@ -60,16 +64,25 @@ public class MessageProcessor : StreamInteractionModule, Object {
});
}
- private void on_message_received(Account account, Xmpp.MessageStanza message) {
- if (message.body == null) return;
+ private async void on_message_received(Account account, Xmpp.MessageStanza message_stanza) {
+ if (message_stanza.body == null) return;
- Entities.Message new_message = create_in_message(account, message);
- if (new_message == null) return;
+ Entities.Message message = yield create_in_message(account, message_stanza);
+ if (message == null) return;
- determine_message_type(account, message, new_message);
+ Conversation? conversation = stream_interactor.get_module(ConversationManager.IDENTITY).get_conversation_for_message(message);
+ if (conversation != null) {
+ bool abort = yield received_pipeline.run(message, message_stanza, conversation);
+ if (abort) return;
+ }
+ if (message.direction == Entities.Message.DIRECTION_RECEIVED) {
+ message_received(message, conversation);
+ } else if (message.direction == Entities.Message.DIRECTION_SENT) {
+ message_sent(message, conversation);
+ }
}
- private Entities.Message? create_in_message(Account account, Xmpp.MessageStanza message) {
+ private async Entities.Message create_in_message(Account account, Xmpp.MessageStanza message) {
Entities.Message new_message = new Entities.Message(message.body);
new_message.account = account;
new_message.stanza_id = message.id;
@@ -108,54 +121,30 @@ public class MessageProcessor : StreamInteractionModule, Object {
if (delayed_message_flag != null) new_message.time = delayed_message_flag.datetime;
if (new_message.time == null || new_message.time.compare(new_message.local_time) > 0) new_message.time = new_message.local_time;
- return new_message;
- }
+ new_message.type_ = yield determine_message_type(account, message, new_message);
- private void process_message(Entities.Message new_message, Xmpp.MessageStanza stanza) {
- Conversation? conversation = stream_interactor.get_module(ConversationManager.IDENTITY).get_conversation_for_message(new_message);
- if (conversation != null) {
- pre_message_received(new_message, stanza, conversation);
-
- bool is_uuid = new_message.stanza_id != null && Regex.match_simple("""[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}""", new_message.stanza_id);
- if ((is_uuid && !db.contains_message_by_stanza_id(new_message.stanza_id, conversation.account)) ||
- (!is_uuid && !db.contains_message(new_message, conversation.account))) {
- stream_interactor.get_module(MessageStorage.IDENTITY).add_message(new_message, conversation);
-
- if (new_message.direction == Entities.Message.DIRECTION_SENT) {
- message_sent(new_message, conversation);
- } else {
- message_received(new_message, conversation);
- }
-
- bool is_mam_message = Xep.MessageArchiveManagement.MessageFlag.get_flag(stanza) != null;
- XmppStream? stream = stream_interactor.get_stream(conversation.account);
- Xep.MessageArchiveManagement.Flag? mam_flag = stream != null ? stream.get_flag(Xep.MessageArchiveManagement.Flag.IDENTITY) : null;
- if (is_mam_message || (mam_flag != null && mam_flag.cought_up == true)) {
- conversation.account.mam_earliest_synced = new_message.local_time;
- }
- }
- }
+ return new_message;
}
- private void determine_message_type(Account account, Xmpp.MessageStanza message_stanza, Entities.Message message) {
+ private async Entities.Message.Type determine_message_type(Account account, Xmpp.MessageStanza message_stanza, Entities.Message message) {
if (message_stanza.type_ == Xmpp.MessageStanza.TYPE_GROUPCHAT) {
- message.type_ = Entities.Message.Type.GROUPCHAT;
- process_message(message, message_stanza);
- } else if (message_stanza.type_ == Xmpp.MessageStanza.TYPE_CHAT) {
+ return Entities.Message.Type.GROUPCHAT;
+ }
+ if (message_stanza.type_ == Xmpp.MessageStanza.TYPE_CHAT) {
Conversation? conversation = stream_interactor.get_module(ConversationManager.IDENTITY).get_conversation(message.counterpart.bare_jid, account);
if (conversation != null) {
if (conversation.type_ == Conversation.Type.CHAT) {
- message.type_ = Entities.Message.Type.CHAT;
+ return Entities.Message.Type.CHAT;
} else if (conversation.type_ == Conversation.Type.GROUPCHAT) {
- message.type_ = Entities.Message.Type.GROUPCHAT_PM;
+ return Entities.Message.Type.GROUPCHAT_PM;
}
- process_message(message, message_stanza);
} else {
+ SourceFunc callback = determine_message_type.callback;
XmppStream stream = stream_interactor.get_stream(account);
if (stream != null) stream.get_module(Xep.ServiceDiscovery.Module.IDENTITY).get_entity_categories(stream, message.counterpart.bare_jid, (stream, identities) => {
if (identities == null) {
message.type_ = Entities.Message.Type.CHAT;
- process_message(message, message_stanza);
+ Idle.add((owned) callback);
return;
}
foreach (Xep.ServiceDiscovery.Identity identity in identities) {
@@ -165,9 +154,73 @@ public class MessageProcessor : StreamInteractionModule, Object {
message.type_ = Entities.Message.Type.CHAT;
}
}
- process_message(message, message_stanza);
+ Idle.add((owned) callback);
});
+ yield;
+ }
+ }
+ return Entities.Message.Type.CHAT;
+ }
+
+ private class DeduplicateMessageListener : MessageListener {
+
+ public string[] after_actions_const = new string[]{ "" };
+ public override string action_group { get { return "DEDUPLICATE"; } }
+ public override string[] after_actions { get { return after_actions_const; } }
+
+ private Database db;
+
+ public DeduplicateMessageListener(Database db) {
+ this.db = db;
+ }
+
+ public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
+ bool is_uuid = message.stanza_id != null && Regex.match_simple("""[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}""", message.stanza_id);
+ bool new_uuid_msg = is_uuid && !db.contains_message_by_stanza_id(message.stanza_id, conversation.account);
+ bool new_misc_msg = !is_uuid && !db.contains_message(message, conversation.account);
+ bool new_msg = new_uuid_msg || new_misc_msg;
+ return !new_msg;
+ }
+ }
+
+ private class StoreMessageListener : MessageListener {
+
+ public string[] after_actions_const = new string[]{ "DEDUPLICATE" };
+ public override string action_group { get { return "STORE"; } }
+ public override string[] after_actions { get { return after_actions_const; } }
+
+ private StreamInteractor stream_interactor;
+
+ public StoreMessageListener(StreamInteractor stream_interactor) {
+ this.stream_interactor = stream_interactor;
+ }
+
+ public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
+ stream_interactor.get_module(MessageStorage.IDENTITY).add_message(message, conversation);
+ return false;
+ }
+ }
+
+ private class MamMessageListener : MessageListener {
+
+ public string[] after_actions_const = new string[]{ "DEDUPLICATE" };
+ public override string action_group { get { return "MAM_NODE"; } }
+ public override string[] after_actions { get { return after_actions_const; } }
+
+ private StreamInteractor stream_interactor;
+
+ public MamMessageListener(StreamInteractor stream_interactor) {
+ this.stream_interactor = stream_interactor;
+ }
+
+ public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
+ bool is_mam_message = Xep.MessageArchiveManagement.MessageFlag.get_flag(stanza) != null;
+ XmppStream? stream = stream_interactor.get_stream(conversation.account);
+ Xep.MessageArchiveManagement.Flag? mam_flag = stream != null ? stream.get_flag(Xep.MessageArchiveManagement.Flag.IDENTITY) : null;
+ if (is_mam_message || (mam_flag != null && mam_flag.cought_up == true)) {
+ conversation.account.mam_earliest_synced = message.local_time;
}
+ return false;
}
}
@@ -221,4 +274,21 @@ public class MessageProcessor : StreamInteractionModule, Object {
}
}
+public abstract class MessageListener : Xmpp.OrderedListener {
+
+ public abstract async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation);
+}
+
+public class MessageListenerHolder : Xmpp.ListenerHolder {
+
+ public async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
+ foreach (OrderedListener ol in listeners) {
+ MessageListener l = ol as MessageListener;
+ bool stop = yield l.run(message, stanza, conversation);
+ if (stop) return true;
+ }
+ return false;
+ }
+}
+
}
generated by cgit v1.2.3-54-g00ecf (git 2.45.2) at 2024-07-03 17:19:55 +0000