MessageStorage/Processor, correctly resolve conversations (fixup 22adbd3)

libdino/CMakeLists.txt

@@ -32,12 +32,14 @@ SOURCES
     src/service/counterpart_interaction_manager.vala
     src/service/database.vala
     src/service/entity_capabilities_storage.vala
-    src/service/message_manager.vala
+    src/service/message_processor.vala
+    src/service/message_storage.vala
     src/service/module_manager.vala
     src/service/muc_manager.vala
     src/service/presence_manager.vala
     src/service/roster_manager.vala
     src/service/stream_interactor.vala
+    src/service/util.vala
 
     src/settings.vala
 CUSTOM_VAPIS

libdino/src/application.vala

@@ -24,7 +24,8 @@ public class Dino.Application : Gtk.Application {
         this.stream_interaction = new StreamInteractor(db);
 
         AvatarManager.start(stream_interaction, db);
-        MessageManager.start(stream_interaction, db);
+        MessageProcessor.start(stream_interaction, db);
+        MessageStorage.start(stream_interaction, db);
         CounterpartInteractionManager.start(stream_interaction);
         PresenceManager.start(stream_interaction);
         MucManager.start(stream_interaction);

libdino/src/entity/conversation.vala

@@ -90,6 +90,11 @@ public class Conversation : Object {
     private void on_update(Object o, ParamSpec sp) {
         var update = db.conversation.update().with(db.conversation.jid_id, "=", db.get_jid_id(counterpart))
                 .with(db.conversation.account_id, "=", account.id);
+        if (counterpart.resourcepart != null) {
+            update.with(db.conversation.resource, "=", counterpart.resourcepart);
+        } else {
+            update.with_null(db.conversation.resource);
+        }
         switch (sp.name) {
             case "type-":
                 update.set(db.conversation.type_, type_); break;

libdino/src/service/chat_interaction.vala

@@ -27,8 +27,8 @@ 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(MessageManager.IDENTITY).message_received.connect(on_message_received);
+        stream_interactor.get_module(MessageProcessor.IDENTITY).message_received.connect(on_message_received);
-        stream_interactor.get_module(MessageManager.IDENTITY).message_sent.connect(on_message_sent);
+        stream_interactor.get_module(MessageProcessor.IDENTITY).message_sent.connect(on_message_sent);
     }
 
     public bool is_active_focus(Conversation? conversation = null) {
@@ -79,7 +79,7 @@ public class ChatInteraction : StreamInteractionModule, Object {
         if (conversation == null) return;
         conversation_read(selected_conversation);
         check_send_read();
-        selected_conversation.read_up_to = stream_interactor.get_module(MessageManager.IDENTITY).get_last_message(conversation);
+        selected_conversation.read_up_to = stream_interactor.get_module(MessageStorage.IDENTITY).get_last_message(conversation);
     }
 
     private void on_conversation_unfocused(Conversation? conversation) {
@@ -93,7 +93,7 @@ public class ChatInteraction : StreamInteractionModule, Object {
 
     private void check_send_read() {
         if (selected_conversation == null || selected_conversation.type_ == Conversation.Type.GROUPCHAT) return;
-        Entities.Message? message = stream_interactor.get_module(MessageManager.IDENTITY).get_last_message(selected_conversation);
+        Entities.Message? message = stream_interactor.get_module(MessageStorage.IDENTITY).get_last_message(selected_conversation);
         if (message != null && message.direction == Entities.Message.DIRECTION_RECEIVED &&
                 message.stanza != null && !message.equals(selected_conversation.read_up_to)) {
             selected_conversation.read_up_to = message;

libdino/src/service/conversation_manager.vala

@@ -27,8 +27,8 @@ public class ConversationManager : StreamInteractionModule, Object {
         stream_interactor.add_module(this);
         stream_interactor.account_added.connect(on_account_added);
         stream_interactor.get_module(MucManager.IDENTITY).groupchat_joined.connect(on_groupchat_joined);
-        stream_interactor.get_module(MessageManager.IDENTITY).pre_message_received.connect(on_message_received);
+        stream_interactor.get_module(MessageProcessor.IDENTITY).pre_message_received.connect(on_message_received);
-        stream_interactor.get_module(MessageManager.IDENTITY).message_sent.connect(on_message_sent);
+        stream_interactor.get_module(MessageProcessor.IDENTITY).message_sent.connect(on_message_sent);
     }
 
     public Conversation create_conversation(Jid jid, Account account, Conversation.Type? type = null) {
@@ -55,10 +55,14 @@ public class ConversationManager : StreamInteractionModule, Object {
     }
 
     public Gee.List<Conversation> get_conversations_for_presence(Show show, Account account) {
+        return get_conversations(show.jid, account);
+    }
+
+    public Gee.List<Conversation> get_conversations(Jid jid, Account account) {
         Gee.List<Conversation> ret = new ArrayList<Conversation>(Conversation.equals_func);
-        Conversation? bare_conversation = get_conversation(show.jid, account);
+        Conversation? bare_conversation = get_conversation(jid, account);
         if (bare_conversation != null) ret.add(bare_conversation);
-        Conversation? full_conversation = get_conversation(show.jid.bare_jid, account);
+        Conversation? full_conversation = get_conversation(jid.bare_jid, account);
         if (full_conversation != null) ret.add(full_conversation);
         return ret;
     }

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(MessageManager.IDENTITY).message_received.connect(on_message_received);
+        stream_interactor.get_module(MessageProcessor.IDENTITY).message_received.connect(on_message_received);
     }
 
     public string? get_chat_state(Account account, Jid jid) {
@@ -54,28 +54,24 @@ public class CounterpartInteractionManager : StreamInteractionModule, Object {
     }
 
     private void on_chat_marker_received(Account account, Jid jid, string marker, string stanza_id) {
-        Conversation? conversation = stream_interactor.get_module(ConversationManager.IDENTITY).get_conversation(jid, account);
+        foreach (Conversation conversation in stream_interactor.get_module(ConversationManager.IDENTITY).get_conversations(jid, account)) {
-        if (conversation != null) {
+            Entities.Message? message = stream_interactor.get_module(MessageStorage.IDENTITY).get_message_by_id(stanza_id, conversation);
-            Gee.List<Entities.Message>? messages = stream_interactor.get_module(MessageManager.IDENTITY).get_messages(conversation);
+            if (message != null) {
-            if (messages != null) { // TODO not here
+                switch (marker) {
-                foreach (Entities.Message message in messages) {
+                    case Xep.ChatMarkers.MARKER_RECEIVED:
-                    if (message.stanza_id == stanza_id) {
+                        received_message_received(account, jid, message);
-                        switch (marker) {
+                        message.marked = Entities.Message.Marked.RECEIVED;
-                            case Xep.ChatMarkers.MARKER_RECEIVED:
+                        break;
-                                received_message_received(account, jid, message);
+                    case Xep.ChatMarkers.MARKER_DISPLAYED:
-                                message.marked = Entities.Message.Marked.RECEIVED;
+                        last_read[jid] = message;
-                                break;
+                        received_message_displayed(account, jid, message);
-                            case Xep.ChatMarkers.MARKER_DISPLAYED:
+                        Gee.List<Entities.Message> messages = stream_interactor.get_module(MessageStorage.IDENTITY).get_messages(conversation);
-                                last_read[jid] = message;
+                        foreach (Entities.Message m in messages) {
-                                received_message_displayed(account, jid, message);
+                            if (m.equals(message)) break;
-                                foreach (Entities.Message m in messages) {
+                            if (m.marked == Entities.Message.Marked.RECEIVED) m.marked = Entities.Message.Marked.READ;
-                                    if (m.equals(message)) break;
-                                    if (m.marked == Entities.Message.Marked.RECEIVED) m.marked = Entities.Message.Marked.READ;
-                                }
-                                message.marked = Entities.Message.Marked.READ;
-                                break;
                         }
-                    }
+                        message.marked = Entities.Message.Marked.READ;
+                        break;
                 }
             }
         }

libdino/src/service/database.vala

@@ -190,7 +190,10 @@ public class Database : Qlite.Database {
 
     public Gee.List<Message> get_unsend_messages(Account account) {
         Gee.List<Message> ret = new ArrayList<Message>();
-        foreach (Row row in message.select().with(message.marked, "=", (int) Message.Marked.UNSENT)) {
+        var select = message.select()
+            .with(message.account_id, "=", account.id)
+            .with(message.marked, "=", (int) Message.Marked.UNSENT);
+        foreach (Row row in select) {
             ret.add(new Message.from_row(this, row));
         }
         return ret;

libdino/src/service/message_processor.vala

@@ -5,8 +5,8 @@ using Dino.Entities;
 
 namespace Dino {
 
-public class MessageManager : StreamInteractionModule, Object {
+public class MessageProcessor : StreamInteractionModule, Object {
-    public static ModuleIdentity<MessageManager> IDENTITY = new ModuleIdentity<MessageManager>("message_manager");
+    public static ModuleIdentity<MessageProcessor> IDENTITY = new ModuleIdentity<MessageProcessor>("message_manager");
     public string id { get { return IDENTITY.id; } }
 
     public signal void pre_message_received(Entities.Message message, Xmpp.Message.Stanza message_stanza, Conversation conversation);
@@ -17,14 +17,14 @@ public class MessageManager : StreamInteractionModule, Object {
 
     private StreamInteractor stream_interactor;
     private Database db;
-    private HashMap<Conversation, ArrayList<Entities.Message>> messages = new HashMap<Conversation, ArrayList<Entities.Message>>(Conversation.hash_func, Conversation.equals_func);
+    private Object lock_send_unsent;
 
     public static void start(StreamInteractor stream_interactor, Database db) {
-        MessageManager m = new MessageManager(stream_interactor, db);
+        MessageProcessor m = new MessageProcessor(stream_interactor, db);
         stream_interactor.add_module(m);
     }
 
-    private MessageManager(StreamInteractor stream_interactor, Database db) {
+    private MessageProcessor(StreamInteractor stream_interactor, Database db) {
         this.stream_interactor = stream_interactor;
         this.db = db;
         stream_interactor.account_added.connect(on_account_added);
@@ -35,52 +35,12 @@ public class MessageManager : StreamInteractionModule, Object {
 
     public void send_message(string text, Conversation conversation) {
         Entities.Message message = create_out_message(text, conversation);
-        add_message(message, conversation);
+        stream_interactor.get_module(MessageStorage.IDENTITY).add_message(message, conversation);
         message.persist(db);
         send_xmpp_message(message, conversation);
         message_sent(message, conversation);
     }
 
-    public Gee.List<Entities.Message>? get_messages(Conversation conversation, int count = 50) {
-        if (messages.has_key(conversation) && messages[conversation].size > 0) {
-            Gee.List<Entities.Message> db_messages = db.get_messages(conversation.counterpart, conversation.account, get_message_type_for_conversation(conversation), count, messages[conversation][0]);
-            db_messages.add_all(messages[conversation]);
-            return db_messages;
-        } else {
-            Gee.List<Entities.Message> db_messages = db.get_messages(conversation.counterpart, conversation.account, get_message_type_for_conversation(conversation), count, null);
-            return db_messages;
-        }
-    }
-
-    public Entities.Message? get_last_message(Conversation conversation) {
-        if (messages.has_key(conversation) && messages[conversation].size > 0) {
-            return messages[conversation][messages[conversation].size - 1];
-        } else {
-            Gee.List<Entities.Message> db_messages = db.get_messages(conversation.counterpart, conversation.account, get_message_type_for_conversation(conversation), 1, null);
-            if (db_messages.size >= 1) {
-                return db_messages[0];
-            }
-        }
-        return null;
-    }
-
-    public Gee.List<Entities.Message>? get_messages_before(Conversation? conversation, Entities.Message before) {
-        Gee.List<Entities.Message> db_messages = db.get_messages(conversation.counterpart, conversation.account, get_message_type_for_conversation(conversation), 20, before);
-        return db_messages;
-    }
-
-    private Entities.Message.Type get_message_type_for_conversation(Conversation conversation) {
-        switch (conversation.type_) {
-            case Conversation.Type.CHAT:
-                return Entities.Message.Type.CHAT;
-            case Conversation.Type.GROUPCHAT:
-                return Entities.Message.Type.GROUPCHAT;
-            case Conversation.Type.GROUPCHAT_PM:
-                return Entities.Message.Type.GROUPCHAT_PM;
-        }
-        assert_not_reached();
-    }
-
     private void on_account_added(Account account) {
         stream_interactor.module_manager.get_module(account, Xmpp.Message.Module.IDENTITY).received_message.connect( (stream, message) => {
             on_message_received(account, message);
@@ -135,8 +95,7 @@ public class MessageManager : StreamInteractionModule, Object {
             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))) {
-                new_message.persist(db);
+                stream_interactor.get_module(MessageStorage.IDENTITY).add_message(new_message, conversation);
-                add_message(new_message, conversation);
                 if (new_message.direction == Entities.Message.DIRECTION_SENT) {
                     message_sent(new_message, conversation);
                 } else {
@@ -161,7 +120,7 @@ public class MessageManager : StreamInteractionModule, Object {
             } else {
                 Core.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.to_string(), (stream, identities, store) => {
-                    Triple<MessageManager, Entities.Message, Xmpp.Message.Stanza> triple = store as Triple<MessageManager, Entities.Message, Xmpp.Message.Stanza>;
+                    Triple<MessageProcessor, Entities.Message, Xmpp.Message.Stanza> triple = store as Triple<MessageProcessor, Entities.Message, Xmpp.Message.Stanza>;
                     Entities.Message m = triple.b;
                     if (identities == null) {
                         m.type_ = Entities.Message.Type.CHAT;
@@ -181,16 +140,9 @@ public class MessageManager : StreamInteractionModule, Object {
         }
     }
 
-    private void add_message(Entities.Message message, Conversation conversation) {
-        if (!messages.has_key(conversation)) {
-            messages[conversation] = new ArrayList<Entities.Message>(Entities.Message.equals_func);
-        }
-        messages[conversation].add(message);
-    }
-
     private Entities.Message create_out_message(string text, Conversation conversation) {
         Entities.Message message = new Entities.Message(text);
-        message.type_ = get_message_type_for_conversation(conversation);
+        message.type_ = Util.get_message_type_for_conversation(conversation);
         message.stanza_id = random_uuid();
         message.account = conversation.account;
         message.body = text;
@@ -207,7 +159,7 @@ public class MessageManager : StreamInteractionModule, Object {
     }
 
     public void send_xmpp_message(Entities.Message message, Conversation conversation, bool delayed = false) {
-        lock (messages) {
+        lock (lock_send_unsent) {
             Core.XmppStream stream = stream_interactor.get_stream(conversation.account);
             message.marked = Entities.Message.Marked.NONE;
             if (stream != null) {

libdino/src/service/message_storage.vala

@@ -0,0 +1,68 @@
+using Gee;
+
+using Dino.Entities;
+
+namespace Dino {
+
+public class MessageStorage : StreamInteractionModule, Object {
+    public static ModuleIdentity<MessageStorage> IDENTITY = new ModuleIdentity<MessageStorage>("message_cache");
+    public string id { get { return IDENTITY.id; } }
+
+    private StreamInteractor stream_interactor;
+    private Database db;
+
+    private HashMap<Conversation, Gee.List<Message>> messages = new HashMap<Conversation, Gee.List<Entities.Message>>(Conversation.hash_func, Conversation.equals_func);
+
+    public static void start(StreamInteractor stream_interactor, Database db) {
+        MessageStorage m = new MessageStorage(stream_interactor, db);
+        stream_interactor.add_module(m);
+    }
+
+    private MessageStorage(StreamInteractor stream_interactor, Database db) {
+        this.stream_interactor = stream_interactor;
+        this.db = db;
+    }
+
+    public void add_message(Message message, Conversation conversation) {
+        message.persist(db);
+        init_conversation(conversation);
+        messages[conversation].add(message);
+    }
+
+    public Gee.List<Message> get_messages(Conversation conversation, int count = 50) {
+        init_conversation(conversation);
+        if (messages[conversation].size > 0) {
+            return messages[conversation][int.max(messages[conversation].size - count - 1, 0) : messages[conversation].size];
+        }
+        return new ArrayList<Conversation>();
+    }
+
+    public Message? get_last_message(Conversation conversation) {
+        init_conversation(conversation);
+        if (messages[conversation].size > 0) {
+            return messages[conversation][messages[conversation].size - 1];
+        }
+        return null;
+    }
+
+    public Gee.List<Message>? get_messages_before(Conversation? conversation, Message before, int count = 20) {
+        Gee.List<Message> db_messages = db.get_messages(conversation.counterpart, conversation.account, Util.get_message_type_for_conversation(conversation), count, before);
+        return db_messages;
+    }
+
+    public Message? get_message_by_id(string stanza_id, Conversation conversation) {
+        init_conversation(conversation);
+        for (int i = messages[conversation].size - 1; i > 0; i--) {
+            if (messages[conversation][i].stanza_id == stanza_id) return messages[conversation][i];
+        }
+        return null;
+    }
+
+    private void init_conversation(Conversation conversation) {
+        if (!messages.has_key(conversation)) {
+            messages[conversation] = db.get_messages(conversation.counterpart, conversation.account, Util.get_message_type_for_conversation(conversation), 50, null);
+        }
+    }
+}
+
+}

libdino/src/service/muc_manager.vala

@@ -24,7 +24,7 @@ public class MucManager : StreamInteractionModule, Object {
         this.stream_interactor = stream_interactor;
         stream_interactor.account_added.connect(on_account_added);
         stream_interactor.stream_negotiated.connect(on_stream_negotiated);
-        stream_interactor.get_module(MessageManager.IDENTITY).pre_message_received.connect(on_pre_message_received);
+        stream_interactor.get_module(MessageProcessor.IDENTITY).pre_message_received.connect(on_pre_message_received);
     }
 
     public void join(Account account, Jid jid, string nick, string? password = null) {
@@ -170,12 +170,10 @@ public class MucManager : StreamInteractionModule, Object {
         }
         string? muc_nick = stream.get_flag(Xep.Muc.Flag.IDENTITY).get_muc_nick(conversation.counterpart.bare_jid.to_string());
         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(MessageManager.IDENTITY).get_messages(conversation);
+            Gee.List<Entities.Message> messages = stream_interactor.get_module(MessageStorage.IDENTITY).get_messages(conversation);
-            if (messages != null) { // TODO not here
+            foreach (Entities.Message m in messages) { // TODO not here
-                foreach (Entities.Message m in messages) {
+                if (m.equals(message)) {
-                    if (m.equals(message)) {
+                    m.marked = Entities.Message.Marked.RECEIVED;
-                        m.marked = Entities.Message.Marked.RECEIVED;
-                    }
                 }
             }
         }

libdino/src/service/util.vala

@@ -0,0 +1,19 @@
+using Dino.Entities;
+
+namespace Dino {
+
+public class Util {
+    public static Entities.Message.Type get_message_type_for_conversation(Conversation conversation) {
+        switch (conversation.type_) {
+            case Conversation.Type.CHAT:
+                return Entities.Message.Type.CHAT;
+            case Conversation.Type.GROUPCHAT:
+                return Entities.Message.Type.GROUPCHAT;
+            case Conversation.Type.GROUPCHAT_PM:
+                return Entities.Message.Type.GROUPCHAT_PM;
+        }
+        assert_not_reached();
+    }
+}
+
+}

main/src/ui/chat_input/occupants_tab_completer.vala

@@ -87,14 +87,12 @@ class OccupantsTabCompletor {
 
     private Gee.List<string> generate_completions_from_messages() {
         Gee.List<string> ret = new ArrayList<string>();
-        Gee.List<Message>? messages = stream_interactor.get_module(MessageManager.IDENTITY).get_messages(conversation, 10);
+        Gee.List<Message> messages = stream_interactor.get_module(MessageStorage.IDENTITY).get_messages(conversation, 10);
-        if (messages != null) {
+        for (int i = messages.size - 1; i > 0; i--) {
-            for (int i = messages.size - 1; i > 0; i--) {
+            string resourcepart = messages[i].from.resourcepart;
-                string resourcepart = messages[i].from.resourcepart;
+            string own_nick = stream_interactor.get_module(MucManager.IDENTITY).get_nick(conversation.counterpart, conversation.account);
-                string own_nick = stream_interactor.get_module(MucManager.IDENTITY).get_nick(conversation.counterpart, conversation.account);
+            if (resourcepart != null && resourcepart != "" && resourcepart != own_nick && !ret.contains(resourcepart)) {
-                if (resourcepart != null && resourcepart != "" && resourcepart != own_nick && !ret.contains(resourcepart)) {
+                ret.add(resourcepart);
-                    ret.add(resourcepart);
-                }
             }
         }
         return ret;

main/src/ui/chat_input/view.vala

@@ -56,7 +56,7 @@ public class View : Box {
                     stream_interactor.get_module(MucManager.IDENTITY).kick(conversation.account, conversation.counterpart, token[1]);
                     break;
                 case "/me":
-                    stream_interactor.get_module(MessageManager.IDENTITY).send_message(text, conversation);
+                    stream_interactor.get_module(MessageProcessor.IDENTITY).send_message(text, conversation);
                     break;
                 case "/nick":
                     stream_interactor.get_module(MucManager.IDENTITY).change_nick(conversation.account, conversation.counterpart, token[1]);
@@ -66,7 +66,7 @@ public class View : Box {
                     break;
             }
         } else {
-            stream_interactor.get_module(MessageManager.IDENTITY).send_message(text, conversation);
+            stream_interactor.get_module(MessageProcessor.IDENTITY).send_message(text, conversation);
         }
         text_input.buffer.text = "";
     }

main/src/ui/conversation_selector/conversation_row.vala

@@ -46,7 +46,7 @@ public abstract class ConversationRow : ListBoxRow {
         x_button.clicked.connect(on_x_button_clicked);
 
         update_name(Util.get_conversation_display_name(stream_interactor, conversation));
-        Entities.Message message = stream_interactor.get_module(MessageManager.IDENTITY).get_last_message(conversation);
+        Entities.Message message = stream_interactor.get_module(MessageStorage.IDENTITY).get_last_message(conversation);
         if (message != null) {
             message_received(message);
         }

main/src/ui/conversation_selector/list.vala

@@ -31,10 +31,10 @@ public class List : ListBox {
         stream_interactor.get_module(ConversationManager.IDENTITY).conversation_activated.connect((conversation) => {
             Idle.add(() => {add_conversation(conversation); return false;});
         });
-        stream_interactor.get_module(MessageManager.IDENTITY).message_received.connect((message, conversation) => {
+        stream_interactor.get_module(MessageProcessor.IDENTITY).message_received.connect((message, conversation) => {
             Idle.add(() => {on_message_received(message, conversation); return false;});
         });
-        stream_interactor.get_module(MessageManager.IDENTITY).message_sent.connect((message, conversation) => {
+        stream_interactor.get_module(MessageProcessor.IDENTITY).message_sent.connect((message, conversation) => {
             Idle.add(() => {on_message_received(message, conversation); return false;});
         });
         stream_interactor.get_module(PresenceManager.IDENTITY).show_received.connect((show, jid, account) => {

main/src/ui/conversation_summary/view.vala

@@ -35,10 +35,10 @@ public class View : Box {
         stream_interactor.get_module(CounterpartInteractionManager.IDENTITY).received_state.connect((account, jid, state) => {
             Idle.add(() => { on_received_state(account, jid, state); return false; });
         });
-        stream_interactor.get_module(MessageManager.IDENTITY).message_received.connect((message, conversation) => {
+        stream_interactor.get_module(MessageProcessor.IDENTITY).message_received.connect((message, conversation) => {
             Idle.add(() => { show_message(message, conversation, true); return false; });
         });
-        stream_interactor.get_module(MessageManager.IDENTITY).message_sent.connect((message, conversation) => {
+        stream_interactor.get_module(MessageProcessor.IDENTITY).message_sent.connect((message, conversation) => {
             Idle.add(() => { show_message(message, conversation, true); return false; });
         });
         stream_interactor.get_module(PresenceManager.IDENTITY).show_received.connect((show, jid, account) => {
@@ -65,8 +65,8 @@ public class View : Box {
         last_conversation_item = null;
 
         ArrayList<Object> objects = new ArrayList<Object>();
-        Gee.List<Entities.Message>? messages = stream_interactor.get_module(MessageManager.IDENTITY).get_messages(conversation);
+        Gee.List<Entities.Message> messages = stream_interactor.get_module(MessageStorage.IDENTITY).get_messages(conversation);
-        if (messages != null && messages.size > 0) {
+        if (messages.size > 0) {
             earliest_message = messages[0];
             objects.add_all(messages);
         }
@@ -158,7 +158,7 @@ public class View : Box {
             if(reloading) return;
             reloading = true;
         }
-        Gee.List<Entities.Message>? messages = stream_interactor.get_module(MessageManager.IDENTITY).get_messages_before(conversation, earliest_message);
+        Gee.List<Entities.Message>? messages = stream_interactor.get_module(MessageStorage.IDENTITY).get_messages_before(conversation, earliest_message);
         if (messages != null && messages.size > 0) {
             earliest_message = messages[0];
             MergedMessageItem? current_item = null;

main/src/ui/notifications.vala

@@ -24,7 +24,7 @@ public class Notifications : Object {
     }
 
     public void start() {
-        stream_interactor.get_module(MessageManager.IDENTITY).message_received.connect(on_message_received);
+        stream_interactor.get_module(MessageProcessor.IDENTITY).message_received.connect(on_message_received);
         stream_interactor.get_module(PresenceManager.IDENTITY).received_subscription_request.connect(on_received_subscription_request);
     }
 

plugins/omemo/src/manager.vala

@@ -65,8 +65,8 @@ public class Manager : StreamInteractionModule, Object {
 
         stream_interactor.stream_negotiated.connect(on_stream_negotiated);
         stream_interactor.account_added.connect(on_account_added);
-        stream_interactor.get_module(MessageManager.IDENTITY).pre_message_received.connect(on_pre_message_received);
+        stream_interactor.get_module(MessageProcessor.IDENTITY).pre_message_received.connect(on_pre_message_received);
-        stream_interactor.get_module(MessageManager.IDENTITY).pre_message_send.connect(on_pre_message_send);
+        stream_interactor.get_module(MessageProcessor.IDENTITY).pre_message_send.connect(on_pre_message_send);
     }
 
     private void on_pre_message_received(Entities.Message message, Xmpp.Message.Stanza message_stanza, Conversation conversation) {
@@ -145,7 +145,7 @@ public class Manager : StreamInteractionModule, Object {
         }
         foreach (Entities.Message msg in send_now) {
             Entities.Conversation conv = stream_interactor.get_module(ConversationManager.IDENTITY).get_conversation(msg.counterpart, account);
-            stream_interactor.get_module(MessageManager.IDENTITY).send_xmpp_message(msg, conv, true);
+            stream_interactor.get_module(MessageProcessor.IDENTITY).send_xmpp_message(msg, conv, true);
         }
     }
 
@@ -169,7 +169,7 @@ public class Manager : StreamInteractionModule, Object {
         }
         foreach (Entities.Message msg in send_now) {
             Entities.Conversation conv = stream_interactor.get_module(ConversationManager.IDENTITY).get_conversation(msg.counterpart, account);
-            stream_interactor.get_module(MessageManager.IDENTITY).send_xmpp_message(msg, conv, true);
+            stream_interactor.get_module(MessageProcessor.IDENTITY).send_xmpp_message(msg, conv, true);
         }
     }
 

plugins/openpgp/src/manager.vala

@@ -26,8 +26,8 @@ namespace Dino.Plugins.OpenPgp {
             this.db = db;
 
             stream_interactor.account_added.connect(on_account_added);
-            stream_interactor.get_module(MessageManager.IDENTITY).pre_message_received.connect(on_pre_message_received);
+            stream_interactor.get_module(MessageProcessor.IDENTITY).pre_message_received.connect(on_pre_message_received);
-            stream_interactor.get_module(MessageManager.IDENTITY).pre_message_send.connect(on_pre_message_send);
+            stream_interactor.get_module(MessageProcessor.IDENTITY).pre_message_send.connect(on_pre_message_send);
         }
 
         private void on_pre_message_received(Entities.Message message, Xmpp.Message.Stanza message_stanza, Conversation conversation) {