From 57c72d2818dec6c713834cfbb8c4c566a1602907 Mon Sep 17 00:00:00 2001 From: fiaxh Date: Fri, 19 Jan 2018 22:37:02 +0100 Subject: Pipeline for incoming messages in libdino --- libdino/src/service/chat_interaction.vala | 36 +++-- libdino/src/service/conversation_manager.vala | 27 +++- .../service/counterpart_interaction_manager.vala | 20 ++- libdino/src/service/message_processor.vala | 156 +++++++++++++++------ libdino/src/service/muc_manager.vala | 60 +++++--- 5 files changed, 220 insertions(+), 79 deletions(-) (limited to 'libdino') diff --git a/libdino/src/service/chat_interaction.vala b/libdino/src/service/chat_interaction.vala index fe008439..eefde28e 100644 --- a/libdino/src/service/chat_interaction.vala +++ b/libdino/src/service/chat_interaction.vala @@ -27,7 +27,7 @@ public class ChatInteraction : StreamInteractionModule, Object { private ChatInteraction(StreamInteractor stream_interactor) { this.stream_interactor = stream_interactor; Timeout.add_seconds(30, update_interactions); - stream_interactor.get_module(MessageProcessor.IDENTITY).message_received.connect(on_message_received); + stream_interactor.get_module(MessageProcessor.IDENTITY).received_pipeline.connect(new ReceivedMessageListener(stream_interactor)); stream_interactor.get_module(MessageProcessor.IDENTITY).message_sent.connect(on_message_sent); } @@ -124,19 +124,35 @@ public class ChatInteraction : StreamInteractionModule, Object { return true; } - private void on_message_received(Entities.Message message, Conversation conversation) { - if (Xep.MessageArchiveManagement.MessageFlag.get_flag(message.stanza) != null) return; + private class ReceivedMessageListener : MessageListener { - send_delivery_receipt(conversation, message); - if (is_active_focus(conversation)) { - check_send_read(); - conversation.read_up_to = message; - send_chat_marker(conversation, message, Xep.ChatMarkers.MARKER_DISPLAYED); - } else { - send_chat_marker(conversation, message, Xep.ChatMarkers.MARKER_RECEIVED); + public string[] after_actions_const = new string[]{ "" }; + public override string action_group { get { return "OTHER_NODES"; } } + public override string[] after_actions { get { return after_actions_const; } } + + private StreamInteractor stream_interactor; + + public ReceivedMessageListener(StreamInteractor stream_interactor) { + this.stream_interactor = stream_interactor; + } + + public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) { + if (Xep.MessageArchiveManagement.MessageFlag.get_flag(message.stanza) != null) return false; + + ChatInteraction outer = stream_interactor.get_module(ChatInteraction.IDENTITY); + outer.send_delivery_receipt(conversation, message); + if (outer.is_active_focus(conversation)) { + outer.check_send_read(); + conversation.read_up_to = message; + outer.send_chat_marker(conversation, message, Xep.ChatMarkers.MARKER_DISPLAYED); + } else { + outer.send_chat_marker(conversation, message, Xep.ChatMarkers.MARKER_RECEIVED); + } + return false; } } + private void send_chat_marker(Conversation conversation, Entities.Message message, string marker) { XmppStream stream = stream_interactor.get_stream(conversation.account); if (stream != null && diff --git a/libdino/src/service/conversation_manager.vala b/libdino/src/service/conversation_manager.vala index e66ecca0..f6e3553d 100644 --- a/libdino/src/service/conversation_manager.vala +++ b/libdino/src/service/conversation_manager.vala @@ -27,7 +27,7 @@ public class ConversationManager : StreamInteractionModule, Object { stream_interactor.add_module(this); stream_interactor.account_added.connect(on_account_added); stream_interactor.get_module(MucManager.IDENTITY).joined.connect(on_groupchat_joined); - stream_interactor.get_module(MessageProcessor.IDENTITY).message_received.connect(handle_new_message); + stream_interactor.get_module(MessageProcessor.IDENTITY).received_pipeline.connect(new MessageListener(stream_interactor)); stream_interactor.get_module(MessageProcessor.IDENTITY).message_sent.connect(handle_new_message); } @@ -119,6 +119,31 @@ public class ConversationManager : StreamInteractionModule, Object { } } + private class MessageListener : Dino.MessageListener { + + public string[] after_actions_const = new string[]{ "DEDUPLICATE" }; + public override string action_group { get { return "MANAGER"; } } + public override string[] after_actions { get { return after_actions_const; } } + + private StreamInteractor stream_interactor; + + public MessageListener(StreamInteractor stream_interactor) { + this.stream_interactor = stream_interactor; + } + + public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) { + conversation.last_active = message.time; + + if (message.stanza != null) { + bool is_mam_message = Xep.MessageArchiveManagement.MessageFlag.get_flag(message.stanza) != null; + bool is_recent = message.local_time.compare(new DateTime.now_utc().add_hours(-24)) > 0; + if (is_mam_message && !is_recent) return false; + } + stream_interactor.get_module(ConversationManager.IDENTITY).start_conversation(conversation); + return false; + } + } + private void handle_new_message(Entities.Message message, Conversation conversation) { conversation.last_active = message.time; diff --git a/libdino/src/service/counterpart_interaction_manager.vala b/libdino/src/service/counterpart_interaction_manager.vala index eeb7f773..26be0096 100644 --- a/libdino/src/service/counterpart_interaction_manager.vala +++ b/libdino/src/service/counterpart_interaction_manager.vala @@ -25,7 +25,7 @@ public class CounterpartInteractionManager : StreamInteractionModule, Object { private CounterpartInteractionManager(StreamInteractor stream_interactor) { this.stream_interactor = stream_interactor; stream_interactor.account_added.connect(on_account_added); - stream_interactor.get_module(MessageProcessor.IDENTITY).message_received.connect(on_message_received); + stream_interactor.get_module(MessageProcessor.IDENTITY).received_pipeline.connect(new ReceivedMessageListener(this)); stream_interactor.get_module(MessageProcessor.IDENTITY).message_sent.connect(check_if_got_marker); stream_interactor.stream_negotiated.connect(() => chat_states.clear() ); } @@ -88,8 +88,22 @@ public class CounterpartInteractionManager : StreamInteractionModule, Object { } } - private void on_message_received(Entities.Message message, Conversation conversation) { - on_chat_state_received(conversation.account, conversation.counterpart, Xep.ChatStateNotifications.STATE_ACTIVE); + private class ReceivedMessageListener : 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 CounterpartInteractionManager outer; + + public ReceivedMessageListener(CounterpartInteractionManager outer) { + this.outer = outer; + } + + public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) { + outer.on_chat_state_received(conversation.account, conversation.counterpart, Xep.ChatStateNotifications.STATE_ACTIVE); + return false; + } } private void on_receipt_received(Account account, Jid jid, string id) { 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 IDENTITY = new ModuleIdentity("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; + } +} + } diff --git a/libdino/src/service/muc_manager.vala b/libdino/src/service/muc_manager.vala index 3b452f20..f224a8ec 100644 --- a/libdino/src/service/muc_manager.vala +++ b/libdino/src/service/muc_manager.vala @@ -16,6 +16,7 @@ public class MucManager : StreamInteractionModule, Object { private StreamInteractor stream_interactor; private HashMap enter_errors = new HashMap(Jid.hash_func, Jid.equals_func); + private ReceivedMessageListener received_message_listener; public static void start(StreamInteractor stream_interactor) { MucManager m = new MucManager(stream_interactor); @@ -24,9 +25,10 @@ public class MucManager : StreamInteractionModule, Object { private MucManager(StreamInteractor stream_interactor) { this.stream_interactor = stream_interactor; + this.received_message_listener = new ReceivedMessageListener(stream_interactor); stream_interactor.account_added.connect(on_account_added); stream_interactor.stream_negotiated.connect(on_stream_negotiated); - stream_interactor.get_module(MessageProcessor.IDENTITY).pre_message_received.connect(on_pre_message_received); + stream_interactor.get_module(MessageProcessor.IDENTITY).received_pipeline.connect(received_message_listener); } public void join(Account account, Jid jid, string? nick, string? password) { @@ -241,27 +243,6 @@ public class MucManager : StreamInteractionModule, Object { }); } - private void on_pre_message_received(Entities.Message message, Xmpp.MessageStanza message_stanza, Conversation conversation) { - if (conversation.type_ != Conversation.Type.GROUPCHAT) return; - XmppStream stream = stream_interactor.get_stream(conversation.account); - if (stream == null) return; - if (Xep.DelayedDelivery.MessageFlag.get_flag(message.stanza) == null) { - Jid? real_jid = stream.get_flag(Xep.Muc.Flag.IDENTITY).get_real_jid(message.counterpart); - if (real_jid != null && !real_jid.equals(message.counterpart)) { - message.real_jid = real_jid.bare_jid; - } - } - string? muc_nick = stream.get_flag(Xep.Muc.Flag.IDENTITY).get_muc_nick(conversation.counterpart.bare_jid); - if (muc_nick != null && message.from.equals(message.from.with_resource(muc_nick))) { // TODO better from own - Gee.List messages = stream_interactor.get_module(MessageStorage.IDENTITY).get_messages(conversation); - foreach (Entities.Message m in messages) { // TODO not here - if (m.equals(message)) { - m.marked = Entities.Message.Marked.RECEIVED; - } - } - } - } - private void join_all_active(Account account) { Gee.List conversations = stream_interactor.get_module(ConversationManager.IDENTITY).get_active_conversations(account); foreach (Conversation conversation in conversations) { @@ -336,6 +317,41 @@ public class MucManager : StreamInteractionModule, Object { } }); } + + private class ReceivedMessageListener : MessageListener { + + public string[] after_actions_const = new string[]{ "" }; + public override string action_group { get { return "OTHER_NODES"; } } + public override string[] after_actions { get { return after_actions_const; } } + + private StreamInteractor stream_interactor; + + public ReceivedMessageListener(StreamInteractor stream_interactor) { + this.stream_interactor = stream_interactor; + } + + public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) { + if (conversation.type_ != Conversation.Type.GROUPCHAT) return false; + XmppStream stream = stream_interactor.get_stream(conversation.account); + if (stream == null) return false; + if (Xep.DelayedDelivery.MessageFlag.get_flag(message.stanza) == null) { + Jid? real_jid = stream.get_flag(Xep.Muc.Flag.IDENTITY).get_real_jid(message.counterpart); + if (real_jid != null && !real_jid.equals(message.counterpart)) { + message.real_jid = real_jid.bare_jid; + } + } + string? muc_nick = stream.get_flag(Xep.Muc.Flag.IDENTITY).get_muc_nick(conversation.counterpart.bare_jid); + if (muc_nick != null && message.from.equals(new Jid(@"$(message.from.bare_jid)/$muc_nick"))) { // TODO better from own + Gee.List messages = stream_interactor.get_module(MessageStorage.IDENTITY).get_messages(conversation); + foreach (Entities.Message m in messages) { // TODO not here + if (m.equals(message)) { + m.marked = Entities.Message.Marked.RECEIVED; + } + } + } + return false; + } + } } } -- cgit v1.2.3-54-g00ecf