Read+(write) stream async

1 files changed, 10 insertions, 7 deletions

diff --git a/xmpp-vala/src/module/message/module.vala b/xmpp-vala/src/module/message/module.vala
index f7038ef8..2ca06dc4 100644
--- a/xmpp-vala/src/module/message/module.vala
+++ b/xmpp-vala/src/module/message/module.vala
@@ -8,24 +8,27 @@ namespace Xmpp.Message {
     public class Module : XmppStreamModule {
         public static ModuleIdentity<Module> IDENTITY = new ModuleIdentity<Module>(NS_URI, "message_module");
 
-        public signal void pre_send_message(XmppStream stream, Message.Stanza message);
+        public StanzaListenerHolder<Message.Stanza> received_pipeline = new StanzaListenerHolder<Message.Stanza>();
+        public StanzaListenerHolder<Message.Stanza> send_pipeline = new StanzaListenerHolder<Message.Stanza>();
+
         public signal void pre_received_message(XmppStream stream, Message.Stanza message);
         public signal void received_message(XmppStream stream, Message.Stanza message);
 
         public void send_message(XmppStream stream, Message.Stanza message) {
-            pre_send_message(stream, message);
+            send_pipeline.run.begin(stream, message);
             stream.write(message.stanza);
         }
 
-        public void received_message_stanza(XmppStream stream, StanzaNode node) {
+        public async void received_message_stanza_async(XmppStream stream, StanzaNode node) {
             Message.Stanza message = new Message.Stanza.from_stanza(node, stream.get_flag(Bind.Flag.IDENTITY).my_jid);
-            do {
-                message.rerun_parsing = false;
-                pre_received_message(stream, message);
-            } while(message.rerun_parsing);
+            yield received_pipeline.run(stream, message);
             received_message(stream, message);
         }
 
+        private void received_message_stanza(XmppStream stream, StanzaNode node) {
+            received_message_stanza_async.begin(stream, node);
+        }
+
         public override void attach(XmppStream stream) {
             stream.received_message_stanza.connect(received_message_stanza);
         }