aboutsummaryrefslogtreecommitdiff
path: root/xmpp-vala
diff options
context:
space:
mode:
authorfiaxh <git@lightrise.org>2020-07-21 15:48:42 +0200
committerfiaxh <git@lightrise.org>2020-07-21 15:48:42 +0200
commitc887240fdcdbc8f76c6bdd37b1174b8154907c4c (patch)
tree889151d1947482a7da968c076b7bf8e09da91e8b /xmpp-vala
parent7309c6f3ac4e580f197b4835d05bb011fa90780b (diff)
downloaddino-c887240fdcdbc8f76c6bdd37b1174b8154907c4c.tar.gz
dino-c887240fdcdbc8f76c6bdd37b1174b8154907c4c.zip
Improve stream management queue
Diffstat (limited to 'xmpp-vala')
-rw-r--r--xmpp-vala/src/core/stanza_writer.vala3
-rw-r--r--xmpp-vala/src/core/xmpp_stream.vala6
-rw-r--r--xmpp-vala/src/module/xep/0198_stream_management.vala34
3 files changed, 22 insertions, 21 deletions
diff --git a/xmpp-vala/src/core/stanza_writer.vala b/xmpp-vala/src/core/stanza_writer.vala
index 51f0061f..62fe022a 100644
--- a/xmpp-vala/src/core/stanza_writer.vala
+++ b/xmpp-vala/src/core/stanza_writer.vala
@@ -1,6 +1,8 @@
namespace Xmpp {
public class StanzaWriter {
+ public signal void cancel();
+
private OutputStream output;
private Queue<SourceFuncWrapper> queue = new Queue<SourceFuncWrapper>();
@@ -43,6 +45,7 @@ public class StanzaWriter {
try {
yield output.write_all_async(data, 0, null, null);
} catch (GLib.Error e) {
+ cancel();
throw new XmlError.IO(@"IOError in GLib: $(e.message)");
} finally {
SourceFuncWrapper? sfw = queue.pop_head();
diff --git a/xmpp-vala/src/core/xmpp_stream.vala b/xmpp-vala/src/core/xmpp_stream.vala
index 7e7588b1..ad4dae97 100644
--- a/xmpp-vala/src/core/xmpp_stream.vala
+++ b/xmpp-vala/src/core/xmpp_stream.vala
@@ -97,6 +97,8 @@ public class XmppStream {
this.stream = stream;
reader = new StanzaReader.for_stream(stream.input_stream);
writer = new StanzaWriter.for_stream(stream.output_stream);
+
+ writer.cancel.connect(reader.cancel);
require_setup();
}
@@ -123,7 +125,9 @@ public class XmppStream {
[Version (deprecated = true, deprecated_since = "0.1", replacement = "write_async")]
public void write(StanzaNode node) {
write_async.begin(node, (obj, res) => {
- write_async.end(res);
+ try {
+ write_async.end(res);
+ } catch (Error e) { }
});
}
diff --git a/xmpp-vala/src/module/xep/0198_stream_management.vala b/xmpp-vala/src/module/xep/0198_stream_management.vala
index bc695ddb..b6317425 100644
--- a/xmpp-vala/src/module/xep/0198_stream_management.vala
+++ b/xmpp-vala/src/module/xep/0198_stream_management.vala
@@ -26,25 +26,21 @@ public class Module : XmppStreamNegotiationModule, WriteNodeFunc {
}
public async void write_stanza(XmppStream stream, StanzaNode node) throws IOStreamError {
- if (stream.has_flag(Flag.IDENTITY)) {
- var promise = new Promise<IOError?>();
+ var promise = new Promise<IOError?>();
- node_queue.add(new QueueItem(node, promise));
- check_queue(stream);
+ node_queue.add(new QueueItem(node, promise));
+ check_queue(stream);
- try {
- yield promise.future.wait_async();
- } catch (FutureError e) {
- throw new IOStreamError.WRITE("Future returned error %i".printf(e.code));
- }
- } else {
- yield write_node(stream, node);
+ try {
+ yield promise.future.wait_async();
+ } catch (FutureError e) {
+ throw new IOStreamError.WRITE("Future returned error %i".printf(e.code));
}
}
- internal async void write_node(XmppStream stream, StanzaNode node) throws IOError {
+ internal async void write_node(XmppStream stream, StanzaNode node) {
StanzaWriter? writer = stream.writer;
- if (writer == null) throw new IOStreamError.WRITE("trying to write, but no stream open");
+ if (writer == null) return;
try {
stream.log.node("OUT", node, stream);
if (node.name == "message" || node.name == "iq" || node.name == "presence") {
@@ -54,12 +50,10 @@ public class Module : XmppStreamNegotiationModule, WriteNodeFunc {
} else {
yield ((!)writer).write_node(node);
}
- } catch (XmlError e) {
- throw new IOStreamError.WRITE(e.message);
- }
+ } catch (XmlError e) { }
}
- private void check_queue(XmppStream stream) throws IOError {
+ private void check_queue(XmppStream stream) {
while (!node_queue.is_empty && in_flight_stanzas.size < 10) {
QueueItem queue_item = node_queue.remove_at(0);
StanzaNode node = queue_item.node;
@@ -115,7 +109,7 @@ public class Module : XmppStreamNegotiationModule, WriteNodeFunc {
StanzaNode node = new StanzaNode.build("resume", NS_URI).add_self_xmlns()
.put_attribute("h", h_inbound.to_string())
.put_attribute("previd", session_id);
- write_node(stream, node);
+ write_node.begin(stream, node);
stream.add_flag(new Flag());
}
}
@@ -124,7 +118,7 @@ public class Module : XmppStreamNegotiationModule, WriteNodeFunc {
private void check_enable(XmppStream stream) {
if (stream_has_sm_feature(stream) && session_id == null) {
StanzaNode node = new StanzaNode.build("enable", NS_URI).add_self_xmlns().put_attribute("resume", "true");
- write_node(stream, node);
+ write_node.begin(stream, node);
stream.add_flag(new Flag());
h_outbound = 0;
}
@@ -174,7 +168,7 @@ public class Module : XmppStreamNegotiationModule, WriteNodeFunc {
private void send_ack(XmppStream stream) {
StanzaNode node = new StanzaNode.build("a", NS_URI).add_self_xmlns().put_attribute("h", h_inbound.to_string());
- write_node(stream, node);
+ write_node.begin(stream, node);
}
private void handle_ack(XmppStream stream, StanzaNode node) {