Pipeline for incoming messages in libdino
This commit is contained in:
parent
ca331e85ef
commit
57c72d2818
|
@ -27,7 +27,7 @@ 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(MessageProcessor.IDENTITY).message_received.connect(on_message_received);
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).received_pipeline.connect(new ReceivedMessageListener(stream_interactor));
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).message_sent.connect(on_message_sent);
|
||||
}
|
||||
|
||||
|
@ -124,19 +124,35 @@ public class ChatInteraction : StreamInteractionModule, Object {
|
|||
return true;
|
||||
}
|
||||
|
||||
private void on_message_received(Entities.Message message, Conversation conversation) {
|
||||
if (Xep.MessageArchiveManagement.MessageFlag.get_flag(message.stanza) != null) return;
|
||||
private class ReceivedMessageListener : MessageListener {
|
||||
|
||||
send_delivery_receipt(conversation, message);
|
||||
if (is_active_focus(conversation)) {
|
||||
check_send_read();
|
||||
public string[] after_actions_const = new string[]{ "" };
|
||||
public override string action_group { get { return "OTHER_NODES"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
private StreamInteractor stream_interactor;
|
||||
|
||||
public ReceivedMessageListener(StreamInteractor stream_interactor) {
|
||||
this.stream_interactor = stream_interactor;
|
||||
}
|
||||
|
||||
public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
|
||||
if (Xep.MessageArchiveManagement.MessageFlag.get_flag(message.stanza) != null) return false;
|
||||
|
||||
ChatInteraction outer = stream_interactor.get_module(ChatInteraction.IDENTITY);
|
||||
outer.send_delivery_receipt(conversation, message);
|
||||
if (outer.is_active_focus(conversation)) {
|
||||
outer.check_send_read();
|
||||
conversation.read_up_to = message;
|
||||
send_chat_marker(conversation, message, Xep.ChatMarkers.MARKER_DISPLAYED);
|
||||
outer.send_chat_marker(conversation, message, Xep.ChatMarkers.MARKER_DISPLAYED);
|
||||
} else {
|
||||
send_chat_marker(conversation, message, Xep.ChatMarkers.MARKER_RECEIVED);
|
||||
outer.send_chat_marker(conversation, message, Xep.ChatMarkers.MARKER_RECEIVED);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private void send_chat_marker(Conversation conversation, Entities.Message message, string marker) {
|
||||
XmppStream stream = stream_interactor.get_stream(conversation.account);
|
||||
if (stream != null &&
|
||||
|
|
|
@ -27,7 +27,7 @@ public class ConversationManager : StreamInteractionModule, Object {
|
|||
stream_interactor.add_module(this);
|
||||
stream_interactor.account_added.connect(on_account_added);
|
||||
stream_interactor.get_module(MucManager.IDENTITY).joined.connect(on_groupchat_joined);
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).message_received.connect(handle_new_message);
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).received_pipeline.connect(new MessageListener(stream_interactor));
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).message_sent.connect(handle_new_message);
|
||||
}
|
||||
|
||||
|
@ -119,6 +119,31 @@ public class ConversationManager : StreamInteractionModule, Object {
|
|||
}
|
||||
}
|
||||
|
||||
private class MessageListener : Dino.MessageListener {
|
||||
|
||||
public string[] after_actions_const = new string[]{ "DEDUPLICATE" };
|
||||
public override string action_group { get { return "MANAGER"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
private StreamInteractor stream_interactor;
|
||||
|
||||
public MessageListener(StreamInteractor stream_interactor) {
|
||||
this.stream_interactor = stream_interactor;
|
||||
}
|
||||
|
||||
public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
|
||||
conversation.last_active = message.time;
|
||||
|
||||
if (message.stanza != null) {
|
||||
bool is_mam_message = Xep.MessageArchiveManagement.MessageFlag.get_flag(message.stanza) != null;
|
||||
bool is_recent = message.local_time.compare(new DateTime.now_utc().add_hours(-24)) > 0;
|
||||
if (is_mam_message && !is_recent) return false;
|
||||
}
|
||||
stream_interactor.get_module(ConversationManager.IDENTITY).start_conversation(conversation);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private void handle_new_message(Entities.Message message, Conversation conversation) {
|
||||
conversation.last_active = message.time;
|
||||
|
||||
|
|
|
@ -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(MessageProcessor.IDENTITY).message_received.connect(on_message_received);
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).received_pipeline.connect(new ReceivedMessageListener(this));
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).message_sent.connect(check_if_got_marker);
|
||||
stream_interactor.stream_negotiated.connect(() => chat_states.clear() );
|
||||
}
|
||||
|
@ -88,8 +88,22 @@ public class CounterpartInteractionManager : StreamInteractionModule, Object {
|
|||
}
|
||||
}
|
||||
|
||||
private void on_message_received(Entities.Message message, Conversation conversation) {
|
||||
on_chat_state_received(conversation.account, conversation.counterpart, Xep.ChatStateNotifications.STATE_ACTIVE);
|
||||
private class ReceivedMessageListener : MessageListener {
|
||||
|
||||
public string[] after_actions_const = new string[]{ "DEDUPLICATE" };
|
||||
public override string action_group { get { return "STORE"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
private CounterpartInteractionManager outer;
|
||||
|
||||
public ReceivedMessageListener(CounterpartInteractionManager outer) {
|
||||
this.outer = outer;
|
||||
}
|
||||
|
||||
public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
|
||||
outer.on_chat_state_received(conversation.account, conversation.counterpart, Xep.ChatStateNotifications.STATE_ACTIVE);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private void on_receipt_received(Account account, Jid jid, string id) {
|
||||
|
|
|
@ -9,12 +9,13 @@ public class MessageProcessor : StreamInteractionModule, Object {
|
|||
public static ModuleIdentity<MessageProcessor> IDENTITY = new ModuleIdentity<MessageProcessor>("message_processor");
|
||||
public string id { get { return IDENTITY.id; } }
|
||||
|
||||
public signal void pre_message_received(Entities.Message message, Xmpp.MessageStanza message_stanza, Conversation conversation);
|
||||
public signal void message_received(Entities.Message message, Conversation conversation);
|
||||
public signal void build_message_stanza(Entities.Message message, Xmpp.MessageStanza message_stanza, Conversation conversation);
|
||||
public signal void pre_message_send(Entities.Message message, Xmpp.MessageStanza message_stanza, Conversation conversation);
|
||||
public signal void message_sent(Entities.Message message, Conversation conversation);
|
||||
|
||||
public MessageListenerHolder received_pipeline = new MessageListenerHolder();
|
||||
|
||||
private StreamInteractor stream_interactor;
|
||||
private Database db;
|
||||
private Object lock_send_unsent;
|
||||
|
@ -31,6 +32,9 @@ public class MessageProcessor : StreamInteractionModule, Object {
|
|||
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 StoreMessageListener(stream_interactor));
|
||||
received_pipeline.connect(new MamMessageListener(stream_interactor));
|
||||
}
|
||||
|
||||
public void send_message(string text, Conversation conversation) {
|
||||
|
@ -52,7 +56,7 @@ public class MessageProcessor : StreamInteractionModule, Object {
|
|||
|
||||
private void on_account_added(Account account) {
|
||||
stream_interactor.module_manager.get_module(account, Xmpp.MessageModule.IDENTITY).received_message.connect( (stream, message) => {
|
||||
on_message_received(account, message);
|
||||
on_message_received.begin(account, message);
|
||||
});
|
||||
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;
|
||||
|
@ -60,16 +64,25 @@ public class MessageProcessor : StreamInteractionModule, Object {
|
|||
});
|
||||
}
|
||||
|
||||
private void on_message_received(Account account, Xmpp.MessageStanza message) {
|
||||
if (message.body == null) return;
|
||||
private async void on_message_received(Account account, Xmpp.MessageStanza message_stanza) {
|
||||
if (message_stanza.body == null) return;
|
||||
|
||||
Entities.Message new_message = create_in_message(account, message);
|
||||
if (new_message == null) return;
|
||||
Entities.Message message = yield create_in_message(account, message_stanza);
|
||||
if (message == null) return;
|
||||
|
||||
determine_message_type(account, message, new_message);
|
||||
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 (message.direction == Entities.Message.DIRECTION_RECEIVED) {
|
||||
message_received(message, conversation);
|
||||
} else if (message.direction == Entities.Message.DIRECTION_SENT) {
|
||||
message_sent(message, conversation);
|
||||
}
|
||||
}
|
||||
|
||||
private Entities.Message? create_in_message(Account account, Xmpp.MessageStanza message) {
|
||||
private async Entities.Message create_in_message(Account account, Xmpp.MessageStanza message) {
|
||||
Entities.Message new_message = new Entities.Message(message.body);
|
||||
new_message.account = account;
|
||||
new_message.stanza_id = message.id;
|
||||
|
@ -108,54 +121,30 @@ public class MessageProcessor : StreamInteractionModule, Object {
|
|||
if (delayed_message_flag != null) new_message.time = delayed_message_flag.datetime;
|
||||
if (new_message.time == null || new_message.time.compare(new_message.local_time) > 0) new_message.time = new_message.local_time;
|
||||
|
||||
new_message.type_ = yield determine_message_type(account, message, new_message);
|
||||
|
||||
return new_message;
|
||||
}
|
||||
|
||||
private void process_message(Entities.Message new_message, Xmpp.MessageStanza 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);
|
||||
}
|
||||
|
||||
bool is_mam_message = Xep.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;
|
||||
if (is_mam_message || (mam_flag != null && mam_flag.cought_up == true)) {
|
||||
conversation.account.mam_earliest_synced = new_message.local_time;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void determine_message_type(Account account, Xmpp.MessageStanza message_stanza, Entities.Message message) {
|
||||
private async Entities.Message.Type determine_message_type(Account account, Xmpp.MessageStanza message_stanza, Entities.Message message) {
|
||||
if (message_stanza.type_ == Xmpp.MessageStanza.TYPE_GROUPCHAT) {
|
||||
message.type_ = Entities.Message.Type.GROUPCHAT;
|
||||
process_message(message, message_stanza);
|
||||
} else if (message_stanza.type_ == Xmpp.MessageStanza.TYPE_CHAT) {
|
||||
return Entities.Message.Type.GROUPCHAT;
|
||||
}
|
||||
if (message_stanza.type_ == Xmpp.MessageStanza.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;
|
||||
return Entities.Message.Type.CHAT;
|
||||
} else if (conversation.type_ == Conversation.Type.GROUPCHAT) {
|
||||
message.type_ = Entities.Message.Type.GROUPCHAT_PM;
|
||||
return Entities.Message.Type.GROUPCHAT_PM;
|
||||
}
|
||||
process_message(message, message_stanza);
|
||||
} else {
|
||||
SourceFunc callback = determine_message_type.callback;
|
||||
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, (stream, identities) => {
|
||||
if (identities == null) {
|
||||
message.type_ = Entities.Message.Type.CHAT;
|
||||
process_message(message, message_stanza);
|
||||
Idle.add((owned) callback);
|
||||
return;
|
||||
}
|
||||
foreach (Xep.ServiceDiscovery.Identity identity in identities) {
|
||||
|
@ -165,10 +154,74 @@ public class MessageProcessor : StreamInteractionModule, Object {
|
|||
message.type_ = Entities.Message.Type.CHAT;
|
||||
}
|
||||
}
|
||||
process_message(message, message_stanza);
|
||||
Idle.add((owned) callback);
|
||||
});
|
||||
yield;
|
||||
}
|
||||
}
|
||||
return Entities.Message.Type.CHAT;
|
||||
}
|
||||
|
||||
private class DeduplicateMessageListener : MessageListener {
|
||||
|
||||
public string[] after_actions_const = new string[]{ "" };
|
||||
public override string action_group { get { return "DEDUPLICATE"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
private Database db;
|
||||
|
||||
public DeduplicateMessageListener(Database db) {
|
||||
this.db = db;
|
||||
}
|
||||
|
||||
public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
|
||||
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);
|
||||
bool new_uuid_msg = is_uuid && !db.contains_message_by_stanza_id(message.stanza_id, conversation.account);
|
||||
bool new_misc_msg = !is_uuid && !db.contains_message(message, conversation.account);
|
||||
bool new_msg = new_uuid_msg || new_misc_msg;
|
||||
return !new_msg;
|
||||
}
|
||||
}
|
||||
|
||||
private class StoreMessageListener : MessageListener {
|
||||
|
||||
public string[] after_actions_const = new string[]{ "DEDUPLICATE" };
|
||||
public override string action_group { get { return "STORE"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
private StreamInteractor stream_interactor;
|
||||
|
||||
public StoreMessageListener(StreamInteractor stream_interactor) {
|
||||
this.stream_interactor = stream_interactor;
|
||||
}
|
||||
|
||||
public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
|
||||
stream_interactor.get_module(MessageStorage.IDENTITY).add_message(message, conversation);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private class MamMessageListener : MessageListener {
|
||||
|
||||
public string[] after_actions_const = new string[]{ "DEDUPLICATE" };
|
||||
public override string action_group { get { return "MAM_NODE"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
private StreamInteractor stream_interactor;
|
||||
|
||||
public MamMessageListener(StreamInteractor stream_interactor) {
|
||||
this.stream_interactor = stream_interactor;
|
||||
}
|
||||
|
||||
public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
|
||||
bool is_mam_message = Xep.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;
|
||||
if (is_mam_message || (mam_flag != null && mam_flag.cought_up == true)) {
|
||||
conversation.account.mam_earliest_synced = message.local_time;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private Entities.Message create_out_message(string text, Conversation conversation) {
|
||||
|
@ -221,4 +274,21 @@ public class MessageProcessor : StreamInteractionModule, Object {
|
|||
}
|
||||
}
|
||||
|
||||
public abstract class MessageListener : Xmpp.OrderedListener {
|
||||
|
||||
public abstract async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation);
|
||||
}
|
||||
|
||||
public class MessageListenerHolder : Xmpp.ListenerHolder {
|
||||
|
||||
public async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
|
||||
foreach (OrderedListener ol in listeners) {
|
||||
MessageListener l = ol as MessageListener;
|
||||
bool stop = yield l.run(message, stanza, conversation);
|
||||
if (stop) return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
@ -16,6 +16,7 @@ public class MucManager : StreamInteractionModule, Object {
|
|||
|
||||
private StreamInteractor stream_interactor;
|
||||
private HashMap<Jid, Xep.Muc.MucEnterError> enter_errors = new HashMap<Jid, Xep.Muc.MucEnterError>(Jid.hash_func, Jid.equals_func);
|
||||
private ReceivedMessageListener received_message_listener;
|
||||
|
||||
public static void start(StreamInteractor stream_interactor) {
|
||||
MucManager m = new MucManager(stream_interactor);
|
||||
|
@ -24,9 +25,10 @@ public class MucManager : StreamInteractionModule, Object {
|
|||
|
||||
private MucManager(StreamInteractor stream_interactor) {
|
||||
this.stream_interactor = stream_interactor;
|
||||
this.received_message_listener = new ReceivedMessageListener(stream_interactor);
|
||||
stream_interactor.account_added.connect(on_account_added);
|
||||
stream_interactor.stream_negotiated.connect(on_stream_negotiated);
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).pre_message_received.connect(on_pre_message_received);
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).received_pipeline.connect(received_message_listener);
|
||||
}
|
||||
|
||||
public void join(Account account, Jid jid, string? nick, string? password) {
|
||||
|
@ -241,27 +243,6 @@ public class MucManager : StreamInteractionModule, Object {
|
|||
});
|
||||
}
|
||||
|
||||
private void on_pre_message_received(Entities.Message message, Xmpp.MessageStanza message_stanza, Conversation conversation) {
|
||||
if (conversation.type_ != Conversation.Type.GROUPCHAT) return;
|
||||
XmppStream stream = stream_interactor.get_stream(conversation.account);
|
||||
if (stream == null) return;
|
||||
if (Xep.DelayedDelivery.MessageFlag.get_flag(message.stanza) == null) {
|
||||
Jid? real_jid = stream.get_flag(Xep.Muc.Flag.IDENTITY).get_real_jid(message.counterpart);
|
||||
if (real_jid != null && !real_jid.equals(message.counterpart)) {
|
||||
message.real_jid = real_jid.bare_jid;
|
||||
}
|
||||
}
|
||||
string? muc_nick = stream.get_flag(Xep.Muc.Flag.IDENTITY).get_muc_nick(conversation.counterpart.bare_jid);
|
||||
if (muc_nick != null && message.from.equals(message.from.with_resource(muc_nick))) { // TODO better from own
|
||||
Gee.List<Entities.Message> 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void join_all_active(Account account) {
|
||||
Gee.List<Conversation> conversations = stream_interactor.get_module(ConversationManager.IDENTITY).get_active_conversations(account);
|
||||
foreach (Conversation conversation in conversations) {
|
||||
|
@ -336,6 +317,41 @@ public class MucManager : StreamInteractionModule, Object {
|
|||
}
|
||||
});
|
||||
}
|
||||
|
||||
private class ReceivedMessageListener : MessageListener {
|
||||
|
||||
public string[] after_actions_const = new string[]{ "" };
|
||||
public override string action_group { get { return "OTHER_NODES"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
private StreamInteractor stream_interactor;
|
||||
|
||||
public ReceivedMessageListener(StreamInteractor stream_interactor) {
|
||||
this.stream_interactor = stream_interactor;
|
||||
}
|
||||
|
||||
public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
|
||||
if (conversation.type_ != Conversation.Type.GROUPCHAT) return false;
|
||||
XmppStream stream = stream_interactor.get_stream(conversation.account);
|
||||
if (stream == null) return false;
|
||||
if (Xep.DelayedDelivery.MessageFlag.get_flag(message.stanza) == null) {
|
||||
Jid? real_jid = stream.get_flag(Xep.Muc.Flag.IDENTITY).get_real_jid(message.counterpart);
|
||||
if (real_jid != null && !real_jid.equals(message.counterpart)) {
|
||||
message.real_jid = real_jid.bare_jid;
|
||||
}
|
||||
}
|
||||
string? muc_nick = stream.get_flag(Xep.Muc.Flag.IDENTITY).get_muc_nick(conversation.counterpart.bare_jid);
|
||||
if (muc_nick != null && message.from.equals(new Jid(@"$(message.from.bare_jid)/$muc_nick"))) { // TODO better from own
|
||||
Gee.List<Entities.Message> messages = stream_interactor.get_module(MessageStorage.IDENTITY).get_messages(conversation);
|
||||
foreach (Entities.Message m in messages) { // TODO not here
|
||||
if (m.equals(message)) {
|
||||
m.marked = Entities.Message.Marked.RECEIVED;
|
||||
}
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
@ -20,23 +20,39 @@ public class FileProvider : Dino.FileProvider, Object {
|
|||
this.url_regex = new Regex("""^(?i)\b((?:[a-z][\w-]+:(?:\/{1,3}|[a-z0-9%])|www\d{0,3}[.]|[a-z0-9.\-]+[.][a-z]{2,4}\/)(?:[^\s()<>]+|\(([^\s()<>]+|(\([^\s()<>]+\)))*\))+(?:\(([^\s()<>]+|(\([^\s()<>]+\)))*\)|[^\s`!()\[\]{};:'".,<>?«»“”‘’]))$""");
|
||||
this.file_ext_regex = new Regex("""\.(png|jpg|jpeg|svg|gif|pgp)$""");
|
||||
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).message_received.connect(check_in_message);
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).message_sent.connect(check_out_message);
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).received_pipeline.connect(new ReceivedMessageListener(this));
|
||||
stream_interactor.get_module(Manager.IDENTITY).uploaded.connect((file_transfer, url) => {
|
||||
file_transfer.info = url;
|
||||
ignore_once.add(url);
|
||||
});
|
||||
}
|
||||
|
||||
private void check_in_message(Message message, Conversation conversation) {
|
||||
if (!url_regex.match(message.body)) return;
|
||||
private class ReceivedMessageListener : MessageListener {
|
||||
|
||||
public string[] after_actions_const = new string[]{ "" };
|
||||
public override string action_group { get { return "DECRYPT"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
private FileProvider outer;
|
||||
private StreamInteractor stream_interactor;
|
||||
|
||||
public ReceivedMessageListener(FileProvider outer) {
|
||||
this.outer = outer;
|
||||
this.stream_interactor = outer.stream_interactor;
|
||||
}
|
||||
|
||||
public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
|
||||
if (!outer.url_regex.match(message.body)) return false;
|
||||
Jid relevant_jid = stream_interactor.get_module(MucManager.IDENTITY).get_real_jid(message.from, conversation.account) ?? conversation.counterpart;
|
||||
bool in_roster = stream_interactor.get_module(RosterManager.IDENTITY).get_roster_item(conversation.account, relevant_jid) != null;
|
||||
if (message.direction == Message.DIRECTION_RECEIVED && !in_roster) return;
|
||||
if (message.direction == Message.DIRECTION_RECEIVED && !in_roster) return false;
|
||||
|
||||
string? oob_url = Xmpp.Xep.OutOfBandData.get_url_from_message(message.stanza);
|
||||
if ((oob_url != null && oob_url == message.body) || file_ext_regex.match(message.body)) {
|
||||
download_url(message, conversation);
|
||||
if ((oob_url != null && oob_url == message.body) || outer.file_ext_regex.match(message.body)) {
|
||||
yield outer.download_url(message, conversation);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -45,14 +61,15 @@ public class FileProvider : Dino.FileProvider, Object {
|
|||
if (message.body.length < 5) return;
|
||||
if (!url_regex.match(message.body)) return;
|
||||
if (!file_ext_regex.match(message.body)) return;
|
||||
|
||||
download_url(message, conversation);
|
||||
}
|
||||
|
||||
private void download_url(Message message, Conversation conversation) {
|
||||
private async bool download_url(Message message, Conversation conversation) {
|
||||
bool success = false;
|
||||
var session = new Soup.Session();
|
||||
var head_message = new Soup.Message("HEAD", message.body);
|
||||
if (head_message != null) {
|
||||
SourceFunc callback = download_url.callback;
|
||||
session.send_async.begin(head_message, null, (obj, res) => {
|
||||
string? content_type = null, content_length = null;
|
||||
print(message.body + ":\n");
|
||||
|
@ -69,6 +86,7 @@ public class FileProvider : Dino.FileProvider, Object {
|
|||
try {
|
||||
file_transfer.input_stream = request.send_async.end(res);
|
||||
} catch (Error e) {
|
||||
Idle.add((owned)callback);
|
||||
return;
|
||||
}
|
||||
file_transfer.account = conversation.account;
|
||||
|
@ -85,11 +103,19 @@ public class FileProvider : Dino.FileProvider, Object {
|
|||
file_transfer.provider = 0;
|
||||
file_transfer.info = message.body;
|
||||
file_incoming(file_transfer);
|
||||
success = true;
|
||||
Idle.add((owned)callback);
|
||||
});
|
||||
} catch (Error e) { }
|
||||
} catch (Error e) {
|
||||
Idle.add((owned)callback);
|
||||
}
|
||||
} else {
|
||||
Idle.add((owned)callback);
|
||||
}
|
||||
});
|
||||
yield;
|
||||
}
|
||||
return success;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -37,7 +37,8 @@ public class Manager : StreamInteractionModule, FileSender, Object {
|
|||
uploaded(file_transfer, url_down);
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).send_message(url_down, conversation);
|
||||
},
|
||||
() => {
|
||||
(stream, error_str) => {
|
||||
print(@"Failed getting upload url + $error_str\n");
|
||||
file_transfer.state = FileTransfer.State.FAILED;
|
||||
}
|
||||
);
|
||||
|
|
|
@ -11,6 +11,7 @@ public class UploadStreamModule : XmppStreamModule {
|
|||
public static Xmpp.ModuleIdentity<UploadStreamModule> IDENTITY = new Xmpp.ModuleIdentity<UploadStreamModule>(NS_URI, "0363_http_file_upload");
|
||||
|
||||
public signal void feature_available(XmppStream stream, long max_file_size);
|
||||
public signal void received_url(XmppStream stream, MessageStanza message);
|
||||
|
||||
public delegate void OnUploadOk(XmppStream stream, string url_down);
|
||||
public delegate void OnError(XmppStream stream, string error);
|
||||
|
@ -159,6 +160,22 @@ public class UploadStreamModule : XmppStreamModule {
|
|||
}
|
||||
}
|
||||
|
||||
public class ReceivedPipelineListener : StanzaListener<MessageStanza> {
|
||||
|
||||
private const string[] after_actions_const = {"EXTRACT_MESSAGE_2"};
|
||||
|
||||
public override string action_group { get { return "EXTRACT_MESSAGE_2"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
public override async bool run(XmppStream stream, MessageStanza message) {
|
||||
string? oob_url = OutOfBandData.get_url_from_message(message);
|
||||
if (oob_url != null && oob_url == message.body) {
|
||||
stream.get_module(UploadStreamModule.IDENTITY).received_url(stream, message);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
public class Flag : XmppStreamFlag {
|
||||
public static FlagIdentity<Flag> IDENTITY = new FlagIdentity<Flag>(NS_URI, "service_discovery");
|
||||
|
||||
|
|
|
@ -13,6 +13,7 @@ public class Manager : StreamInteractionModule, Object {
|
|||
private StreamInteractor stream_interactor;
|
||||
private Database db;
|
||||
private Map<Entities.Message, MessageState> message_states = new HashMap<Entities.Message, MessageState>(Entities.Message.hash_func, Entities.Message.equals_func);
|
||||
private ReceivedMessageListener received_message_listener = new ReceivedMessageListener();
|
||||
|
||||
private class MessageState {
|
||||
public Entities.Message msg { get; private set; }
|
||||
|
@ -65,15 +66,23 @@ public class Manager : StreamInteractionModule, Object {
|
|||
|
||||
stream_interactor.stream_negotiated.connect(on_stream_negotiated);
|
||||
stream_interactor.account_added.connect(on_account_added);
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).pre_message_received.connect(on_pre_message_received);
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).received_pipeline.connect(received_message_listener);
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).pre_message_send.connect(on_pre_message_send);
|
||||
}
|
||||
|
||||
private void on_pre_message_received(Entities.Message message, Xmpp.MessageStanza message_stanza, Conversation conversation) {
|
||||
MessageFlag? flag = MessageFlag.get_flag(message_stanza);
|
||||
private class ReceivedMessageListener : MessageListener {
|
||||
|
||||
public string[] after_actions_const = new string[]{ "" };
|
||||
public override string action_group { get { return "DECRYPT"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
|
||||
MessageFlag? flag = MessageFlag.get_flag(stanza);
|
||||
if (flag != null && ((!)flag).decrypted) {
|
||||
message.encryption = Encryption.OMEMO;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private void on_pre_message_send(Entities.Message message, Xmpp.MessageStanza message_stanza, Conversation conversation) {
|
||||
|
|
|
@ -397,17 +397,17 @@ public class ReceivedPipelineListener : StanzaListener<MessageStanza> {
|
|||
this.store = store;
|
||||
}
|
||||
|
||||
public override async void run(XmppStream stream, MessageStanza message) {
|
||||
public override async bool run(XmppStream stream, MessageStanza message) {
|
||||
StanzaNode? _encrypted = message.stanza.get_subnode("encrypted", NS_URI);
|
||||
if (_encrypted == null || MessageFlag.get_flag(message) != null || message.from == null) return;
|
||||
if (_encrypted == null || MessageFlag.get_flag(message) != null || message.from == null) return false;
|
||||
StanzaNode encrypted = (!)_encrypted;
|
||||
if (!Plugin.ensure_context()) return;
|
||||
if (!Plugin.ensure_context()) return false;
|
||||
MessageFlag flag = new MessageFlag();
|
||||
message.add_flag(flag);
|
||||
StanzaNode? _header = encrypted.get_subnode("header");
|
||||
if (_header == null) return;
|
||||
if (_header == null) return false;
|
||||
StanzaNode header = (!)_header;
|
||||
if (header.get_attribute_int("sid") <= 0) return;
|
||||
if (header.get_attribute_int("sid") <= 0) return false;
|
||||
foreach (StanzaNode key_node in header.get_subnodes("key")) {
|
||||
if (key_node.get_attribute_int("rid") == store.local_registration_id) {
|
||||
try {
|
||||
|
@ -448,6 +448,7 @@ public class ReceivedPipelineListener : StanzaListener<MessageStanza> {
|
|||
}
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private string arr_to_str(uint8[] arr) {
|
||||
|
|
|
@ -15,6 +15,7 @@ public class Manager : StreamInteractionModule, Object {
|
|||
private StreamInteractor stream_interactor;
|
||||
private Database db;
|
||||
private HashMap<Jid, string> pgp_key_ids = new HashMap<Jid, string>(Jid.hash_bare_func, Jid.equals_bare_func);
|
||||
private ReceivedMessageListener received_message_listener = new ReceivedMessageListener();
|
||||
|
||||
public static void start(StreamInteractor stream_interactor, Database db) {
|
||||
Manager m = new Manager(stream_interactor, db);
|
||||
|
@ -26,7 +27,7 @@ public class Manager : StreamInteractionModule, Object {
|
|||
this.db = db;
|
||||
|
||||
stream_interactor.account_added.connect(on_account_added);
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).pre_message_received.connect(on_pre_message_received);
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).received_pipeline.connect(received_message_listener);
|
||||
stream_interactor.get_module(MessageProcessor.IDENTITY).pre_message_send.connect(check_encypt);
|
||||
}
|
||||
|
||||
|
@ -63,12 +64,6 @@ public class Manager : StreamInteractionModule, Object {
|
|||
return gpgkeys;
|
||||
}
|
||||
|
||||
private void on_pre_message_received(Entities.Message message, Xmpp.MessageStanza message_stanza, Conversation conversation) {
|
||||
if (MessageFlag.get_flag(message_stanza) != null && MessageFlag.get_flag(message_stanza).decrypted) {
|
||||
message.encryption = Encryption.PGP;
|
||||
}
|
||||
}
|
||||
|
||||
private void check_encypt(Entities.Message message, Xmpp.MessageStanza message_stanza, Conversation conversation) {
|
||||
try {
|
||||
if (message.encryption == Encryption.PGP) {
|
||||
|
@ -104,6 +99,20 @@ public class Manager : StreamInteractionModule, Object {
|
|||
pgp_key_ids[jid] = key_id;
|
||||
}
|
||||
}
|
||||
|
||||
private class ReceivedMessageListener : MessageListener {
|
||||
|
||||
public string[] after_actions_const = new string[]{ "" };
|
||||
public override string action_group { get { return "DECRYPT"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
|
||||
if (MessageFlag.get_flag(stanza) != null && MessageFlag.get_flag(stanza).decrypted) {
|
||||
message.encryption = Encryption.PGP;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
@ -140,7 +140,7 @@ public class ReceivedPipelineDecryptListener : StanzaListener<MessageStanza> {
|
|||
public override string action_group { get { return "ENCRYPT_BODY"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
public override async void run(XmppStream stream, MessageStanza message) {
|
||||
public override async bool run(XmppStream stream, MessageStanza message) {
|
||||
string? encrypted = get_cyphertext(message);
|
||||
if (encrypted != null) {
|
||||
MessageFlag flag = new MessageFlag();
|
||||
|
@ -151,6 +151,7 @@ public class ReceivedPipelineDecryptListener : StanzaListener<MessageStanza> {
|
|||
message.body = decrypted;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private static async string? gpg_decrypt(string enc) {
|
||||
|
|
|
@ -11,7 +11,6 @@ namespace Xmpp {
|
|||
public StanzaListenerHolder<MessageStanza> received_pipeline = new StanzaListenerHolder<MessageStanza>();
|
||||
public StanzaListenerHolder<MessageStanza> send_pipeline = new StanzaListenerHolder<MessageStanza>();
|
||||
|
||||
public signal void pre_received_message(XmppStream stream, MessageStanza message);
|
||||
public signal void received_message(XmppStream stream, MessageStanza message);
|
||||
|
||||
public void send_message(XmppStream stream, MessageStanza message) {
|
||||
|
|
|
@ -12,34 +12,42 @@ public string random_uuid() {
|
|||
return "%08x-%04x-%04x-%04x-%04x%08x".printf(b1, b2, b3, b4, b5_1, b5_2);
|
||||
}
|
||||
|
||||
public abstract class StanzaListener<T> : Object {
|
||||
public abstract string action_group { get; }
|
||||
public abstract string[] after_actions { get; }
|
||||
public abstract class StanzaListener<T> : OrderedListener {
|
||||
|
||||
public abstract async void run(XmppStream stream, T stanza);
|
||||
public abstract async bool run(XmppStream stream, T stanza);
|
||||
}
|
||||
|
||||
public class StanzaListenerHolder<T> : Object {
|
||||
private ArrayList<StanzaListener<T>> listeners = new ArrayList<StanzaListener<T>>();
|
||||
public class StanzaListenerHolder<T> : ListenerHolder {
|
||||
|
||||
public new void connect(StanzaListener<T> listener) {
|
||||
public async void run(XmppStream stream, T stanza) {
|
||||
foreach (OrderedListener ol in listeners) {
|
||||
StanzaListener<T> l = ol as StanzaListener<T>;
|
||||
bool stop = yield l.run(stream, stanza);
|
||||
if (stop) break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public abstract class OrderedListener : Object {
|
||||
public abstract string action_group { get; }
|
||||
public abstract string[] after_actions { get; }
|
||||
}
|
||||
|
||||
public abstract class ListenerHolder : Object {
|
||||
protected ArrayList<OrderedListener> listeners = new ArrayList<OrderedListener>();
|
||||
|
||||
public new void connect(OrderedListener listener) {
|
||||
listeners.add(listener);
|
||||
resort_list();
|
||||
}
|
||||
|
||||
public new void disconnect(StanzaListener<T> listener) {
|
||||
public new void disconnect(OrderedListener listener) {
|
||||
listeners.remove(listener);
|
||||
resort_list();
|
||||
}
|
||||
|
||||
public async void run(XmppStream stream, T stanza) {
|
||||
foreach (StanzaListener<T> l in listeners) {
|
||||
yield l.run(stream, stanza);
|
||||
}
|
||||
}
|
||||
|
||||
private bool set_contains_action(Gee.List<StanzaListener<T>> s, string[] actions) {
|
||||
foreach (StanzaListener<T> l in s) {
|
||||
private bool set_contains_action(Gee.List<OrderedListener> s, string[] actions) {
|
||||
foreach(OrderedListener l in s) {
|
||||
if (l.action_group in actions) {
|
||||
return true;
|
||||
}
|
||||
|
@ -48,22 +56,23 @@ public class StanzaListenerHolder<T> : Object {
|
|||
}
|
||||
|
||||
private void resort_list() {
|
||||
ArrayList<StanzaListener<T>> new_list = new ArrayList<StanzaListener<T>>();
|
||||
ArrayList<StanzaListener<T>> remaining = new ArrayList<StanzaListener<T>>();
|
||||
ArrayList<OrderedListener> new_list = new ArrayList<OrderedListener>();
|
||||
ArrayList<OrderedListener> remaining = new ArrayList<OrderedListener>();
|
||||
remaining.add_all(listeners);
|
||||
while (remaining.size > 0) {
|
||||
bool changed = false;
|
||||
Gee.Iterator<StanzaListener<T>> iter = remaining.iterator();
|
||||
Gee.Iterator<OrderedListener> iter = remaining.iterator();
|
||||
while (iter.has_next()) {
|
||||
if (!iter.valid) {
|
||||
iter.next();
|
||||
}
|
||||
StanzaListener<T> l = iter.get();
|
||||
OrderedListener l = iter.get();
|
||||
if (!set_contains_action(remaining, l.after_actions)) {
|
||||
new_list.add(l);
|
||||
iter.remove();
|
||||
changed = true;
|
||||
}
|
||||
iter.next();
|
||||
}
|
||||
if (!changed) error("Can't sort listeners");
|
||||
}
|
||||
|
|
|
@ -63,10 +63,11 @@ public class SendPipelineListener : StanzaListener<MessageStanza> {
|
|||
public override string action_group { get { return "ADD_NODES"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
public override async void run(XmppStream stream, MessageStanza message) {
|
||||
if (message.body == null) return;
|
||||
if (message.type_ != MessageStanza.TYPE_CHAT) return;
|
||||
public override async bool run(XmppStream stream, MessageStanza message) {
|
||||
if (message.body == null) return false;
|
||||
if (message.type_ != MessageStanza.TYPE_CHAT) return false;
|
||||
message.stanza.put_node(new StanzaNode.build(STATE_ACTIVE, NS_URI).add_self_xmlns());
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -48,12 +48,13 @@ public class SendPipelineListener : StanzaListener<MessageStanza> {
|
|||
public override string action_group { get { return "ADD_NODES"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
public override async void run(XmppStream stream, MessageStanza message) {
|
||||
public override async bool run(XmppStream stream, MessageStanza message) {
|
||||
StanzaNode? received_node = message.stanza.get_subnode("received", NS_URI);
|
||||
if (received_node != null) return;
|
||||
if (message.body == null) return;
|
||||
if (message.type_ == MessageStanza.TYPE_GROUPCHAT) return;
|
||||
if (received_node != null) return false;
|
||||
if (message.body == null) return false;
|
||||
if (message.type_ == MessageStanza.TYPE_GROUPCHAT) return false;
|
||||
message.stanza.put_node(new StanzaNode.build("request", NS_URI).add_self_xmlns());
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -51,9 +51,10 @@ public class ReceivedPipelineListener : StanzaListener<MessageStanza> {
|
|||
public override string action_group { get { return "ADD_NODE"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
public override async void run(XmppStream stream, MessageStanza message) {
|
||||
public override async bool run(XmppStream stream, MessageStanza message) {
|
||||
DateTime? datetime = Module.get_time_for_message(message);
|
||||
if (datetime != null) message.add_flag(new MessageFlag(datetime));
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -44,7 +44,7 @@ public class ReceivedPipelineListener : StanzaListener<MessageStanza> {
|
|||
public override string action_group { get { return "EXTRACT_MESSAGE_2"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
public override async void run(XmppStream stream, MessageStanza message) {
|
||||
public override async bool run(XmppStream stream, MessageStanza message) {
|
||||
StanzaNode? received_node = message.stanza.get_subnode("received", NS_URI);
|
||||
StanzaNode? sent_node = received_node == null ? message.stanza.get_subnode("sent", NS_URI) : null;
|
||||
StanzaNode? carbons_node = received_node != null ? received_node : sent_node;
|
||||
|
@ -55,6 +55,8 @@ public class ReceivedPipelineListener : StanzaListener<MessageStanza> {
|
|||
string? from_attribute = message_node.get_attribute("from", Xmpp.NS_URI);
|
||||
// Any forwarded copies received by a Carbons-enabled client MUST be from that user's bare JID; any copies that do not meet this requirement MUST be ignored.
|
||||
if (from_attribute != null && from_attribute == stream.get_flag(Bind.Flag.IDENTITY).my_jid.bare_jid.to_string()) {
|
||||
return true;
|
||||
}
|
||||
if (received_node != null) {
|
||||
message.add_flag(new MessageFlag(MessageFlag.TYPE_RECEIVED));
|
||||
} else if (sent_node != null) {
|
||||
|
@ -63,10 +65,8 @@ public class ReceivedPipelineListener : StanzaListener<MessageStanza> {
|
|||
message.stanza = message_node;
|
||||
message.rerun_parsing = true;
|
||||
}
|
||||
message.stanza = message_node;
|
||||
message.rerun_parsing = true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -91,9 +91,9 @@ public class ReceivedPipelineListener : StanzaListener<MessageStanza> {
|
|||
public override string action_group { get { return "EXTRACT_MESSAGE_1"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
public override async void run(XmppStream stream, MessageStanza message) {
|
||||
public override async bool run(XmppStream stream, MessageStanza message) {
|
||||
// if (message.from != stream.remote_name) return;
|
||||
if (stream.get_flag(Flag.IDENTITY) == null) return;
|
||||
if (stream.get_flag(Flag.IDENTITY) == null) return false;
|
||||
|
||||
StanzaNode? message_node = message.stanza.get_deep_subnode(NS_VER(stream) + ":result", "urn:xmpp:forward:0:forwarded", Xmpp.NS_URI + ":message");
|
||||
if (message_node != null) {
|
||||
|
@ -104,6 +104,7 @@ public class ReceivedPipelineListener : StanzaListener<MessageStanza> {
|
|||
message.stanza = message_node;
|
||||
message.rerun_parsing = true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -61,12 +61,13 @@ public class SendPipelineListener : StanzaListener<MessageStanza> {
|
|||
public override string action_group { get { return "ADD_NODES"; } }
|
||||
public override string[] after_actions { get { return after_actions_const; } }
|
||||
|
||||
public override async void run(XmppStream stream, MessageStanza message) {
|
||||
public override async bool run(XmppStream stream, MessageStanza message) {
|
||||
StanzaNode? received_node = message.stanza.get_subnode("received", NS_URI);
|
||||
if (received_node != null) return;
|
||||
if (message.body == null) return;
|
||||
if (message.type_ != MessageStanza.TYPE_CHAT) return;
|
||||
if (received_node != null) return false;
|
||||
if (message.body == null) return false;
|
||||
if (message.type_ != MessageStanza.TYPE_CHAT) return false;
|
||||
message.stanza.put_node(new StanzaNode.build("markable", NS_URI).add_self_xmlns());
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
Loading…
Reference in a new issue