WIP fix stream management/lack of resumecatch-disconnect-error

3 files changed, 55 insertions, 20 deletions

diff --git a/libdino/src/entity/conversation.vala b/libdino/src/entity/conversation.vala
index 57f0037d..56e74d8b 100644
--- a/libdino/src/entity/conversation.vala
+++ b/libdino/src/entity/conversation.vala
@@ -74,6 +74,7 @@ public class Conversation : Object {
         notify_setting = (NotifySetting) row[db.conversation.notification];
         send_typing = (Setting) row[db.conversation.send_typing];
         send_marker = (Setting) row[db.conversation.send_marker];
+        debug(@"account=$(account.display_name),encryption=$(encryption),resource=$(resource ?? "null")");
 
         notify.connect(on_update);
     }
diff --git a/libdino/src/service/connection_manager.vala b/libdino/src/service/connection_manager.vala
index b01f1ecd..9d5e0ebb 100644
--- a/libdino/src/service/connection_manager.vala
+++ b/libdino/src/service/connection_manager.vala
@@ -20,6 +20,7 @@ public class ConnectionManager : Object {
 
     private HashMap<Account, Connection> connections = new HashMap<Account, Connection>(Account.hash_func, Account.equals_func);
     private HashMap<Account, ConnectionError> connection_errors = new HashMap<Account, ConnectionError>(Account.hash_func, Account.equals_func);
+    private HashMap<Account, string> session_ids = new HashMap<Account, string>(Account.hash_func, Account.equals_func);
     private NetworkMonitor? network_monitor;
     private ModuleManager module_manager;
     public string? log_options;
@@ -67,7 +68,7 @@ public class ConnectionManager : Object {
                 try {
                     yield stream.disconnect();
                 } catch (IOError e) {
-                    warning(@"caught IOError $(e.code)");
+                    warning(@"caught IOError: $(e.message)");
                 }
             }
             stream = null;
@@ -163,7 +164,7 @@ public class ConnectionManager : Object {
     public async void disconnect_account(Account account) {
         if (connections.has_key(account)) {
             make_offline(account);
-            connections[account].disconnect_account.begin();
+            yield connections[account].disconnect_account();
             connections.unset(account);
         }
     }
@@ -184,7 +185,6 @@ public class ConnectionManager : Object {
                     change_connection_state(account, ConnectionState.DISCONNECTED);
                     return on_invalid_certificate(account.domainpart, peer_cert, errors); }
         );
-        connections[account].stream = stream_result.stream;
 
         if (stream_result.stream == null) {
             if (stream_result.tls_errors != null) {
@@ -192,7 +192,7 @@ public class ConnectionManager : Object {
                 return;
             }
 
-            debug("[%s] Could not connect", account.bare_jid.to_string());
+            warning("[%s] Could not connect", account.bare_jid.to_string());
 
             change_connection_state(account, ConnectionState.DISCONNECTED);
 
@@ -201,7 +201,18 @@ public class ConnectionManager : Object {
 
         XmppStream stream = stream_result.stream;
 
-        debug("[%s] New connection with resource %s: %p", account.bare_jid.to_string(), resource, stream);
+        print("[%s] New connection with resource %s: %p\n", account.bare_jid.to_string(), resource, stream);
+
+        var? module = stream.get_module(Xep.StreamManagement.Module.IDENTITY);
+        if (module != null) {
+            if (session_ids[account] != null) {
+                module.session_id = session_ids[account];
+            }
+
+            module.got_session_id.connect((session_id) =>  {
+                session_ids[account] = session_id;
+            });
+        }
 
         stream.attached_modules.connect((stream) => {
             stream_attached_modules(account, stream);
@@ -220,6 +231,7 @@ public class ConnectionManager : Object {
                 warning("Got node for outdated connection");
             }
         });
+        connections[account].stream = stream_result.stream;
         stream_opened(account, stream);
 
         try {
@@ -256,13 +268,14 @@ public class ConnectionManager : Object {
         }
     }
 
-    private void check_ping_reconnect(Xmpp.Xep.Ping.Module identity,
-            XmppStream stream, Account account) {
-        identity.send_ping.begin(stream, account.bare_jid.domain_jid, () => {
-            if (connections[account].stream != stream) return;
-            connections[account].acked = true;
-            change_connection_state(account, ConnectionState.CONNECTED);
-        });
+    private async void check_ping_reconnect(Account account) {
+        yield connections[account].reset();
+        yield connect_stream(account);
+
+        XmppStream stream = connections[account].stream;
+
+        var? identity = stream.get_module(Xep.Ping.Module.IDENTITY);
+        if (identity == null) return;
 
         Timeout.add_seconds(10, () => {
             if (!connections.has_key(account)) return false;
@@ -270,7 +283,7 @@ public class ConnectionManager : Object {
             if (connections[account].acked) return false;
 
             // Reconnect. Nothing gets through the stream.
-            debug("[%s %p] Ping timeouted. Reconnecting", account.bare_jid.to_string(), stream);
+            warning("[%s %p] Ping timeouted. Reconnecting", account.bare_jid.to_string(), stream);
             change_connection_state(account, ConnectionState.DISCONNECTED);
 
             connections[account].reset.begin((_, res) => {
@@ -279,9 +292,20 @@ public class ConnectionManager : Object {
             });
             return false;
         });
+
+        yield identity.send_ping(stream, account.bare_jid.domain_jid);
+
+        connections[account].acked = true;
+        change_connection_state(account, ConnectionState.CONNECTED);
+
+        var? module = stream.get_module(Xep.StreamManagement.Module.IDENTITY);
+        if (module != null) {
+            module.session_id = session_ids[account];
+            stream.require_setup();
+        }
     }
 
-    private void check_reconnect(Account account, bool directly_reconnect = false) {
+    private void check_reconnect(Account account) {
         if (!connections.has_key(account)) return;
 
         var cancellable = new Cancellable();
@@ -294,16 +318,14 @@ public class ConnectionManager : Object {
             debug(@"can-reach: $(reachable)");
 
             if (reachable) {
-                Xmpp.Xep.Ping.Module? identity = null;
                 if (connections[account].connection_state == ConnectionState.CONNECTING) return;
 
                 XmppStream? stream = connections[account].stream;
                 connections[account].acked = false;
 
-                if (stream != null
-                    && (identity = stream.get_module(Xep.Ping.Module.IDENTITY)) != null) {
+                if (stream != null) {
                     change_connection_state(account, ConnectionState.CONNECTING);
-                    check_ping_reconnect(identity, stream, account);
+                    check_ping_reconnect.begin(account);
                 }
                 else {
                     change_connection_state(account, ConnectionState.DISCONNECTED);
@@ -316,7 +338,7 @@ public class ConnectionManager : Object {
             }
 
         } catch (Error e) {
-            print ("Error: %s\n", e.message);
+            warning(@"$(e.message)");
         }
 
     }
diff --git a/xmpp-vala/src/module/xep/0198_stream_management.vala b/xmpp-vala/src/module/xep/0198_stream_management.vala
index 68eee8ae..db362e33 100644
--- a/xmpp-vala/src/module/xep/0198_stream_management.vala
+++ b/xmpp-vala/src/module/xep/0198_stream_management.vala
@@ -7,6 +7,8 @@ public const string NS_URI = "urn:xmpp:sm:3";
 public class Module : XmppStreamNegotiationModule, WriteNodeFunc {
     public static ModuleIdentity<Module> IDENTITY = new ModuleIdentity<Module>(NS_URI, "0198_stream_management");
 
+    public signal void got_session_id(string session_id);
+
     public int h_inbound = 0;
     public int h_outbound = 0;
 
@@ -127,11 +129,12 @@ public class Module : XmppStreamNegotiationModule, WriteNodeFunc {
         h_inbound++;
     }
 
-    private void check_resume(XmppStream stream) {
+    public void check_resume(XmppStream stream) {
         if (stream_has_sm_feature(stream) && session_id != null) {
             StanzaNode node = new StanzaNode.build("resume", NS_URI).add_self_xmlns()
                 .put_attribute("h", h_inbound.to_string())
                 .put_attribute("previd", session_id);
+            session_id = null;
             write_node.begin(stream, node);
             stream.add_flag(new Flag());
         }
@@ -158,6 +161,8 @@ public class Module : XmppStreamNegotiationModule, WriteNodeFunc {
                 if (node.name == "enabled") {
                     h_inbound = 0;
                     session_id = node.get_attribute("id", NS_URI);
+                    print(@"this=%p, new session_id=$(session_id)\n", this);
+                    got_session_id(session_id);
                     flags = stream.flags;
                     ((IoXmppStream)stream).write_obj = this;
                 } else if (node.name == "resumed") {
@@ -219,6 +224,13 @@ public class Module : XmppStreamNegotiationModule, WriteNodeFunc {
     }
 
     private bool stream_has_sm_feature(XmppStream stream) {
+        print("stream=%p\n", stream);
+        if (session_id != null) {
+            print(@"session_id=$(session_id)\n");
+        }
+        if (stream.features.get_subnode("sm", NS_URI) != null) {
+            print(@"stream.features.get_subnode=$(stream.features.get_subnode("sm", NS_URI))\n");
+        }
         return stream.features.get_subnode("sm", NS_URI) != null;
     }
 }