Pipeline for incoming messages in libdino

5 files changed, 220 insertions, 79 deletions

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<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;
+    }
+}
+
 }
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<Jid, Xep.Muc.MucEnterError> enter_errors = new HashMap<Jid, Xep.Muc.MucEnterError>(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<Entities.Message> 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<Conversation> 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<Entities.Message> 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;
+        }
+    }
 }
 
 }