diff options
Diffstat (limited to 'xmpp-vala/src/module/xep/0198_stream_management.vala')
| -rw-r--r-- | xmpp-vala/src/module/xep/0198_stream_management.vala | 157 |
1 files changed, 116 insertions, 41 deletions
diff --git a/xmpp-vala/src/module/xep/0198_stream_management.vala b/xmpp-vala/src/module/xep/0198_stream_management.vala index 68eee8ae..6b0d5162 100644 --- a/xmpp-vala/src/module/xep/0198_stream_management.vala +++ b/xmpp-vala/src/module/xep/0198_stream_management.vala @@ -128,20 +128,23 @@ public class Module : XmppStreamNegotiationModule, WriteNodeFunc { } private void check_resume(XmppStream stream) { - if (stream_has_sm_feature(stream) && session_id != null) { + if (stream_has_sm_feature(stream) && session_id != null && !Flag.get_add_flag(stream).finished) { 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.begin(stream, node); - stream.add_flag(new Flag()); + + Flag.get_add_flag(stream); } } private void check_enable(XmppStream stream) { - if (stream_has_sm_feature(stream) && session_id == null) { + if (stream_has_sm_feature(stream) && session_id == null && !Flag.get_add_flag(stream).finished) { StanzaNode node = new StanzaNode.build("enable", NS_URI).add_self_xmlns().put_attribute("resume", "true"); write_node.begin(stream, node); - stream.add_flag(new Flag()); + + Flag.get_add_flag(stream); + h_outbound = 0; } } @@ -153,46 +156,64 @@ public class Module : XmppStreamNegotiationModule, WriteNodeFunc { } else if (node.name == "a") { handle_ack(stream, node); } else if (node.name in new string[]{"enabled", "resumed", "failed"}) { - stream.get_flag(Flag.IDENTITY).finished = true; - if (node.name == "enabled") { - h_inbound = 0; - session_id = node.get_attribute("id", NS_URI); - flags = stream.flags; - ((IoXmppStream)stream).write_obj = this; + on_enabled(stream, node); } else if (node.name == "resumed") { - stream.get_flag(Flag.IDENTITY).resumed = true; - - foreach (XmppStreamFlag flag in flags) { - stream.add_flag(flag); - } - - h_outbound = int.parse(node.get_attribute("h", NS_URI)); - handle_incoming_h(stream, h_outbound); - foreach (var id in in_flight_stanzas.keys) { - node_queue.add(in_flight_stanzas[id]); - } - in_flight_stanzas.clear(); - check_queue(stream); - ((IoXmppStream)stream).write_obj = this; + on_resumed(stream, node); } else if (node.name == "failed") { - session_id = null; - string? h_acked = node.get_attribute("h", NS_URI); - if (h_acked != null) { - h_outbound = int.parse(h_acked); - handle_incoming_h(stream, h_outbound); - } - foreach (var id in in_flight_stanzas.keys) { - in_flight_stanzas[id].promise.set_exception(new IOError.FAILED("Stanza not acked and session not resumed")); - } - in_flight_stanzas.clear(); - check_queue(stream); - stream.received_features_node(stream); + on_resume_failed(stream, node); } } } } + public void on_resumed(XmppStream stream, StanzaNode resumed_node) { + var flag = stream.get_flag(Flag.IDENTITY); + flag.resumed = true; + flag.finished = true; + + foreach (XmppStreamFlag f in flags) { + stream.add_flag(f); + } + + h_outbound = int.parse(resumed_node.get_attribute("h", NS_URI)); + handle_incoming_h(stream, h_outbound); + foreach (var id in in_flight_stanzas.keys) { + node_queue.add(in_flight_stanzas[id]); + } + in_flight_stanzas.clear(); + check_queue(stream); + ((IoXmppStream)stream).write_obj = this; + + stream.get_flag(Flag.IDENTITY).finished = true; + } + + public void on_resume_failed(XmppStream stream, StanzaNode failed_node) { + session_id = null; + string? h_acked = failed_node.get_attribute("h", NS_URI); + if (h_acked != null) { + h_outbound = int.parse(h_acked); + handle_incoming_h(stream, h_outbound); + } + foreach (var id in in_flight_stanzas.keys) { + in_flight_stanzas[id].promise.set_exception(new IOError.FAILED("Stanza not acked and session not resumed")); + } + in_flight_stanzas.clear(); + check_queue(stream); + stream.received_features_node(stream); + + stream.get_flag(Flag.IDENTITY).finished = true; + } + + public void on_enabled(XmppStream stream, StanzaNode enabled_node) { + h_inbound = 0; + session_id = enabled_node.get_attribute("id", NS_URI); + flags = stream.flags; + ((IoXmppStream)stream).write_obj = this; + + stream.get_flag(Flag.IDENTITY).finished = true; + } + 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.begin(stream, node); @@ -217,19 +238,73 @@ public class Module : XmppStreamNegotiationModule, WriteNodeFunc { in_flight_stanzas.unset(nr); } } - - private bool stream_has_sm_feature(XmppStream stream) { - return stream.features.get_subnode("sm", NS_URI) != null; - } } public class Flag : XmppStreamFlag { public static FlagIdentity<Flag> IDENTITY = new FlagIdentity<Flag>(NS_URI, "stream_management"); - public bool finished = false; + public bool finished { get; set; default=false; } public bool resumed = false; public override string get_ns() { return NS_URI; } public override string get_id() { return IDENTITY.id; } + + public static Flag get_add_flag(XmppStream stream) { + Flag? flag = stream.get_flag(IDENTITY); + if (flag == null) { + flag = new Flag(); + stream.add_flag(flag); + } + return flag; + } } + public class Bind2Activation : Bind2.Bind2InlineActivation { + public override StanzaNode? get_activation_node(XmppStream stream) { + // Always insert an enable node, even if we also try to resume in Sasl2. Because if resumption fails, enable will be used. + Flag.get_add_flag(stream); + + stream.get_module(Module.IDENTITY).h_outbound = 0; + + return new StanzaNode.build("enable", NS_URI).add_self_xmlns().put_attribute("resume", "true"); + } + + public override void on_bound(XmppStream stream, StanzaNode bound_node) { + StanzaNode? enabled_node = bound_node.get_subnode("enabled", NS_URI); + if (enabled_node != null) { + stream.get_module(Module.IDENTITY).on_enabled(stream, enabled_node); + } + } + } + + public class Sasl2Activation : ExtensibleSaslProfile.Sasl2InlineActivation { + public override StanzaNode? get_activation_node(XmppStream stream, StanzaNode node) { + if (stream.get_module(Module.IDENTITY).session_id != null) { + Flag.get_add_flag(stream); + + var sm_module = stream.get_module(Module.IDENTITY); + + StanzaNode resume_node = new StanzaNode.build("resume", NS_URI).add_self_xmlns() + .put_attribute("h", sm_module.h_inbound.to_string()) + .put_attribute("previd", sm_module.session_id); + return resume_node; + } + return null; + } + + public override void on_bound(XmppStream stream, Jid authorization_identifier, StanzaNode success_node) { + StanzaNode? resumed_node = success_node.get_subnode("resumed", NS_URI); + if (resumed_node != null) { + stream.get_module(Module.IDENTITY).on_resumed(stream, resumed_node); + } + + StanzaNode? failed_node = success_node.get_subnode("failed", NS_URI); + if (failed_node != null) { + stream.get_module(Module.IDENTITY).on_resume_failed(stream, failed_node); + } + } + } + + private bool stream_has_sm_feature(XmppStream stream) { + return stream.features.get_subnode("sm", NS_URI) != null; + } } |