aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--libdino/src/entity/message.vala3
-rw-r--r--libdino/src/service/message_processor.vala3
-rw-r--r--xmpp-vala/src/core/stanza_writer.vala16
-rw-r--r--xmpp-vala/src/core/xmpp_stream.vala28
-rw-r--r--xmpp-vala/src/module/xep/0198_stream_management.vala114
5 files changed, 146 insertions, 18 deletions
diff --git a/libdino/src/entity/message.vala b/libdino/src/entity/message.vala
index 89ad241a..6670ec5d 100644
--- a/libdino/src/entity/message.vala
+++ b/libdino/src/entity/message.vala
@@ -14,7 +14,8 @@ public class Message : Object {
READ,
ACKNOWLEDGED,
UNSENT,
- WONTSEND
+ WONTSEND,
+ SENT
}
public enum Type {
diff --git a/libdino/src/service/message_processor.vala b/libdino/src/service/message_processor.vala
index 74192e16..6c415deb 100644
--- a/libdino/src/service/message_processor.vala
+++ b/libdino/src/service/message_processor.vala
@@ -638,6 +638,9 @@ public class MessageProcessor : StreamInteractionModule, Object {
stream.get_module(MessageModule.IDENTITY).send_message.begin(stream, new_message, (_, res) => {
try {
stream.get_module(MessageModule.IDENTITY).send_message.end(res);
+ if (message.marked == Message.Marked.NONE/* && (yield stream.get_module(Xep.ServiceDiscovery.Module.IDENTITY).has_entity_feature(stream, conversation.account.bare_jid, Xep.UniqueStableStanzaIDs.NS_URI))*/) {
+ message.marked = Message.Marked.SENT;
+ }
// The server might not have given us the resource we asked for. In that case, store the actual resource the message was sent with. Relevant for deduplication.
Jid? current_own_jid = stream.get_flag(Bind.Flag.IDENTITY).my_jid;
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;
+}
+
}
diff --git a/xmpp-vala/src/module/xep/0198_stream_management.vala b/xmpp-vala/src/module/xep/0198_stream_management.vala
index 8ac41eb7..c44e2c70 100644
--- a/xmpp-vala/src/module/xep/0198_stream_management.vala
+++ b/xmpp-vala/src/module/xep/0198_stream_management.vala
@@ -1,13 +1,71 @@
+using Gee;
+
namespace Xmpp.Xep.StreamManagement {
public const string NS_URI = "urn:xmpp:sm:3";
-public class Module : XmppStreamNegotiationModule {
+public class Module : XmppStreamNegotiationModule, WriteNodeFunc {
public static ModuleIdentity<Module> IDENTITY = new ModuleIdentity<Module>(NS_URI, "0198_stream_management");
- public int h_inbound { get; private set; default=0; }
+ public int h_inbound = 0;
+ public int h_outbound = 0;
+
public string? session_id { get; set; default=null; }
public Gee.List<XmppStreamFlag> flags = null;
+ private HashMap<int, QueueItem> in_flight_stanzas = new HashMap<int, QueueItem>();
+ private Gee.List<QueueItem> node_queue = new ArrayList<QueueItem>();
+
+ private class QueueItem {
+ public StanzaNode node;
+ public Promise<IOError?> promise;
+
+ public QueueItem(StanzaNode node, Promise<IOError?> promise) {
+ this.node = node;
+ this.promise = promise;
+ }
+ }
+
+ public async void write_stanza(XmppStream stream, StanzaNode node) throws IOError {
+ if (stream.has_flag(Flag.IDENTITY)) {
+ var promise = new Promise<IOError?>();
+
+ node_queue.add(new QueueItem(node, promise));
+ check_queue(stream);
+
+ yield promise.future.wait_async();
+ } else {
+ yield write_node(stream, node);
+ }
+ }
+
+ internal async void write_node(XmppStream stream, StanzaNode node) throws IOError {
+ StanzaWriter? writer = stream.writer;
+ if (writer == null) throw new IOStreamError.WRITE("trying to write, but no stream open");
+ try {
+ stream.log.node("OUT", node, stream);
+ if (node.name == "message" || node.name == "iq" || node.name == "presence") {
+ var r_node = new StanzaNode.build("r", NS_URI).add_self_xmlns();
+ stream.log.node("OUT", r_node, stream);
+ yield ((!)writer).write_nodes(node, r_node);
+ } else {
+ yield ((!)writer).write_node(node);
+ }
+ } catch (XmlError e) {
+ throw new IOStreamError.WRITE(e.message);
+ }
+ }
+
+ private void check_queue(XmppStream stream) throws IOError {
+ while (!node_queue.is_empty && in_flight_stanzas.size < 10) {
+ QueueItem queue_item = node_queue.remove_at(0);
+ StanzaNode node = queue_item.node;
+
+ if (node.name == "message" || node.name == "iq" || node.name == "presence") {
+ in_flight_stanzas[++h_outbound] = queue_item;
+ }
+ write_node.begin(stream, node);
+ }
+ }
public override void attach(XmppStream stream) {
stream.get_module(Bind.Module.IDENTITY).bound_to_resource.connect(check_enable);
@@ -19,7 +77,15 @@ public class Module : XmppStreamNegotiationModule {
stream.received_iq_stanza.connect(on_stanza_received);
}
- public override void detach(XmppStream stream) { }
+ public override void detach(XmppStream stream) {
+ stream.get_module(Bind.Module.IDENTITY).bound_to_resource.disconnect(check_enable);
+ stream.received_features_node.disconnect(check_resume);
+
+ stream.received_nonza.disconnect(on_received_nonza);
+ stream.received_message_stanza.disconnect(on_stanza_received);
+ stream.received_presence_stanza.disconnect(on_stanza_received);
+ stream.received_iq_stanza.disconnect(on_stanza_received);
+ }
public static void require(XmppStream stream) {
if (stream.get_module(IDENTITY) == null) stream.add_module(new PrivateXmlStorage.Module());
@@ -35,7 +101,7 @@ public class Module : XmppStreamNegotiationModule {
public override string get_id() { return IDENTITY.id; }
private void on_stanza_received(XmppStream stream, StanzaNode node) {
- lock (h_inbound) h_inbound++;
+ h_inbound++;
}
private void check_resume(XmppStream stream) {
@@ -45,7 +111,7 @@ public class Module : XmppStreamNegotiationModule {
StanzaNode node = new StanzaNode.build("resume", NS_URI).add_self_xmlns()
.put_attribute("h", h_inbound.to_string())
.put_attribute("previd", session_id);
- stream.write(node);
+ write_node(stream, node);
stream.add_flag(new Flag());
}
}
@@ -54,8 +120,9 @@ public class Module : XmppStreamNegotiationModule {
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");
- stream.write(node);
+ write_node(stream, node);
stream.add_flag(new Flag());
+ h_outbound = 0;
}
}
@@ -69,17 +136,32 @@ public class Module : XmppStreamNegotiationModule {
stream.get_flag(Flag.IDENTITY).finished = true;
if (node.name == "enabled") {
- lock(h_inbound) h_inbound = 0;
+ h_inbound = 0;
session_id = node.get_attribute("id", NS_URI);
flags = stream.flags;
+ stream.write_obj = this;
} else if (node.name == "resumed") {
foreach (XmppStreamFlag flag in flags) {
stream.add_flag(flag);
}
stream.negotiation_complete = true;
+
+ 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);
+ stream.write_obj = this;
} else if (node.name == "failed") {
stream.received_features_node(stream);
session_id = null;
+ foreach (var id in in_flight_stanzas.keys) {
+ in_flight_stanzas[id].promise.set_exception(new IOError.FAILED("bla"));
+ }
+ in_flight_stanzas.clear();
+ check_queue(stream);
}
}
}
@@ -87,11 +169,27 @@ public class Module : XmppStreamNegotiationModule {
private void send_ack(XmppStream stream) {
StanzaNode node = new StanzaNode.build("a", NS_URI).add_self_xmlns().put_attribute("h", h_inbound.to_string());
- stream.write(node);
+ write_node(stream, node);
}
private void handle_ack(XmppStream stream, StanzaNode node) {
+ string? h_acked = node.get_attribute("h", NS_URI);
+ int parsed_int = int.parse(h_acked);
+ handle_incoming_h(stream, parsed_int);
+ check_queue(stream);
+ }
+ private void handle_incoming_h(XmppStream stream, int h) {
+ var remove_nrs = new ArrayList<int>();
+ foreach (int nr in in_flight_stanzas.keys) {
+ if (nr <= h) {
+ in_flight_stanzas[nr].promise.set_value(null);
+ remove_nrs.add(nr);
+ }
+ }
+ foreach (int nr in remove_nrs) {
+ in_flight_stanzas.unset(nr);
+ }
}
private bool stream_has_sm_feature(XmppStream stream) {
generated by cgit v1.2.3-70-g09d2 (git 2.47.0) at 2024-11-15 02:56:33 +0000