diff options
| -rw-r--r-- | libdino/src/service/conversation_manager.vala | 2 | ||||
| -rw-r--r-- | libdino/src/service/database.vala | 71 | ||||
| -rw-r--r-- | libdino/src/service/message_processor.vala | 313 | ||||
| -rw-r--r-- | libdino/src/service/notification_events.vala | 34 | ||||
| -rw-r--r-- | plugins/crypto-vala/src/cipher_converter.vala | 4 | ||||
| -rw-r--r-- | xmpp-vala/src/module/message/module.vala | 4 | ||||
| -rw-r--r-- | xmpp-vala/src/module/xep/0045_muc/module.vala | 9 | ||||
| -rw-r--r-- | xmpp-vala/src/module/xep/0313_message_archive_management.vala | 83 |
8 files changed, 401 insertions, 119 deletions
diff --git a/libdino/src/service/conversation_manager.vala b/libdino/src/service/conversation_manager.vala index 1ba53b35..c473ea77 100644 --- a/libdino/src/service/conversation_manager.vala +++ b/libdino/src/service/conversation_manager.vala @@ -163,7 +163,7 @@ public class ConversationManager : StreamInteractionModule, Object { if (stanza != null) { bool is_mam_message = Xep.MessageArchiveManagement.MessageFlag.get_flag(stanza) != null; - bool is_recent = message.local_time.compare(new DateTime.now_utc().add_hours(-24)) > 0; + bool is_recent = message.local_time.compare(new DateTime.now_utc().add_days(-3)) > 0; if (is_mam_message && !is_recent) return false; } stream_interactor.get_module(ConversationManager.IDENTITY).start_conversation(conversation); diff --git a/libdino/src/service/database.vala b/libdino/src/service/database.vala index 54854fd1..e230c3af 100644 --- a/libdino/src/service/database.vala +++ b/libdino/src/service/database.vala @@ -7,7 +7,7 @@ using Dino.Entities; namespace Dino { public class Database : Qlite.Database { - private const int VERSION = 10; + private const int VERSION = 11; public class AccountTable : Table { public Column<int> id = new Column.Integer("id") { primary_key = true, auto_increment = true }; @@ -185,6 +185,21 @@ public class Database : Qlite.Database { } } + public class MamCatchupTable : Table { + public Column<int> id = new Column.Integer("id") { primary_key = true, auto_increment = true }; + public Column<int> account_id = new Column.Integer("account_id") { not_null = true }; + public Column<bool> from_end = new Column.BoolInt("from_end"); + public Column<string> from_id = new Column.Text("from_id"); + public Column<long> from_time = new Column.Long("from_time") { not_null = true }; + public Column<string> to_id = new Column.Text("to_id"); + public Column<long> to_time = new Column.Long("to_time") { not_null = true }; + + internal MamCatchupTable(Database db) { + base(db, "mam_catchup"); + init({id, account_id, from_end, from_id, from_time, to_id, to_time}); + } + } + public class SettingsTable : Table { public Column<int> id = new Column.Integer("id") { primary_key = true, auto_increment = true }; public Column<string> key = new Column.Text("key") { unique = true, not_null = true }; @@ -206,6 +221,7 @@ public class Database : Qlite.Database { public AvatarTable avatar { get; private set; } public EntityFeatureTable entity_feature { get; private set; } public RosterTable roster { get; private set; } + public MamCatchupTable mam_catchup { get; private set; } public SettingsTable settings { get; private set; } public Map<int, Jid> jid_table_cache = new HashMap<int, Jid>(); @@ -224,8 +240,9 @@ public class Database : Qlite.Database { avatar = new AvatarTable(this); entity_feature = new EntityFeatureTable(this); roster = new RosterTable(this); + mam_catchup = new MamCatchupTable(this); settings = new SettingsTable(this); - init({ account, jid, content_item, message, real_jid, file_transfer, conversation, avatar, entity_feature, roster, settings }); + init({ account, jid, content_item, message, real_jid, file_transfer, conversation, avatar, entity_feature, roster, mam_catchup, settings }); try { exec("PRAGMA synchronous=0"); } catch (Error e) { } @@ -280,6 +297,15 @@ public class Database : Qlite.Database { error("Failed to upgrade to database version 9: %s", e.message); } } + if (oldVersion < 11) { + try { + exec(""" + insert into mam_catchup (account_id, from_end, from_time, to_time) + select id, 1, 0, mam_earliest_synced from account where mam_earliest_synced not null and mam_earliest_synced > 0"""); + } catch (Error e) { + error("Failed to upgrade to database version 11: %s", e.message); + } + } } public ArrayList<Account> get_accounts() { @@ -373,47 +399,6 @@ public class Database : Qlite.Database { return ret; } - public bool contains_message(Message query_message, Account account) { - QueryBuilder builder = message.select() - .with(message.account_id, "=", account.id) - .with(message.counterpart_id, "=", get_jid_id(query_message.counterpart)) - .with(message.body, "=", query_message.body) - .with(message.time, "<", (long) query_message.time.add_minutes(1).to_unix()) - .with(message.time, ">", (long) query_message.time.add_minutes(-1).to_unix()); - if (query_message.stanza_id != null) { - builder.with(message.stanza_id, "=", query_message.stanza_id); - } else { - builder.with_null(message.stanza_id); - } - if (query_message.counterpart.resourcepart != null) { - builder.with(message.counterpart_resource, "=", query_message.counterpart.resourcepart); - } else { - builder.with_null(message.counterpart_resource); - } - return builder.count() > 0; - } - - public bool contains_message_by_stanza_id(Message query_message, Account account) { - QueryBuilder builder = message.select() - .with(message.stanza_id, "=", query_message.stanza_id) - .with(message.counterpart_id, "=", get_jid_id(query_message.counterpart)) - .with(message.account_id, "=", account.id); - if (query_message.counterpart.resourcepart != null) { - builder.with(message.counterpart_resource, "=", query_message.counterpart.resourcepart); - } else { - builder.with_null(message.counterpart_resource); - } - return builder.count() > 0; - } - - public bool contains_message_by_server_id(Account account, Jid counterpart, string server_id) { - QueryBuilder builder = message.select() - .with(message.server_id, "=", server_id) - .with(message.counterpart_id, "=", get_jid_id(counterpart)) - .with(message.account_id, "=", account.id); - return builder.count() > 0; - } - public Message? get_message_by_id(int id) { Row? row = message.row_with(message.id, id).inner; if (row != null) { diff --git a/libdino/src/service/message_processor.vala b/libdino/src/service/message_processor.vala index 604578b5..98239eb8 100644 --- a/libdino/src/service/message_processor.vala +++ b/libdino/src/service/message_processor.vala @@ -1,7 +1,9 @@ using Gee; using Xmpp; +using Xmpp.Xep; using Dino.Entities; +using Qlite; namespace Dino { @@ -20,6 +22,11 @@ public class MessageProcessor : StreamInteractionModule, Object { private StreamInteractor stream_interactor; private Database db; private Object lock_send_unsent; + 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); @@ -29,14 +36,23 @@ public class MessageProcessor : StreamInteractionModule, Object { private MessageProcessor(StreamInteractor stream_interactor, Database db) { this.stream_interactor = stream_interactor; this.db = db; + + received_pipeline.connect(new DeduplicateMessageListener(this, db)); + received_pipeline.connect(new FilterMessageListener()); + received_pipeline.connect(new StoreMessageListener(stream_interactor)); + received_pipeline.connect(new MamMessageListener(stream_interactor)); + 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); }); - received_pipeline.connect(new DeduplicateMessageListener(db)); - received_pipeline.connect(new FilterMessageListener()); - received_pipeline.connect(new StoreMessageListener(stream_interactor)); - received_pipeline.connect(new MamMessageListener(stream_interactor)); + + 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); + mam_times[account] = new HashMap<string, DateTime>(); + }); } public Entities.Message send_text(string text, Conversation conversation) { @@ -65,22 +81,227 @@ public class MessageProcessor : StreamInteractionModule, Object { 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) => { - DateTime start_time = account.mam_earliest_synced.to_unix() > 60 ? account.mam_earliest_synced.add_minutes(-1) : account.mam_earliest_synced; - stream.get_module(Xep.MessageArchiveManagement.Module.IDENTITY).query_archive(stream, null, start_time, null, () => { - history_synced(account); - }); + 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) return; + DateTime? time = DelayedDelivery.Module.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; + } }); } + 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) { + debug("MAM: [%s] IN: %s", account.bare_jid.to_string(), iq.stanza.to_string()); + 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; + + if (!mam_times[account].has_key(earliest_id)) error("wtf"); + + debug("MAM: [%s] Update from_id %s\n", 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()); + string? latest_id = iq.stanza.get_deep_string_content("urn:xmpp:mam:2:fin", "http://jabber.org/protocol/rsm" + ":set", "last"); + if (!mam_times[account].has_key(latest_id)) error("wtf2"); + 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()) // need to make sure we have this + .with(db.mam_catchup.id, "=", current_catchup_id[account]) + .perform(); + } + + TimeSpan catchup_time_ago = (new DateTime.now_utc()).difference(mam_times[account][earliest_id]); + int wait_ms = 10; + 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; + } + + mam_times[account] = new HashMap<string, DateTime>(); + + 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; + } + + 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; + } + private async void on_message_received(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) { - bool abort = yield received_pipeline.run(message, message_stanza, conversation); - if (abort) return; + 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; + if (message.direction == Entities.Message.DIRECTION_RECEIVED) { message_received(message, conversation); } else if (message.direction == Entities.Message.DIRECTION_SENT) { @@ -170,24 +391,78 @@ public class MessageProcessor : StreamInteractionModule, Object { public override string action_group { get { return "DEDUPLICATE"; } } public override string[] after_actions { get { return after_actions_const; } } + private MessageProcessor outer; private Database db; - public DeduplicateMessageListener(Database db) { + public DeduplicateMessageListener(MessageProcessor outer, Database db) { + this.outer = outer; this.db = db; } 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) { - return db.contains_message_by_server_id(conversation.account, message.counterpart, message.server_id); - } else if (message.stanza_id != null) { - bool is_uuid = 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); - if (is_uuid) { - return db.contains_message_by_stanza_id(message, conversation.account); + QueryBuilder builder = db.message.select() + .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 (duplicate) return true; + } + + // Deduplicate messages by uuid + 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); + if (is_uuid) { + QueryBuilder builder = db.message.select() + .with(db.message.stanza_id, "=", message.stanza_id) + .with(db.message.counterpart_id, "=", db.get_jid_id(message.counterpart)) + .with(db.message.account_id, "=", account.id); + if (message.counterpart.resourcepart != null) { + builder.with(db.message.counterpart_resource, "=", message.counterpart.resourcepart); } else { - return db.contains_message(message, conversation.account); + builder.with_null(db.message.counterpart_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); + } + return duplicate; } - return false; + + // Deduplicate messages based on content and metadata + QueryBuilder builder = db.message.select() + .with(db.message.account_id, "=", account.id) + .with(db.message.counterpart_id, "=", db.get_jid_id(message.counterpart)) + .with(db.message.body, "=", message.body) + .with(db.message.time, "<", (long) message.time.add_minutes(1).to_unix()) + .with(db.message.time, ">", (long) message.time.add_minutes(-1).to_unix()); + if (message.stanza_id != null) { + builder.with(db.message.stanza_id, "=", message.stanza_id); + } else { + builder.with_null(db.message.stanza_id); + } + if (message.counterpart.resourcepart != null) { + builder.with(db.message.counterpart_resource, "=", message.counterpart.resourcepart); + } else { + builder.with_null(db.message.counterpart_resource); + } + return builder.count() > 0; } } diff --git a/libdino/src/service/notification_events.vala b/libdino/src/service/notification_events.vala index f47b9a0a..dca81af8 100644 --- a/libdino/src/service/notification_events.vala +++ b/libdino/src/service/notification_events.vala @@ -16,9 +16,6 @@ public class NotificationEvents : StreamInteractionModule, Object { private StreamInteractor stream_interactor; - private HashMap<Account, HashMap<Conversation, ContentItem>> mam_potential_new = new HashMap<Account, HashMap<Conversation, ContentItem>>(Account.hash_func, Account.equals_func); - private Gee.List<Account> synced_accounts = new ArrayList<Account>(Account.equals_func); - public static void start(StreamInteractor stream_interactor) { NotificationEvents m = new NotificationEvents(stream_interactor); stream_interactor.add_module(m); @@ -31,36 +28,19 @@ public class NotificationEvents : StreamInteractionModule, Object { stream_interactor.get_module(PresenceManager.IDENTITY).received_subscription_request.connect(on_received_subscription_request); stream_interactor.get_module(MucManager.IDENTITY).invite_received.connect((account, room_jid, from_jid, password, reason) => notify_muc_invite(account, room_jid, from_jid, password, reason)); stream_interactor.connection_manager.connection_error.connect((account, error) => notify_connection_error(account, error)); - stream_interactor.get_module(MessageProcessor.IDENTITY).history_synced.connect((account) => { - synced_accounts.add(account); - if (!mam_potential_new.has_key(account)) return; - foreach (Conversation c in mam_potential_new[account].keys) { - ContentItem last_mam_item = mam_potential_new[account][c]; - ContentItem last_item = stream_interactor.get_module(ContentItemStore.IDENTITY).get_latest(c); - if (last_mam_item == last_item /* && !c.read_up_to.equals(m) */) { - on_content_item_received(last_mam_item, c); - } - } - mam_potential_new[account].clear(); - }); } private void on_content_item_received(ContentItem item, Conversation conversation) { - // Don't wait for MAM sync on servers without MAM - bool mam_available = true; - XmppStream? stream = stream_interactor.get_stream(conversation.account); - if (stream != null) { - mam_available = stream.get_flag(Xep.MessageArchiveManagement.Flag.IDENTITY) != null; - } + ContentItem last_item = stream_interactor.get_module(ContentItemStore.IDENTITY).get_latest(conversation); - if (mam_available && !synced_accounts.contains(conversation.account)) { - if (!mam_potential_new.has_key(conversation.account)) { - mam_potential_new[conversation.account] = new HashMap<Conversation, ContentItem>(Conversation.hash_func, Conversation.equals_func); - } - mam_potential_new[conversation.account][conversation] = item; - return; + bool not_read_up_to = true; + MessageItem message_item = item as MessageItem; + if (message_item != null) { + not_read_up_to = conversation.read_up_to != null && !conversation.read_up_to.equals(message_item.message); } + if (item.id != last_item.id && not_read_up_to) return; + if (!should_notify(item, conversation)) return; if (stream_interactor.get_module(ChatInteraction.IDENTITY).is_active_focus()) return; notify_content_item(item, conversation); diff --git a/plugins/crypto-vala/src/cipher_converter.vala b/plugins/crypto-vala/src/cipher_converter.vala index f1363fb0..b2b52c5a 100644 --- a/plugins/crypto-vala/src/cipher_converter.vala +++ b/plugins/crypto-vala/src/cipher_converter.vala @@ -85,9 +85,7 @@ public class SymmetricCipherDecrypter : SymmetricCipherConverter { inbuf.length += (int) attached_taglen; if ((flags & ConverterFlags.INPUT_AT_END) != 0) { if (attached_taglen > 0) { - print("Checking tag\n"); check_tag(inbuf[(inbuf.length - attached_taglen):inbuf.length]); - print("tag ok\n"); bytes_read = inbuf.length; } return ConverterResult.FINISHED; @@ -101,4 +99,4 @@ public class SymmetricCipherDecrypter : SymmetricCipherConverter { } } } -}
\ No newline at end of file +} diff --git a/xmpp-vala/src/module/message/module.vala b/xmpp-vala/src/module/message/module.vala index ab3a7d80..5ddbbe1a 100644 --- a/xmpp-vala/src/module/message/module.vala +++ b/xmpp-vala/src/module/message/module.vala @@ -12,6 +12,7 @@ namespace Xmpp { public StanzaListenerHolder<MessageStanza> send_pipeline = new StanzaListenerHolder<MessageStanza>(); public signal void received_message(XmppStream stream, MessageStanza message); + public signal void received_message_unprocessed(XmppStream stream, MessageStanza message); public void send_message(XmppStream stream, MessageStanza message) { send_pipeline.run.begin(stream, message, (obj, res) => { @@ -21,6 +22,9 @@ namespace Xmpp { public async void received_message_stanza_async(XmppStream stream, StanzaNode node) { MessageStanza message = new MessageStanza.from_stanza(node, stream.get_flag(Bind.Flag.IDENTITY).my_jid); + + received_message_unprocessed(stream, message); + if (!message.is_error()) { bool abort = yield received_pipeline.run(stream, message); if (abort) return; diff --git a/xmpp-vala/src/module/xep/0045_muc/module.vala b/xmpp-vala/src/module/xep/0045_muc/module.vala index b30145ff..59d61f3d 100644 --- a/xmpp-vala/src/module/xep/0045_muc/module.vala +++ b/xmpp-vala/src/module/xep/0045_muc/module.vala @@ -487,16 +487,15 @@ public class ReceivedPipelineListener : StanzaListener<MessageStanza> { StanzaNode? invite_node = x_node.get_subnode("invite", NS_URI_USER); string? password = null; StanzaNode? password_node = x_node.get_subnode("password", NS_URI_USER); - if (password_node != null) - password = password_node.get_string_content(); + if (password_node != null) password = password_node.get_string_content(); if (invite_node != null) { string? from_jid = invite_node.get_attribute("from"); if (from_jid != null) { StanzaNode? reason_node = invite_node.get_subnode("reason", NS_URI_USER); string? reason = null; - if (reason_node != null) - reason = reason_node.get_string_content(); - outer.invite_received(stream, message.from, new Jid(from_jid), password, reason); + if (reason_node != null) reason = reason_node.get_string_content(); + bool is_mam_message = Xep.MessageArchiveManagement.MessageFlag.get_flag(message) != null; // TODO + if (!is_mam_message) outer.invite_received(stream, message.from, new Jid(from_jid), password, reason); return true; } } diff --git a/xmpp-vala/src/module/xep/0313_message_archive_management.vala b/xmpp-vala/src/module/xep/0313_message_archive_management.vala index 4f8cadec..247e1b8e 100644 --- a/xmpp-vala/src/module/xep/0313_message_archive_management.vala +++ b/xmpp-vala/src/module/xep/0313_message_archive_management.vala @@ -14,10 +14,7 @@ public class Module : XmppStreamModule { private ReceivedPipelineListener received_pipeline_listener = new ReceivedPipelineListener(); - public delegate void OnFinished(XmppStream stream); - public void query_archive(XmppStream stream, string? jid, DateTime? start, DateTime? end, owned OnFinished? on_finished = null) { - if (stream.get_flag(Flag.IDENTITY) == null) return; - + private StanzaNode crate_base_query(XmppStream stream, string? jid, string? queryid, DateTime? start, DateTime? end) { DataForms.DataForm data_form = new DataForms.DataForm(); DataForms.DataForm.HiddenField form_type_field = new DataForms.DataForm.HiddenField() { var="FORM_TYPE" }; form_type_field.set_value_string(NS_VER(stream)); @@ -38,8 +35,41 @@ public class Module : XmppStreamModule { data_form.add_field(field); } StanzaNode query_node = new StanzaNode.build("query", NS_VER(stream)).add_self_xmlns().put_node(data_form.get_submit_node()); + if (queryid != null) { + query_node.put_attribute("queryid", queryid); + } + return query_node; + } + + private StanzaNode create_set_rsm_node(string? before_id) { + var before_node = new StanzaNode.build("before", "http://jabber.org/protocol/rsm"); + if (before_id != null) { + before_node.put_node(new StanzaNode.text(before_id)); + } + var max_node = (new StanzaNode.build("max", "http://jabber.org/protocol/rsm")).put_node(new StanzaNode.text("20")); + return (new StanzaNode.build("set", "http://jabber.org/protocol/rsm")).add_self_xmlns() + .put_node(before_node) + .put_node(max_node); + } + + public async Iq.Stanza? query_archive(XmppStream stream, string? jid, string? query_id, DateTime? start_time, string? start_id, DateTime? end_time, string? end_id) { + if (stream.get_flag(Flag.IDENTITY) == null) return null; + + var query_node = crate_base_query(stream, jid, query_id, start_time, end_time); + + query_node.put_node(create_set_rsm_node(end_id)); Iq.Stanza iq = new Iq.Stanza.set(query_node); - stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq, (stream, iq) => { page_through_results(stream, iq, (owned)on_finished); }); + + debug(@"OUT INIT: %s", iq.stanza.to_string()); + + Iq.Stanza? result_iq = null; + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq, (stream, iq) => { + result_iq = iq; + Idle.add(query_archive.callback); + }); + yield; + + return result_iq; } public override void attach(XmppStream stream) { @@ -54,22 +84,30 @@ public class Module : XmppStreamModule { public override string get_ns() { return NS_URI; } public override string get_id() { return IDENTITY.id; } - private static void page_through_results(XmppStream stream, Iq.Stanza iq, owned OnFinished? on_finished = null) { - string? last = iq.stanza.get_deep_string_content(NS_VER(stream) + ":fin", "http://jabber.org/protocol/rsm" + ":set", "last"); - if (last == null) { - stream.get_flag(Flag.IDENTITY).cought_up = true; - if (on_finished != null) on_finished(stream); - return; + public async Iq.Stanza? page_through_results(XmppStream stream, string? jid, string? query_id, DateTime? start_time, DateTime? end_time, Iq.Stanza iq) { + + string? complete = iq.stanza.get_deep_attribute("urn:xmpp:mam:2:fin", "complete"); + if (complete == "true") { + return null; + } + string? first = iq.stanza.get_deep_string_content(NS_VER(stream) + ":fin", "http://jabber.org/protocol/rsm" + ":set", "first"); + if (first == null) { + return null; } - Iq.Stanza paging_iq = new Iq.Stanza.set( - new StanzaNode.build("query", NS_VER(stream)).add_self_xmlns().put_node( - new StanzaNode.build("set", "http://jabber.org/protocol/rsm").add_self_xmlns().put_node( - new StanzaNode.build("after", "http://jabber.org/protocol/rsm").put_node(new StanzaNode.text(last)) - ) - ) - ); - stream.get_module(Iq.Module.IDENTITY).send_iq(stream, paging_iq, (stream, iq) => { page_through_results(stream, iq, (owned)on_finished); }); + var query_node = crate_base_query(stream, jid, query_id, start_time, end_time); + query_node.put_node(create_set_rsm_node(first)); + + Iq.Stanza paging_iq = new Iq.Stanza.set(query_node); + + Iq.Stanza? result_iq = null; + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, paging_iq, (stream, iq) => { + result_iq = iq; + Idle.add(page_through_results.callback); + }); + yield; + + return result_iq; } private void query_availability(XmppStream stream) { @@ -107,7 +145,8 @@ public class ReceivedPipelineListener : StanzaListener<MessageStanza> { StanzaNode? forward_node = message.stanza.get_deep_subnode(NS_VER(stream) + ":result", "urn:xmpp:forward:0:forwarded", DelayedDelivery.NS_URI + ":delay"); DateTime? datetime = DelayedDelivery.Module.get_time_for_node(forward_node); string? mam_id = message.stanza.get_deep_attribute(NS_VER(stream) + ":result", NS_VER(stream) + ":id"); - message.add_flag(new MessageFlag(datetime, mam_id)); + string? query_id = message.stanza.get_deep_attribute(NS_VER(stream) + ":result", NS_VER(stream) + ":queryid"); + message.add_flag(new MessageFlag(datetime, mam_id, query_id)); message.stanza = message_node; message.rerun_parsing = true; @@ -134,10 +173,12 @@ public class MessageFlag : Xmpp.MessageFlag { public DateTime? server_time { get; private set; } public string? mam_id { get; private set; } + public string? query_id { get; private set; } - public MessageFlag(DateTime? server_time, string? mam_id) { + public MessageFlag(DateTime? server_time, string? mam_id, string? query_id) { this.server_time = server_time; this.mam_id = mam_id; + this.query_id = query_id; } public static MessageFlag? get_flag(MessageStanza message) { return (MessageFlag) message.get_flag(NS_URI, ID); } |