Refactor MAM catchup. Fetch from latest to earliest message.
This commit is contained in:
fiaxh
parent
321c3529f3
commit
c3532bdf31
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
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 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);
|
||||
});
|
||||
|
||||
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) {
|
||||
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);
|
||||
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) {
|
||||
return db.contains_message_by_stanza_id(message, conversation.account);
|
||||
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 false;
|
||||
return duplicate;
|
||||
}
|
||||
|
||||
// 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;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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); }
|
||||
|
|
|
|||
Loading…
Reference in a new issue