Add queue and resending to stream management

2 files changed, 35 insertions, 9 deletions

diff --git a/xmpp-vala/src/core/stanza_writer.vala b/xmpp-vala/src/core/stanza_writer.vala
index ab51a948..51f0061f 100644
--- a/xmpp-vala/src/core/stanza_writer.vala
+++ b/xmpp-vala/src/core/stanza_writer.vala
@@ -14,6 +14,22 @@ public class StanzaWriter {
         yield write_data(node.to_xml().data);
     }
 
+    public async void write_nodes(StanzaNode node1, StanzaNode node2) throws XmlError {
+        var data1 = node1.to_xml().data;
+        var data2 = node2.to_xml().data;
+
+        uint8[] concat = new uint8[data1.length + data2.length];
+        int i = 0;
+        foreach (var datum in data1) {
+            concat[i++] = datum;
+        }
+        foreach (var datum in data2) {
+            concat[i++] = datum;
+        }
+
+        yield write_data(concat);
+    }
+
     public async void write(string s) throws XmlError {
         yield write_data(s.data);
     }
diff --git a/xmpp-vala/src/core/xmpp_stream.vala b/xmpp-vala/src/core/xmpp_stream.vala
index 39754ba1..f4cdf09a 100644
--- a/xmpp-vala/src/core/xmpp_stream.vala
+++ b/xmpp-vala/src/core/xmpp_stream.vala
@@ -18,12 +18,14 @@ public class XmppStream {
     public StanzaNode? features { get; private set; default = new StanzaNode.build("features", NS_URI); }
 
     private IOStream? stream;
-    private StanzaReader? reader;
-    private StanzaWriter? writer;
+    internal StanzaReader? reader;
+    internal StanzaWriter? writer;
 
     public Gee.List<XmppStreamFlag> flags { get; private set; default=new ArrayList<XmppStreamFlag>(); }
     public Gee.List<XmppStreamModule> modules { get; private set; default=new ArrayList<XmppStreamModule>(); }
     private Gee.List<ConnectionProvider> connection_providers = new ArrayList<ConnectionProvider>();
+
+    internal WriteNodeFunc? write_obj = null;
     public bool negotiation_complete { get; set; default=false; }
     private bool setup_needed = false;
     private bool non_negotiation_modules_attached = false;
@@ -126,13 +128,17 @@ public class XmppStream {
     }
 
     public async void write_async(StanzaNode node) throws IOStreamError {
-        StanzaWriter? writer = this.writer;
-        if (writer == null) throw new IOStreamError.WRITE("trying to write, but no stream open");
-        try {
-            log.node("OUT", node, this);
-            yield ((!)writer).write_node(node);
-        } catch (XmlError e) {
-            throw new IOStreamError.WRITE(e.message);
+        if (write_obj != null) {
+            yield write_obj.write_stanza(this, node);
+        } else {
+            StanzaWriter? writer = this.writer;
+            if (writer == null) throw new IOStreamError.WRITE("trying to write, but no stream open");
+            try {
+                log.node("OUT", node, this);
+                yield ((!)writer).write_node(node);
+            } catch (XmlError e) {
+                throw new IOStreamError.WRITE(e.message);
+            }
         }
     }
 
@@ -403,4 +409,8 @@ public class StartTlsConnectionProvider : ConnectionProvider {
     public override string get_id() { return "start_tls"; }
 }
 
+public interface WriteNodeFunc : Object {
+    public abstract async void write_stanza(XmppStream stream, StanzaNode node) throws IOError;
+}
+
 }