From 75e51b5ed3b639b9cf7b16b0ddbee7e362c44ef1 Mon Sep 17 00:00:00 2001 From: fiaxh Date: Tue, 4 Apr 2017 15:47:00 +0200 Subject: MessageStorage/Processor, correctly resolve conversations (fixup 22adbd3) --- libdino/src/application.vala | 3 +- libdino/src/entity/conversation.vala | 5 + libdino/src/service/chat_interaction.vala | 8 +- libdino/src/service/conversation_manager.vala | 12 +- .../service/counterpart_interaction_manager.vala | 40 ++-- libdino/src/service/database.vala | 5 +- libdino/src/service/message_manager.vala | 237 --------------------- libdino/src/service/message_processor.vala | 189 ++++++++++++++++ libdino/src/service/message_storage.vala | 68 ++++++ libdino/src/service/muc_manager.vala | 12 +- libdino/src/service/util.vala | 19 ++ 11 files changed, 322 insertions(+), 276 deletions(-) delete mode 100644 libdino/src/service/message_manager.vala create mode 100644 libdino/src/service/message_processor.vala create mode 100644 libdino/src/service/message_storage.vala create mode 100644 libdino/src/service/util.vala (limited to 'libdino/src') diff --git a/libdino/src/application.vala b/libdino/src/application.vala index b28bb69b..37fbda8b 100644 --- a/libdino/src/application.vala +++ b/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); diff --git a/libdino/src/entity/conversation.vala b/libdino/src/entity/conversation.vala index 5a41c7fb..7e8b1424 100644 --- a/libdino/src/entity/conversation.vala +++ b/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; diff --git a/libdino/src/service/chat_interaction.vala b/libdino/src/service/chat_interaction.vala index 9943c3c3..891abf29 100644 --- a/libdino/src/service/chat_interaction.vala +++ b/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(MessageManager.IDENTITY).message_sent.connect(on_message_sent); + stream_interactor.get_module(MessageProcessor.IDENTITY).message_received.connect(on_message_received); + 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; diff --git a/libdino/src/service/conversation_manager.vala b/libdino/src/service/conversation_manager.vala index ff4717ee..db9cff91 100644 --- a/libdino/src/service/conversation_manager.vala +++ b/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(MessageManager.IDENTITY).message_sent.connect(on_message_sent); + stream_interactor.get_module(MessageProcessor.IDENTITY).pre_message_received.connect(on_message_received); + 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 get_conversations_for_presence(Show show, Account account) { + return get_conversations(show.jid, account); + } + + public Gee.List get_conversations(Jid jid, Account account) { Gee.List ret = new ArrayList(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; } diff --git a/libdino/src/service/counterpart_interaction_manager.vala b/libdino/src/service/counterpart_interaction_manager.vala index bfd473a2..75d2d7be 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(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); - if (conversation != null) { - Gee.List? messages = stream_interactor.get_module(MessageManager.IDENTITY).get_messages(conversation); - if (messages != null) { // TODO not here - foreach (Entities.Message message in messages) { - if (message.stanza_id == stanza_id) { - switch (marker) { - case Xep.ChatMarkers.MARKER_RECEIVED: - received_message_received(account, jid, message); - message.marked = Entities.Message.Marked.RECEIVED; - break; - case Xep.ChatMarkers.MARKER_DISPLAYED: - last_read[jid] = message; - received_message_displayed(account, jid, message); - foreach (Entities.Message m in messages) { - 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; + foreach (Conversation conversation in stream_interactor.get_module(ConversationManager.IDENTITY).get_conversations(jid, account)) { + Entities.Message? message = stream_interactor.get_module(MessageStorage.IDENTITY).get_message_by_id(stanza_id, conversation); + if (message != null) { + switch (marker) { + case Xep.ChatMarkers.MARKER_RECEIVED: + received_message_received(account, jid, message); + message.marked = Entities.Message.Marked.RECEIVED; + break; + case Xep.ChatMarkers.MARKER_DISPLAYED: + last_read[jid] = message; + received_message_displayed(account, jid, message); + Gee.List messages = stream_interactor.get_module(MessageStorage.IDENTITY).get_messages(conversation); + foreach (Entities.Message m in messages) { + 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; } } } diff --git a/libdino/src/service/database.vala b/libdino/src/service/database.vala index 797fd6c9..a74ac056 100644 --- a/libdino/src/service/database.vala +++ b/libdino/src/service/database.vala @@ -190,7 +190,10 @@ public class Database : Qlite.Database { public Gee.List get_unsend_messages(Account account) { Gee.List ret = new ArrayList(); - 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; diff --git a/libdino/src/service/message_manager.vala b/libdino/src/service/message_manager.vala deleted file mode 100644 index 73f49237..00000000 --- a/libdino/src/service/message_manager.vala +++ /dev/null @@ -1,237 +0,0 @@ -using Gee; - -using Xmpp; -using Dino.Entities; - -namespace Dino { - -public class MessageManager : StreamInteractionModule, Object { - public static ModuleIdentity IDENTITY = new ModuleIdentity("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); - public signal void message_received(Entities.Message message, Conversation conversation); - public signal void out_message_created(Entities.Message message, Conversation conversation); - public signal void pre_message_send(Entities.Message message, Xmpp.Message.Stanza message_stanza, Conversation conversation); - public signal void message_sent(Entities.Message message, Conversation conversation); - - private StreamInteractor stream_interactor; - private Database db; - private HashMap> messages = new HashMap>(Conversation.hash_func, Conversation.equals_func); - - public static void start(StreamInteractor stream_interactor, Database db) { - MessageManager m = new MessageManager(stream_interactor, db); - stream_interactor.add_module(m); - } - - private MessageManager(StreamInteractor stream_interactor, Database db) { - this.stream_interactor = stream_interactor; - this.db = db; - stream_interactor.account_added.connect(on_account_added); - stream_interactor.connection_manager.connection_state_changed.connect((account, state) => { - if (state == ConnectionManager.ConnectionState.CONNECTED) send_unsent_messages(account); - }); - } - - public void send_message(string text, Conversation conversation) { - Entities.Message message = create_out_message(text, conversation); - add_message(message, conversation); - message.persist(db); - send_xmpp_message(message, conversation); - message_sent(message, conversation); - } - - public Gee.List? get_messages(Conversation conversation, int count = 50) { - if (messages.has_key(conversation) && messages[conversation].size > 0) { - Gee.List 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 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 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? get_messages_before(Conversation? conversation, Entities.Message before) { - Gee.List 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); - }); - } - - private void send_unsent_messages(Account account) { - Gee.List unsend_messages = db.get_unsend_messages(account); - foreach (Entities.Message message in unsend_messages) { - Conversation? conversation = stream_interactor.get_module(ConversationManager.IDENTITY).get_conversation(message.counterpart, account); - if (conversation != null) { - send_xmpp_message(message, conversation, true); - } - } - } - - private void on_message_received(Account account, Xmpp.Message.Stanza message) { - if (message.body == null) return; - - Entities.Message new_message = create_in_message(account, message); - - determine_message_type(account, message, new_message); - Conversation? conversation = stream_interactor.get_module(ConversationManager.IDENTITY).get_conversation_for_message(new_message); - if (conversation != null) process_message(new_message, message); - } - - private Entities.Message create_in_message(Account account, Xmpp.Message.Stanza message) { - Entities.Message new_message = new Entities.Message(message.body); - new_message.account = account; - new_message.stanza_id = message.id; - Jid from_jid = new Jid(message.from); - if (!account.bare_jid.equals_bare(from_jid) || - stream_interactor.get_module(MucManager.IDENTITY).get_nick(from_jid.bare_jid, account) == from_jid.resourcepart) { - new_message.direction = Entities.Message.DIRECTION_RECEIVED; - } else { - new_message.direction = Entities.Message.DIRECTION_SENT; - } - new_message.counterpart = new_message.direction == Entities.Message.DIRECTION_SENT ? new Jid(message.to) : new Jid(message.from); - new_message.ourpart = new_message.direction == Entities.Message.DIRECTION_SENT ? new Jid(message.from) : new Jid(message.to); - new_message.stanza = message; - Xep.DelayedDelivery.MessageFlag? deleyed_delivery_flag = Xep.DelayedDelivery.MessageFlag.get_flag(message); - new_message.time = deleyed_delivery_flag != null ? deleyed_delivery_flag.datetime : new DateTime.now_local(); - new_message.local_time = new DateTime.now_local(); - return new_message; - } - - private void process_message(Entities.Message new_message, Xmpp.Message.Stanza 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))) { - new_message.persist(db); - add_message(new_message, conversation); - if (new_message.direction == Entities.Message.DIRECTION_SENT) { - message_sent(new_message, conversation); - } else { - message_received(new_message, conversation); - } - } - } - } - - private void determine_message_type(Account account, Xmpp.Message.Stanza message_stanza, Entities.Message message) { - if (message_stanza.type_ == Xmpp.Message.Stanza.TYPE_GROUPCHAT) { - message.type_ = Entities.Message.Type.GROUPCHAT; - process_message(message, message_stanza); - } else if (message_stanza.type_ == Xmpp.Message.Stanza.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; - } else if (conversation.type_ == Conversation.Type.GROUPCHAT) { - message.type_ = Entities.Message.Type.GROUPCHAT_PM; - } - } 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 triple = store as Triple; - Entities.Message m = triple.b; - if (identities == null) { - m.type_ = Entities.Message.Type.CHAT; - triple.a.process_message(m, triple.c); - return; - } - foreach (Xep.ServiceDiscovery.Identity identity in identities) { - if (identity.category == Xep.ServiceDiscovery.Identity.CATEGORY_CONFERENCE) { - m.type_ = Entities.Message.Type.GROUPCHAT_PM; - } else { - m.type_ = Entities.Message.Type.CHAT; - } - triple.a.process_message(m, triple.c); - } - }, Triple.create(this, message, message_stanza)); - } - } - } - - private void add_message(Entities.Message message, Conversation conversation) { - if (!messages.has_key(conversation)) { - messages[conversation] = new ArrayList(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.stanza_id = random_uuid(); - message.account = conversation.account; - message.body = text; - message.time = new DateTime.now_local(); - message.local_time = new DateTime.now_local(); - message.direction = Entities.Message.DIRECTION_SENT; - message.counterpart = conversation.counterpart; - message.ourpart = new Jid(conversation.account.bare_jid.to_string() + "/" + conversation.account.resourcepart); - message.marked = Entities.Message.Marked.UNSENT; - message.encryption = conversation.encryption; - - out_message_created(message, conversation); - return message; - } - - public void send_xmpp_message(Entities.Message message, Conversation conversation, bool delayed = false) { - lock (messages) { - Core.XmppStream stream = stream_interactor.get_stream(conversation.account); - message.marked = Entities.Message.Marked.NONE; - if (stream != null) { - Xmpp.Message.Stanza new_message = new Xmpp.Message.Stanza(message.stanza_id); - new_message.to = message.counterpart.to_string(); - new_message.body = message.body; - if (conversation.type_ == Conversation.Type.GROUPCHAT) { - new_message.type_ = Xmpp.Message.Stanza.TYPE_GROUPCHAT; - } else { - new_message.type_ = Xmpp.Message.Stanza.TYPE_CHAT; - } - pre_message_send(message, new_message, conversation); - if (message.marked == Entities.Message.Marked.UNSENT || message.marked == Entities.Message.Marked.WONTSEND) return; - if (delayed) { - Xmpp.Xep.DelayedDelivery.Module.set_message_delay(new_message, message.time); - } - stream.get_module(Xmpp.Message.Module.IDENTITY).send_message(stream, new_message); - message.stanza_id = new_message.id; - message.stanza = new_message; - } else { - message.marked = Entities.Message.Marked.UNSENT; - } - } - } -} - -} \ No newline at end of file diff --git a/libdino/src/service/message_processor.vala b/libdino/src/service/message_processor.vala new file mode 100644 index 00000000..10954672 --- /dev/null +++ b/libdino/src/service/message_processor.vala @@ -0,0 +1,189 @@ +using Gee; + +using Xmpp; +using Dino.Entities; + +namespace Dino { + +public class MessageProcessor : StreamInteractionModule, Object { + public static ModuleIdentity IDENTITY = new ModuleIdentity("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); + public signal void message_received(Entities.Message message, Conversation conversation); + public signal void out_message_created(Entities.Message message, Conversation conversation); + public signal void pre_message_send(Entities.Message message, Xmpp.Message.Stanza message_stanza, Conversation conversation); + public signal void message_sent(Entities.Message message, Conversation conversation); + + private StreamInteractor stream_interactor; + private Database db; + private Object lock_send_unsent; + + public static void start(StreamInteractor stream_interactor, Database db) { + MessageProcessor m = new MessageProcessor(stream_interactor, db); + stream_interactor.add_module(m); + } + + private MessageProcessor(StreamInteractor stream_interactor, Database db) { + this.stream_interactor = stream_interactor; + this.db = db; + stream_interactor.account_added.connect(on_account_added); + stream_interactor.connection_manager.connection_state_changed.connect((account, state) => { + if (state == ConnectionManager.ConnectionState.CONNECTED) send_unsent_messages(account); + }); + } + + public void send_message(string text, Conversation conversation) { + Entities.Message message = create_out_message(text, conversation); + stream_interactor.get_module(MessageStorage.IDENTITY).add_message(message, conversation); + message.persist(db); + send_xmpp_message(message, conversation); + message_sent(message, conversation); + } + + 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); + }); + } + + private void send_unsent_messages(Account account) { + Gee.List unsend_messages = db.get_unsend_messages(account); + foreach (Entities.Message message in unsend_messages) { + Conversation? conversation = stream_interactor.get_module(ConversationManager.IDENTITY).get_conversation(message.counterpart, account); + if (conversation != null) { + send_xmpp_message(message, conversation, true); + } + } + } + + private void on_message_received(Account account, Xmpp.Message.Stanza message) { + if (message.body == null) return; + + Entities.Message new_message = create_in_message(account, message); + + determine_message_type(account, message, new_message); + Conversation? conversation = stream_interactor.get_module(ConversationManager.IDENTITY).get_conversation_for_message(new_message); + if (conversation != null) process_message(new_message, message); + } + + private Entities.Message create_in_message(Account account, Xmpp.Message.Stanza message) { + Entities.Message new_message = new Entities.Message(message.body); + new_message.account = account; + new_message.stanza_id = message.id; + Jid from_jid = new Jid(message.from); + if (!account.bare_jid.equals_bare(from_jid) || + stream_interactor.get_module(MucManager.IDENTITY).get_nick(from_jid.bare_jid, account) == from_jid.resourcepart) { + new_message.direction = Entities.Message.DIRECTION_RECEIVED; + } else { + new_message.direction = Entities.Message.DIRECTION_SENT; + } + new_message.counterpart = new_message.direction == Entities.Message.DIRECTION_SENT ? new Jid(message.to) : new Jid(message.from); + new_message.ourpart = new_message.direction == Entities.Message.DIRECTION_SENT ? new Jid(message.from) : new Jid(message.to); + new_message.stanza = message; + Xep.DelayedDelivery.MessageFlag? deleyed_delivery_flag = Xep.DelayedDelivery.MessageFlag.get_flag(message); + new_message.time = deleyed_delivery_flag != null ? deleyed_delivery_flag.datetime : new DateTime.now_local(); + new_message.local_time = new DateTime.now_local(); + return new_message; + } + + private void process_message(Entities.Message new_message, Xmpp.Message.Stanza 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); + } + } + } + } + + private void determine_message_type(Account account, Xmpp.Message.Stanza message_stanza, Entities.Message message) { + if (message_stanza.type_ == Xmpp.Message.Stanza.TYPE_GROUPCHAT) { + message.type_ = Entities.Message.Type.GROUPCHAT; + process_message(message, message_stanza); + } else if (message_stanza.type_ == Xmpp.Message.Stanza.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; + } else if (conversation.type_ == Conversation.Type.GROUPCHAT) { + message.type_ = Entities.Message.Type.GROUPCHAT_PM; + } + } 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 triple = store as Triple; + Entities.Message m = triple.b; + if (identities == null) { + m.type_ = Entities.Message.Type.CHAT; + triple.a.process_message(m, triple.c); + return; + } + foreach (Xep.ServiceDiscovery.Identity identity in identities) { + if (identity.category == Xep.ServiceDiscovery.Identity.CATEGORY_CONFERENCE) { + m.type_ = Entities.Message.Type.GROUPCHAT_PM; + } else { + m.type_ = Entities.Message.Type.CHAT; + } + triple.a.process_message(m, triple.c); + } + }, Triple.create(this, message, message_stanza)); + } + } + } + + private Entities.Message create_out_message(string text, Conversation conversation) { + Entities.Message message = new Entities.Message(text); + message.type_ = Util.get_message_type_for_conversation(conversation); + message.stanza_id = random_uuid(); + message.account = conversation.account; + message.body = text; + message.time = new DateTime.now_local(); + message.local_time = new DateTime.now_local(); + message.direction = Entities.Message.DIRECTION_SENT; + message.counterpart = conversation.counterpart; + message.ourpart = new Jid(conversation.account.bare_jid.to_string() + "/" + conversation.account.resourcepart); + message.marked = Entities.Message.Marked.UNSENT; + message.encryption = conversation.encryption; + + out_message_created(message, conversation); + return message; + } + + public void send_xmpp_message(Entities.Message message, Conversation conversation, bool delayed = false) { + lock (lock_send_unsent) { + Core.XmppStream stream = stream_interactor.get_stream(conversation.account); + message.marked = Entities.Message.Marked.NONE; + if (stream != null) { + Xmpp.Message.Stanza new_message = new Xmpp.Message.Stanza(message.stanza_id); + new_message.to = message.counterpart.to_string(); + new_message.body = message.body; + if (conversation.type_ == Conversation.Type.GROUPCHAT) { + new_message.type_ = Xmpp.Message.Stanza.TYPE_GROUPCHAT; + } else { + new_message.type_ = Xmpp.Message.Stanza.TYPE_CHAT; + } + pre_message_send(message, new_message, conversation); + if (message.marked == Entities.Message.Marked.UNSENT || message.marked == Entities.Message.Marked.WONTSEND) return; + if (delayed) { + Xmpp.Xep.DelayedDelivery.Module.set_message_delay(new_message, message.time); + } + stream.get_module(Xmpp.Message.Module.IDENTITY).send_message(stream, new_message); + message.stanza_id = new_message.id; + message.stanza = new_message; + } else { + message.marked = Entities.Message.Marked.UNSENT; + } + } + } +} + +} \ No newline at end of file diff --git a/libdino/src/service/message_storage.vala b/libdino/src/service/message_storage.vala new file mode 100644 index 00000000..ea765f96 --- /dev/null +++ b/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 IDENTITY = new ModuleIdentity("message_cache"); + public string id { get { return IDENTITY.id; } } + + private StreamInteractor stream_interactor; + private Database db; + + private HashMap> messages = new HashMap>(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 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(); + } + + 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? get_messages_before(Conversation? conversation, Message before, int count = 20) { + Gee.List 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); + } + } +} + +} \ No newline at end of file diff --git a/libdino/src/service/muc_manager.vala b/libdino/src/service/muc_manager.vala index c5bfb8ba..f63a557c 100644 --- a/libdino/src/service/muc_manager.vala +++ b/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? messages = stream_interactor.get_module(MessageManager.IDENTITY).get_messages(conversation); - if (messages != null) { // TODO not here - foreach (Entities.Message m in messages) { - if (m.equals(message)) { - m.marked = Entities.Message.Marked.RECEIVED; - } + 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; } } } diff --git a/libdino/src/service/util.vala b/libdino/src/service/util.vala new file mode 100644 index 00000000..d0e19dc3 --- /dev/null +++ b/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(); + } +} + +} \ No newline at end of file -- cgit v1.2.3-54-g00ecf