Read+(write) stream async

3 files changed, 78 insertions, 71 deletions

diff --git a/xmpp-vala/src/core/stanza_reader.vala b/xmpp-vala/src/core/stanza_reader.vala
index f4b900d1..4b4d98ab 100644
--- a/xmpp-vala/src/core/stanza_reader.vala
+++ b/xmpp-vala/src/core/stanza_reader.vala
@@ -44,12 +44,12 @@ public class StanzaReader {
         cancellable.cancel();
     }
 
-    private void update_buffer() throws XmlError {
+    private async void update_buffer() throws XmlError {
         try {
             InputStream? input = this.input;
             if (input == null) throw new XmlError.EOF("No input stream specified and end of buffer reached.");
             if (cancellable.is_cancelled()) throw new XmlError.EOF("Input stream is canceled.");
-            buffer_fill = (int) ((!)input).read(buffer, cancellable);
+            buffer_fill = (int) yield ((!)input).read_async(buffer, GLib.Priority.DEFAULT, cancellable);
             if (buffer_fill == 0) throw new XmlError.EOF("End of input stream reached.");
             buffer_pos = 0;
         } catch (GLib.IOError e) {
@@ -57,15 +57,16 @@ public class StanzaReader {
         }
     }
 
-    private char read_single() throws XmlError {
+    private async char read_single() throws XmlError {
         if (buffer_pos >= buffer_fill) {
-            update_buffer();
+            yield update_buffer();
         }
-        return (char) buffer[buffer_pos++];
+        char c = (char) buffer[buffer_pos++];
+        return c;
     }
 
-    private char peek_single() throws XmlError {
-        var res = read_single();
+    private async char peek_single() throws XmlError {
+        var res = yield read_single();
         buffer_pos--;
         return res;
     }
@@ -78,53 +79,53 @@ public class StanzaReader {
         buffer_pos++;
     }
 
-    private void skip_until_non_ws() throws XmlError {
-        while (is_ws(peek_single())) {
+    private async void skip_until_non_ws() throws XmlError {
+        while (is_ws(yield peek_single())) {
             skip_single();
         }
     }
 
-    private string read_until_ws() throws XmlError {
+    private async string read_until_ws() throws XmlError {
         var res = new StringBuilder();
-        var what = peek_single();
+        var what = yield peek_single();
         while (!is_ws(what)) {
-            res.append_c(read_single());
-            what = peek_single();
+            res.append_c(yield read_single());
+            what = yield peek_single();
         }
         return res.str;
     }
 
-    private string read_until_char_or_ws(char x, char y = 0) throws XmlError {
+    private async string read_until_char_or_ws(char x, char y = 0) throws XmlError {
         var res = new StringBuilder();
-        var what = peek_single();
+        var what = yield peek_single();
         while (what != x && what != y && !is_ws(what)) {
-            res.append_c(read_single());
-            what = peek_single();
+            res.append_c(yield read_single());
+            what = yield peek_single();
         }
         return res.str;
     }
 
-    private string read_until_char(char x) throws XmlError {
+    private async string read_until_char(char x) throws XmlError {
         var res = new StringBuilder();
-        var what = peek_single();
+        var what = yield peek_single();
         while (what != x) {
-            res.append_c(read_single());
-            what = peek_single();
+            res.append_c(yield read_single());
+            what = yield peek_single();
         }
         return res.str;
     }
 
-    private StanzaAttribute read_attribute() throws XmlError {
+    private async StanzaAttribute read_attribute() throws XmlError {
         var res = new StanzaAttribute();
-        res.name = read_until_char_or_ws('=');
-        if (read_single() == '=') {
-            var quot = peek_single();
+        res.name = yield read_until_char_or_ws('=');
+        if ((yield read_single()) == '=') {
+            var quot = yield peek_single();
             if (quot == '\'' || quot == '"') {
                 skip_single();
-                res.encoded_val = read_until_char(quot);
+                res.encoded_val = yield read_until_char(quot);
                 skip_single();
             } else {
-                res.encoded_val = read_until_ws();
+                res.encoded_val = yield read_until_ws();
             }
         }
         return res;
@@ -161,17 +162,17 @@ public class StanzaReader {
         }
     }
 
-    public StanzaNode read_node_start() throws XmlError {
+    public async StanzaNode read_node_start() throws XmlError {
         var res = new StanzaNode();
         res.attributes = new ArrayList<StanzaAttribute>();
         var eof = false;
-        if (peek_single() == '<') skip_single();
-        if (peek_single() == '?') res.pseudo = true;
-        if (peek_single() == '/') {
+        if ((yield peek_single()) == '<') skip_single();
+        if ((yield peek_single()) == '?') res.pseudo = true;
+        if ((yield peek_single()) == '/') {
             eof = true;
             skip_single();
-            res.name = read_until_char_or_ws('>');
-            while (peek_single() != '>') {
+            res.name = yield read_until_char_or_ws('>');
+            while ((yield peek_single()) != '>') {
                 skip_single();
             }
             skip_single();
@@ -180,13 +181,15 @@ public class StanzaReader {
             handle_stanza_ns(res);
             return res;
         }
-        res.name = read_until_char_or_ws('>', '/');
-        skip_until_non_ws();
-        while (peek_single() != '/' && peek_single() != '>' && peek_single() != '?') {
-            res.attributes.add(read_attribute());
-            skip_until_non_ws();
+        res.name = yield read_until_char_or_ws('>', '/');
+        yield skip_until_non_ws();
+        char next_char = yield peek_single();
+        while (next_char != '/' && next_char != '>' && next_char != '?') {
+            res.attributes.add(yield read_attribute());
+            yield skip_until_non_ws();
+            next_char = yield peek_single();
         }
-        if (read_single() == '/' || res.pseudo) {
+        if ((yield read_single()) == '/' || res.pseudo) {
             res.has_nodes = false;
             skip_single();
         } else {
@@ -196,20 +199,20 @@ public class StanzaReader {
         return res;
     }
 
-    public StanzaNode read_text_node() throws XmlError {
+    public async StanzaNode read_text_node() throws XmlError {
         var res = new StanzaNode();
         res.name = "#text";
         res.ns_uri = ns_state.current_ns_uri;
-        res.encoded_val = read_until_char('<').strip();
+        res.encoded_val = (yield read_until_char('<')).strip();
         return res;
     }
 
-    public StanzaNode read_root_node() throws XmlError {
-        skip_until_non_ws();
-        if (peek_single() == '<') {
-            var res = read_node_start();
+    public async StanzaNode read_root_node() throws XmlError {
+        yield skip_until_non_ws();
+        if ((yield peek_single()) == '<') {
+            var res = yield read_node_start();
             if (res.pseudo) {
-                return read_root_node();
+                return yield read_root_node();
             }
             return res;
         } else {
@@ -217,18 +220,18 @@ public class StanzaReader {
         }
     }
 
-    public StanzaNode read_stanza_node() throws XmlError {
+    public async StanzaNode read_stanza_node() throws XmlError {
         ns_state = ns_state.push();
-        var res = read_node_start();
+        var res = yield read_node_start();
         if (res.has_nodes) {
             bool finishNodeSeen = false;
             do {
-                skip_until_non_ws();
-                if (peek_single() == '<') {
+                yield skip_until_non_ws();
+                if ((yield peek_single()) == '<') {
                     skip_single();
-                    if (peek_single() == '/') {
+                    if ((yield peek_single()) == '/') {
                         skip_single();
-                        string desc = read_until_char('>');
+                        string desc = yield read_until_char('>');
                         skip_single();
                         if (desc.contains(":")) {
                             var split = desc.split(":");
@@ -240,10 +243,10 @@ public class StanzaReader {
                         }
                         finishNodeSeen = true;
                     } else {
-                        res.sub_nodes.add(read_stanza_node());
+                        res.sub_nodes.add(yield read_stanza_node());
                     }
                 } else {
-                    res.sub_nodes.add(read_text_node());
+                    res.sub_nodes.add(yield read_text_node());
                 }
             } while (!finishNodeSeen);
             if (res.sub_nodes.size == 0) res.has_nodes = false;
@@ -252,12 +255,12 @@ public class StanzaReader {
         return res;
     }
 
-    public StanzaNode read_node() throws XmlError {
-        skip_until_non_ws();
-        if (peek_single() == '<') {
-            return read_stanza_node();
+    public async StanzaNode read_node() throws XmlError {
+        yield skip_until_non_ws();
+        if ((yield peek_single()) == '<') {
+            return yield read_stanza_node();
         } else {
-            return read_text_node();
+            return yield read_text_node();
         }
     }
 }
diff --git a/xmpp-vala/src/core/stanza_writer.vala b/xmpp-vala/src/core/stanza_writer.vala
index 26524d7b..e67920db 100644
--- a/xmpp-vala/src/core/stanza_writer.vala
+++ b/xmpp-vala/src/core/stanza_writer.vala
@@ -24,4 +24,4 @@ public class StanzaWriter {
         }
     }
 }
-}
\ No newline at end of file
+}
diff --git a/xmpp-vala/src/core/xmpp_stream.vala b/xmpp-vala/src/core/xmpp_stream.vala
index 0a1f4120..fc4e7fd7 100644
--- a/xmpp-vala/src/core/xmpp_stream.vala
+++ b/xmpp-vala/src/core/xmpp_stream.vala
@@ -42,7 +42,7 @@ public class XmppStream {
         register_connection_provider(new StartTlsConnectionProvider());
     }
 
-    public void connect(string? remote_name = null) throws IOStreamError {
+    public async void connect(string? remote_name = null) throws IOStreamError {
         if (remote_name != null) this.remote_name = (!)remote_name;
         attach_negotation_modules();
         try {
@@ -67,7 +67,7 @@ public class XmppStream {
             stderr.printf("CONNECTION LOST?\n");
             throw new IOStreamError.CONNECT(e.message);
         }
-        loop();
+        yield loop();
     }
 
     public void disconnect() throws IOStreamError {
@@ -96,11 +96,11 @@ public class XmppStream {
         return setup_needed;
     }
 
-    public StanzaNode read() throws IOStreamError {
+    public async StanzaNode read() throws IOStreamError {
         StanzaReader? reader = this.reader;
         if (reader == null) throw new IOStreamError.READ("trying to read, but no stream open");
         try {
-            StanzaNode node = ((!)reader).read_node();
+            StanzaNode node = yield ((!)reader).read_node();
             log.node("IN", node);
             return node;
         } catch (XmlError e) {
@@ -175,7 +175,7 @@ public class XmppStream {
         return false;
     }
 
-    private void setup() throws IOStreamError {
+    private async void setup() throws IOStreamError {
         StanzaNode outs = new StanzaNode.build("stream", "http://etherx.jabber.org/streams")
                             .put_attribute("to", remote_name)
                             .put_attribute("version", "1.0")
@@ -184,17 +184,21 @@ public class XmppStream {
         outs.has_nodes = true;
         log.node("OUT ROOT", outs);
         write(outs);
-        received_root_node(this, read_root());
+        received_root_node(this, yield read_root());
     }
 
-    private void loop() throws IOStreamError {
+    private async void loop() throws IOStreamError {
         while (true) {
             if (setup_needed) {
-                setup();
+                yield setup();
                 setup_needed = false;
             }
 
-            StanzaNode node = read();
+            StanzaNode node = yield read();
+
+            Idle.add(loop.callback);
+            yield;
+
             received_node(this, node);
 
             if (node.ns_uri == NS_URI && node.name == "features") {
@@ -266,11 +270,11 @@ public class XmppStream {
         }
     }
 
-    private StanzaNode read_root() throws IOStreamError {
+    private async StanzaNode read_root() throws IOStreamError {
         StanzaReader? reader = this.reader;
         if (reader == null) throw new IOStreamError.READ("trying to read, but no stream open");
         try {
-            StanzaNode node = ((!)reader).read_root_node();
+            StanzaNode node = yield ((!)reader).read_root_node();
             log.node("IN ROOT", node);
             return node;
         } catch (XmlError e) {