diff options
Diffstat (limited to 'libdino/src/service/message_processor.vala')
| -rw-r--r-- | libdino/src/service/message_processor.vala | 276 |
1 files changed, 20 insertions, 256 deletions
diff --git a/libdino/src/service/message_processor.vala b/libdino/src/service/message_processor.vala index 6445ce40..bfecf340 100644 --- a/libdino/src/service/message_processor.vala +++ b/libdino/src/service/message_processor.vala @@ -18,15 +18,11 @@ public class MessageProcessor : StreamInteractionModule, Object { public signal void message_sent_or_received(Entities.Message message, Conversation conversation); public signal void history_synced(Account account); + public HistorySync history_sync; public MessageListenerHolder received_pipeline = new MessageListenerHolder(); private StreamInteractor stream_interactor; private Database db; - private HashMap<Account, int> current_catchup_id = new HashMap<Account, int>(Account.hash_func, Account.equals_func); - private HashMap<Account, HashMap<string, DateTime>> mam_times = new HashMap<Account, HashMap<string, DateTime>>(); - public HashMap<string, int> hitted_range = new HashMap<string, int>(); - public HashMap<Account, string> catchup_until_id = new HashMap<Account, string>(Account.hash_func, Account.equals_func); - public HashMap<Account, DateTime> catchup_until_time = new HashMap<Account, DateTime>(Account.hash_func, Account.equals_func); public static void start(StreamInteractor stream_interactor, Database db) { MessageProcessor m = new MessageProcessor(stream_interactor, db); @@ -36,6 +32,7 @@ public class MessageProcessor : StreamInteractionModule, Object { private MessageProcessor(StreamInteractor stream_interactor, Database db) { this.stream_interactor = stream_interactor; this.db = db; + this.history_sync = new HistorySync(db, stream_interactor); received_pipeline.connect(new DeduplicateMessageListener(this, db)); received_pipeline.connect(new FilterMessageListener()); @@ -47,11 +44,6 @@ public class MessageProcessor : StreamInteractionModule, Object { stream_interactor.stream_negotiated.connect(send_unsent_chat_messages); stream_interactor.stream_resumed.connect(send_unsent_chat_messages); - - stream_interactor.connection_manager.stream_opened.connect((account, stream) => { - debug("MAM: [%s] Reset catchup_id", account.bare_jid.to_string()); - current_catchup_id.unset(account); - }); } public Entities.Message send_text(string text, Conversation conversation) { @@ -106,43 +98,10 @@ public class MessageProcessor : StreamInteractionModule, Object { } private void on_account_added(Account account) { - mam_times[account] = new HashMap<string, DateTime>(); - stream_interactor.module_manager.get_module(account, Xmpp.MessageModule.IDENTITY).received_message.connect( (stream, message) => { on_message_received.begin(account, message); }); - XmppStream? stream_bak = null; - stream_interactor.module_manager.get_module(account, Xmpp.Xep.MessageArchiveManagement.Module.IDENTITY).feature_available.connect( (stream) => { - if (stream == stream_bak) return; - - current_catchup_id.unset(account); - stream_bak = stream; - debug("MAM: [%s] MAM available", account.bare_jid.to_string()); - do_mam_catchup.begin(account); - }); - stream_interactor.module_manager.get_module(account, Xmpp.MessageModule.IDENTITY).received_message_unprocessed.connect((stream, message) => { - if (!message.from.equals(account.bare_jid)) return; - - Xep.MessageArchiveManagement.Flag? mam_flag = stream != null ? stream.get_flag(Xep.MessageArchiveManagement.Flag.IDENTITY) : null; - if (mam_flag == null) return; - string? id = message.stanza.get_deep_attribute(mam_flag.ns_ver + ":result", "id"); - if (id == null) return; - StanzaNode? delay_node = message.stanza.get_deep_subnode(mam_flag.ns_ver + ":result", "urn:xmpp:forward:0:forwarded", "urn:xmpp:delay:delay"); - if (delay_node == null) { - warning("MAM result did not contain delayed time %s", message.stanza.to_string()); - return; - } - DateTime? time = DelayedDelivery.get_time_for_node(delay_node); - if (time == null) return; - mam_times[account][id] = time; - - string? query_id = message.stanza.get_deep_attribute(mam_flag.ns_ver + ":result", mam_flag.ns_ver + ":queryid"); - if (query_id != null && id == catchup_until_id[account]) { - debug("MAM: [%s] Hitted range (id) %s", account.bare_jid.to_string(), id); - hitted_range[query_id] = -2; - } - }); stream_interactor.module_manager.get_module(account, Xmpp.MessageModule.IDENTITY).received_error.connect((stream, message_stanza, error_stanza) => { Message? message = null; @@ -164,203 +123,20 @@ public class MessageProcessor : StreamInteractionModule, Object { convert_sending_to_unsent_msgs(account); } - private async void do_mam_catchup(Account account) { - debug("MAM: [%s] Start catchup", account.bare_jid.to_string()); - string? earliest_id = null; - DateTime? earliest_time = null; - bool continue_sync = true; - - while (continue_sync) { - continue_sync = false; - - // Get previous row - var previous_qry = db.mam_catchup.select().with(db.mam_catchup.account_id, "=", account.id).order_by(db.mam_catchup.to_time, "DESC"); - if (current_catchup_id.has_key(account)) { - previous_qry.with(db.mam_catchup.id, "!=", current_catchup_id[account]); - } - RowOption previous_row = previous_qry.single().row(); - if (previous_row.is_present()) { - catchup_until_id[account] = previous_row[db.mam_catchup.to_id]; - catchup_until_time[account] = (new DateTime.from_unix_utc(previous_row[db.mam_catchup.to_time])).add_minutes(-5); - debug("MAM: [%s] Previous entry exists", account.bare_jid.to_string()); - } else { - catchup_until_id.unset(account); - catchup_until_time.unset(account); - } - - string query_id = Xmpp.random_uuid(); - yield get_mam_range(account, query_id, null, null, earliest_time, earliest_id); - - if (!hitted_range.has_key(query_id)) { - debug("MAM: [%s] Set catchup end reached", account.bare_jid.to_string()); - db.mam_catchup.update() - .set(db.mam_catchup.from_end, true) - .with(db.mam_catchup.id, "=", current_catchup_id[account]) - .perform(); - } - - if (hitted_range.has_key(query_id)) { - if (merge_ranges(account, null)) { - RowOption current_row = db.mam_catchup.row_with(db.mam_catchup.id, current_catchup_id[account]); - bool range_from_complete = current_row[db.mam_catchup.from_end]; - if (!range_from_complete) { - continue_sync = true; - earliest_id = current_row[db.mam_catchup.from_id]; - earliest_time = (new DateTime.from_unix_utc(current_row[db.mam_catchup.from_time])).add_seconds(1); - } - } - } - } - } - - /* - * Merges the row with `current_catchup_id` with the previous range (optional: with `earlier_id`) - * Changes `current_catchup_id` to the previous range - */ - private bool merge_ranges(Account account, int? earlier_id) { - RowOption current_row = db.mam_catchup.row_with(db.mam_catchup.id, current_catchup_id[account]); - RowOption previous_row = null; - - if (earlier_id != null) { - previous_row = db.mam_catchup.row_with(db.mam_catchup.id, earlier_id); - } else { - previous_row = db.mam_catchup.select() - .with(db.mam_catchup.account_id, "=", account.id) - .with(db.mam_catchup.id, "!=", current_catchup_id[account]) - .order_by(db.mam_catchup.to_time, "DESC").single().row(); - } - - if (!previous_row.is_present()) { - debug("MAM: [%s] Merging: No previous row", account.bare_jid.to_string()); - return false; - } - - var qry = db.mam_catchup.update().with(db.mam_catchup.id, "=", previous_row[db.mam_catchup.id]); - debug("MAM: [%s] Merging %ld-%ld with %ld- %ld", account.bare_jid.to_string(), previous_row[db.mam_catchup.from_time], previous_row[db.mam_catchup.to_time], current_row[db.mam_catchup.from_time], current_row[db.mam_catchup.to_time]); - if (current_row[db.mam_catchup.from_time] < previous_row[db.mam_catchup.from_time]) { - qry.set(db.mam_catchup.from_id, current_row[db.mam_catchup.from_id]) - .set(db.mam_catchup.from_time, current_row[db.mam_catchup.from_time]); - } - if (current_row[db.mam_catchup.to_time] > previous_row[db.mam_catchup.to_time]) { - qry.set(db.mam_catchup.to_id, current_row[db.mam_catchup.to_id]) - .set(db.mam_catchup.to_time, current_row[db.mam_catchup.to_time]); - } - qry.perform(); - - current_catchup_id[account] = previous_row[db.mam_catchup.id]; - - db.mam_catchup.delete().with(db.mam_catchup.id, "=", current_row[db.mam_catchup.id]).perform(); - - return true; - } - - private async bool get_mam_range(Account account, string? query_id, DateTime? from_time, string? from_id, DateTime? to_time, string? to_id) { - debug("MAM: [%s] Get range %s - %s", account.bare_jid.to_string(), from_time != null ? from_time.to_string() : "", to_time != null ? to_time.to_string() : ""); - XmppStream stream = stream_interactor.get_stream(account); - - Iq.Stanza? iq = yield stream.get_module(Xep.MessageArchiveManagement.Module.IDENTITY).query_archive(stream, null, query_id, from_time, from_id, to_time, to_id); - - if (iq == null) { - debug(@"MAM: [%s] IQ null", account.bare_jid.to_string()); - return true; - } - - if (iq.stanza.get_deep_string_content("urn:xmpp:mam:2:fin", "http://jabber.org/protocol/rsm" + ":set", "first") == null) { - return true; - } - - while (iq != null) { - string? earliest_id = iq.stanza.get_deep_string_content("urn:xmpp:mam:2:fin", "http://jabber.org/protocol/rsm" + ":set", "first"); - if (earliest_id == null) return true; - string? latest_id = iq.stanza.get_deep_string_content("urn:xmpp:mam:2:fin", "http://jabber.org/protocol/rsm" + ":set", "last"); - - // We wait until all the messages from the page are processed (and we got the `mam_times` from them) - Idle.add(get_mam_range.callback, Priority.LOW); - yield; - - int wait_ms = 1000; - - - if (mam_times[account].has_key(earliest_id) && (current_catchup_id.has_key(account) || mam_times[account].has_key(latest_id))) { - - debug("MAM: [%s] Update from_id %s", account.bare_jid.to_string(), earliest_id); - if (!current_catchup_id.has_key(account)) { - debug("MAM: [%s] We get our first MAM page", account.bare_jid.to_string()); - current_catchup_id[account] = (int) db.mam_catchup.insert() - .value(db.mam_catchup.account_id, account.id) - .value(db.mam_catchup.from_id, earliest_id) - .value(db.mam_catchup.from_time, (long)mam_times[account][earliest_id].to_unix()) - .value(db.mam_catchup.to_id, latest_id) - .value(db.mam_catchup.to_time, (long)mam_times[account][latest_id].to_unix()) - .perform(); - } else { - // Update existing id - db.mam_catchup.update() - .set(db.mam_catchup.from_id, earliest_id) - .set(db.mam_catchup.from_time, (long)mam_times[account][earliest_id].to_unix()) - .with(db.mam_catchup.id, "=", current_catchup_id[account]) - .perform(); - } - - TimeSpan catchup_time_ago = (new DateTime.now_utc()).difference(mam_times[account][earliest_id]); - - if (catchup_time_ago > 14 * TimeSpan.DAY) { - wait_ms = 2000; - } else if (catchup_time_ago > 5 * TimeSpan.DAY) { - wait_ms = 1000; - } else if (catchup_time_ago > 2 * TimeSpan.DAY) { - wait_ms = 200; - } else if (catchup_time_ago > TimeSpan.DAY) { - wait_ms = 50; - } else { - wait_ms = 10; - } - } else { - warning("Didn't have time for MAM id; earliest_id:%s latest_id:%s", mam_times[account].has_key(earliest_id).to_string(), mam_times[account].has_key(latest_id).to_string()); - } - - mam_times[account] = new HashMap<string, DateTime>(); + private async void on_message_received(Account account, Xmpp.MessageStanza message_stanza) { - Timeout.add(wait_ms, () => { - if (hitted_range.has_key(query_id)) { - debug(@"MAM: [%s] Hitted contains key %s", account.bare_jid.to_string(), query_id); - iq = null; - Idle.add(get_mam_range.callback); - return false; - } + // If it's a message from MAM, it's going to be processed by HistorySync which calls run_pipeline_announce later. + if (history_sync.process(account, message_stanza)) return; - stream.get_module(Xep.MessageArchiveManagement.Module.IDENTITY).page_through_results.begin(stream, null, query_id, from_time, to_time, iq, (_, res) => { - iq = stream.get_module(Xep.MessageArchiveManagement.Module.IDENTITY).page_through_results.end(res); - Idle.add(get_mam_range.callback); - }); - return false; - }); - yield; - } - return false; + run_pipeline_announce(account, message_stanza); } - private async void on_message_received(Account account, Xmpp.MessageStanza message_stanza) { + public async void run_pipeline_announce(Account account, Xmpp.MessageStanza message_stanza) { Entities.Message message = yield parse_message_stanza(account, message_stanza); Conversation? conversation = stream_interactor.get_module(ConversationManager.IDENTITY).get_conversation_for_message(message); if (conversation == null) return; - // MAM state database update - Xep.MessageArchiveManagement.MessageFlag? mam_flag = Xep.MessageArchiveManagement.MessageFlag.get_flag(message_stanza); - if (mam_flag == null) { - if (current_catchup_id.has_key(account)) { - string? stanza_id = UniqueStableStanzaIDs.get_stanza_id(message_stanza, account.bare_jid); - if (stanza_id != null) { - db.mam_catchup.update() - .with(db.mam_catchup.id, "=", current_catchup_id[account]) - .set(db.mam_catchup.to_time, (long)message.local_time.to_unix()) - .set(db.mam_catchup.to_id, stanza_id) - .perform(); - } - } - } - bool abort = yield received_pipeline.run(message, message_stanza, conversation); if (abort) return; @@ -373,7 +149,7 @@ public class MessageProcessor : StreamInteractionModule, Object { message_sent_or_received(message, conversation); } - private async Entities.Message parse_message_stanza(Account account, Xmpp.MessageStanza message) { + public async Entities.Message parse_message_stanza(Account account, Xmpp.MessageStanza message) { string? body = message.body; if (body != null) body = body.strip(); Entities.Message new_message = new Entities.Message(body); @@ -393,20 +169,20 @@ public class MessageProcessor : StreamInteractionModule, Object { new_message.ourpart = new_message.direction == Entities.Message.DIRECTION_SENT ? message.from : message.to; XmppStream? stream = stream_interactor.get_stream(account); - Xep.MessageArchiveManagement.MessageFlag? mam_message_flag = Xep.MessageArchiveManagement.MessageFlag.get_flag(message); - Xep.MessageArchiveManagement.Flag? mam_flag = stream != null ? stream.get_flag(Xep.MessageArchiveManagement.Flag.IDENTITY) : null; + Xmpp.MessageArchiveManagement.MessageFlag? mam_message_flag = Xmpp.MessageArchiveManagement.MessageFlag.get_flag(message); + Xmpp.MessageArchiveManagement.Flag? mam_flag = stream != null ? stream.get_flag(Xmpp.MessageArchiveManagement.Flag.IDENTITY) : null; EntityInfo entity_info = stream_interactor.get_module(EntityInfo.IDENTITY); - if (mam_message_flag != null && mam_flag != null && mam_flag.ns_ver == Xep.MessageArchiveManagement.NS_URI_2 && mam_message_flag.mam_id != null) { + if (mam_message_flag != null && mam_flag != null && mam_flag.ns_ver == Xmpp.MessageArchiveManagement.NS_URI_2 && mam_message_flag.mam_id != null) { new_message.server_id = mam_message_flag.mam_id; } else if (message.type_ == Xmpp.MessageStanza.TYPE_GROUPCHAT) { bool server_supports_sid = (yield entity_info.has_feature(account, new_message.counterpart.bare_jid, Xep.UniqueStableStanzaIDs.NS_URI)) || - (yield entity_info.has_feature(account, new_message.counterpart.bare_jid, Xep.MessageArchiveManagement.NS_URI_2)); + (yield entity_info.has_feature(account, new_message.counterpart.bare_jid, Xmpp.MessageArchiveManagement.NS_URI_2)); if (server_supports_sid) { new_message.server_id = Xep.UniqueStableStanzaIDs.get_stanza_id(message, new_message.counterpart.bare_jid); } } else if (message.type_ == Xmpp.MessageStanza.TYPE_CHAT) { bool server_supports_sid = (yield entity_info.has_feature(account, account.bare_jid, Xep.UniqueStableStanzaIDs.NS_URI)) || - (yield entity_info.has_feature(account, account.bare_jid, Xep.MessageArchiveManagement.NS_URI_2)); + (yield entity_info.has_feature(account, account.bare_jid, Xmpp.MessageArchiveManagement.NS_URI_2)); if (server_supports_sid) { new_message.server_id = Xep.UniqueStableStanzaIDs.get_stanza_id(message, account.bare_jid); } @@ -474,7 +250,6 @@ public class MessageProcessor : StreamInteractionModule, Object { public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) { Account account = conversation.account; - Xep.MessageArchiveManagement.MessageFlag? mam_flag = Xep.MessageArchiveManagement.MessageFlag.get_flag(stanza); // Deduplicate by server_id if (message.server_id != null) { @@ -482,16 +257,12 @@ public class MessageProcessor : StreamInteractionModule, Object { .with(db.message.server_id, "=", message.server_id) .with(db.message.counterpart_id, "=", db.get_jid_id(message.counterpart)) .with(db.message.account_id, "=", account.id); - bool duplicate = builder.count() > 0; - if (duplicate && mam_flag != null) { - debug(@"MAM: [%s] Hitted range duplicate server id. id %s qid %s", account.bare_jid.to_string(), message.server_id, mam_flag.query_id); - if (outer.catchup_until_time.has_key(account) && mam_flag.server_time.compare(outer.catchup_until_time[account]) < 0) { - outer.hitted_range[mam_flag.query_id] = -1; - debug(@"MAM: [%s] In range (time) %s < %s", account.bare_jid.to_string(), mam_flag.server_time.to_string(), outer.catchup_until_time[account].to_string()); - } + // If the message is a duplicate + if (builder.count() > 0) { + outer.history_sync.on_server_id_duplicate(account, stanza, message); + return true; } - if (duplicate) return true; } // Deduplicate messages by uuid @@ -514,14 +285,7 @@ public class MessageProcessor : StreamInteractionModule, Object { builder.with_null(db.message.our_resource); } } - RowOption row_opt = builder.single().row(); - bool duplicate = row_opt.is_present(); - - if (duplicate && mam_flag != null && row_opt[db.message.server_id] == null && - outer.catchup_until_time.has_key(account) && mam_flag.server_time.compare(outer.catchup_until_time[account]) > 0) { - outer.hitted_range[mam_flag.query_id] = -1; - debug(@"MAM: [%s] Hitted range duplicate message id. id %s qid %s", account.bare_jid.to_string(), message.stanza_id, mam_flag.query_id); - } + bool duplicate = builder.single().row().is_present(); return duplicate; } @@ -608,9 +372,9 @@ public class MessageProcessor : StreamInteractionModule, Object { } public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) { - bool is_mam_message = Xep.MessageArchiveManagement.MessageFlag.get_flag(stanza) != null; + bool is_mam_message = Xmpp.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; + Xmpp.MessageArchiveManagement.Flag? mam_flag = stream != null ? stream.get_flag(Xmpp.MessageArchiveManagement.Flag.IDENTITY) : null; if (is_mam_message || (mam_flag != null && mam_flag.cought_up == true)) { conversation.account.mam_earliest_synced = message.local_time; } |