fix: Add the closure expectation to the event

Call _reconnectionPolicy.onSuccess when negotiations are done. Never
call onFailure directly; only ever call handleError.

packages/moxxmpp/lib/src/connection.dart

@@ -198,9 +198,6 @@ class XmppConnection {
   // ignore: use_late_for_private_fields_and_variables
   Completer<XmppConnectionResult>? _connectionCompleter;
 
-  /// Controls whether an XmppSocketClosureEvent triggers a reconnection.
-  bool _socketClosureTriggersReconnect = true;
-
   /// Negotiators
   final Map<String, XmppFeatureNegotiatorBase> _featureNegotiators = {};
   XmppFeatureNegotiatorBase? _currentNegotiator;
@@ -428,7 +425,6 @@ class XmppConnection {
     }
 
     await _setConnectionState(XmppConnectionState.error);
-    await _resetIsConnectionRunning();
     await _reconnectionPolicy.onFailure();
   }
 
@@ -437,9 +433,9 @@ class XmppConnection {
     if (event is XmppSocketErrorEvent) {
       await handleError(SocketError(event));
     } else if (event is XmppSocketClosureEvent) {
-      if (_socketClosureTriggersReconnect) {
+      if (!event.expected) {
-        _log.fine('Received XmppSocketClosureEvent. Reconnecting...');
+        _log.fine('Received unexpected XmppSocketClosureEvent. Reconnecting...');
-        await _reconnectionPolicy.onFailure();
+        await handleError(SocketError(XmppSocketErrorEvent(event)));
       } else {
         _log.fine('Received XmppSocketClosureEvent. No reconnection attempt since _socketClosureTriggersReconnect is false...');
       }
@@ -850,6 +846,7 @@ class XmppConnection {
     final result = await _currentNegotiator!.negotiate(nonza);
     if (result.isType<NegotiatorError>()) {
       _log.severe('Negotiator returned an error');
+      await _resetIsConnectionRunning();
       await handleError(result.get<NegotiatorError>());
       return;
     }
@@ -873,6 +870,8 @@ class XmppConnection {
         if (_isMandatoryNegotiationDone(_streamFeatures) && !_isNegotiationPossible(_streamFeatures)) {
           _log.finest('Negotiations done!');
           _updateRoutingState(RoutingState.handleStanzas);
+          await _reconnectionPolicy.onSuccess();
+          await _resetIsConnectionRunning();
           await _onNegotiationsDone();
         } else {
           _currentNegotiator = getNextNegotiator(_streamFeatures);
@@ -896,6 +895,8 @@ class XmppConnection {
         _log.finest('Negotiations done!');
 
         _updateRoutingState(RoutingState.handleStanzas);
+        await _reconnectionPolicy.onSuccess();
+        await _resetIsConnectionRunning();
         await _onNegotiationsDone();
       } else {
         _log.finest('Picking new negotiator...');
@@ -912,6 +913,8 @@ class XmppConnection {
       _log.finest('Negotiator wants to skip the remaining negotiation... Negotiations (assumed) done!');
 
       _updateRoutingState(RoutingState.handleStanzas);
+      await _reconnectionPolicy.onSuccess();
+      await _resetIsConnectionRunning();
       await _onNegotiationsDone();
       break;
     }
@@ -1023,7 +1026,6 @@ class XmppConnection {
 
   Future<void> _disconnect({required XmppConnectionState state, bool triggeredByUser = true}) async {
     _reconnectionPolicy.setShouldReconnect(false);
-    _socketClosureTriggersReconnect = false;
 
     if (triggeredByUser) {
       getPresenceManager().sendUnavailablePresence();
@@ -1079,7 +1081,6 @@ class XmppConnection {
     }
 
     await _reconnectionPolicy.reset();
-    _socketClosureTriggersReconnect = true;
     await _sendEvent(ConnectingEvent());
 
     final smManager = getStreamManagementManager();

packages/moxxmpp/lib/src/socket.dart

@@ -5,13 +5,17 @@ abstract class XmppSocketEvent {}
 
 /// Triggered by the socket when an error occurs.
 class XmppSocketErrorEvent extends XmppSocketEvent {
-
   XmppSocketErrorEvent(this.error);
   final Object error;
 }
 
 /// Triggered when the socket is closed
-class XmppSocketClosureEvent extends XmppSocketEvent {}
+class XmppSocketClosureEvent extends XmppSocketEvent {
+  XmppSocketClosureEvent(this.expected);
+
+  /// Indicate that the socket did not close unexpectedly.
+  final bool expected;
+}
 
 /// This class is the base for a socket that XmppConnection can use.
 abstract class BaseSocketWrapper {

packages/moxxmpp/lib/src/xeps/xep_0198/xep_0198.dart

@@ -185,9 +185,18 @@ class StreamManagementManager extends XmppManagerBase {
       _disableStreamManagement();
       _streamResumed = false;
     } else if (event is ConnectionStateChangedEvent) {
-      if (event.state == XmppConnectionState.connected) {
+      switch (event.state) {
+      case XmppConnectionState.connected:
         // Push out all pending stanzas
         await onStreamResumed(0);
+        break;
+      case XmppConnectionState.error:
+      case XmppConnectionState.notConnected:
+        _stopAckTimer();
+        break;
+      case XmppConnectionState.connecting:
+        // NOOP
+        break;
       }
     }
   }

packages/moxxmpp_socket_tcp/lib/src/socket.dart

@@ -10,22 +10,31 @@ import 'package:moxxmpp_socket_tcp/src/rfc_2782.dart';
 
 /// TCP socket implementation for XmppConnection
 class TCPSocketWrapper extends BaseSocketWrapper {
-  TCPSocketWrapper(this._logData)
+  TCPSocketWrapper(this._logData);
-  : _log = Logger('TCPSocketWrapper'),
+
-    _dataStream = StreamController.broadcast(),
+  /// The underlying Socket/SecureSocket instance.
-    _eventStream = StreamController.broadcast(),
-    _secure = false,
-    _ignoreSocketClosure = false;
   Socket? _socket;
-  bool _ignoreSocketClosure;
+
-  final StreamController<String> _dataStream;
+  /// Indicates that we expect a socket closure.
-  final StreamController<XmppSocketEvent> _eventStream;
+  bool _expectSocketClosure = false;
+
+  /// The stream of incoming data from the socket.
+  final StreamController<String> _dataStream = StreamController.broadcast();
+
+  /// The stream of outgoing (TCPSocketWrapper -> XmppConnection) events.
+  final StreamController<XmppSocketEvent> _eventStream = StreamController.broadcast();
+
+  /// A subscription on the socket's data stream.
   StreamSubscription<dynamic>? _socketSubscription;
 
-  final Logger _log;
+  /// Logger
+  final Logger _log = Logger('TCPSocketWrapper');
+
+  /// Flag to indicate if incoming and outgoing data should get logged.
   final bool _logData;
 
-  bool _secure;
+  /// Indiacted whether the connection is secure.
+  bool _secure = false;
 
   @override
   bool isSecure() => _secure;
@@ -72,7 +81,6 @@ class TCPSocketWrapper extends BaseSocketWrapper {
     for (final srv in results) {
       try {
         _log.finest('Attempting secure connection to ${srv.target}:${srv.port}...');
-        _ignoreSocketClosure = true;
 
         // Workaround: We cannot set the SNI directly when using SecureSocket.connect.
         // instead, we connect using a regular socket and then secure it. This allows
@@ -89,13 +97,11 @@ class TCPSocketWrapper extends BaseSocketWrapper {
           onBadCertificate: (cert) => onBadCertificate(cert, domain),
         );
 
-        _ignoreSocketClosure = false;
         _secure = true;
         _log.finest('Success!');
         return true;
       } on Exception catch(e) {
         _log.finest('Failure! $e');
-        _ignoreSocketClosure = false;
 
         if (e is HandshakeException) {
           failedDueToTLS = true;
@@ -118,19 +124,16 @@ class TCPSocketWrapper extends BaseSocketWrapper {
     for (final srv in results) {
       try {
         _log.finest('Attempting connection to ${srv.target}:${srv.port}...');
-        _ignoreSocketClosure = true;
         _socket = await Socket.connect(
           srv.target,
           srv.port,
           timeout: const Duration(seconds: 5),
         );
 
-        _ignoreSocketClosure = false;
         _log.finest('Success!');
         return true;
       } on Exception catch(e) {
         _log.finest('Failure! $e');
-        _ignoreSocketClosure = false;
         continue;
       }
     }
@@ -144,7 +147,6 @@ class TCPSocketWrapper extends BaseSocketWrapper {
   Future<bool> _hostPortConnect(String host, int port) async {
     try {
       _log.finest('Attempting fallback connection to $host:$port...');
-      _ignoreSocketClosure = true;
       _socket = await Socket.connect(
         host,
         port,
@@ -154,7 +156,6 @@ class TCPSocketWrapper extends BaseSocketWrapper {
       return true;
     } on Exception catch(e) {
       _log.finest('Failure! $e');
-      _ignoreSocketClosure = false;
       return false;
     }
   }
@@ -179,10 +180,11 @@ class TCPSocketWrapper extends BaseSocketWrapper {
       _log.severe('Failed to secure socket since _socket is null');
       return false;
     }
-
-    _ignoreSocketClosure = true;
     
     try {
+      // The socket is closed during the entire process
+      _expectSocketClosure = true;
+
       _socket = await SecureSocket.secure(
         _socket!,
         supportedProtocols: const [ xmppClientALPNId ],
@@ -190,12 +192,10 @@ class TCPSocketWrapper extends BaseSocketWrapper {
       );
 
       _secure = true;
-      _ignoreSocketClosure = false;
       _setupStreams();
       return true;
     } on Exception catch (e) {
       _log.severe('Failed to secure socket: $e');
-      _ignoreSocketClosure = false;
 
       if (e is HandshakeException) {
         _eventStream.add(XmppSocketTLSFailedEvent());
@@ -226,15 +226,14 @@ class TCPSocketWrapper extends BaseSocketWrapper {
     );
     // ignore: implicit_dynamic_parameter
     _socket!.done.then((_) {
-      if (!_ignoreSocketClosure) {
+      _eventStream.add(XmppSocketClosureEvent(_expectSocketClosure));
-        _eventStream.add(XmppSocketClosureEvent());
+      _expectSocketClosure = false;
-      }
     });
   }
   
   @override
   Future<bool> connect(String domain, { String? host, int? port }) async {
-    _ignoreSocketClosure = false;
+    _expectSocketClosure = false;
     _secure = false;
 
     // Connection order:
@@ -267,6 +266,8 @@ class TCPSocketWrapper extends BaseSocketWrapper {
 
   @override
   void close() {
+    _expectSocketClosure = true;
+
     if (_socketSubscription != null) {
       _log.finest('Closing socket subscription');
       _socketSubscription!.cancel();
@@ -277,13 +278,11 @@ class TCPSocketWrapper extends BaseSocketWrapper {
       return;
     }
 
-    _ignoreSocketClosure = true;
     try {
       _socket!.close();
     } catch(e) {
       _log.warning('Closing socket threw exception: $e');
     }
-    _ignoreSocketClosure = false;
   }
 
   @override
@@ -316,7 +315,5 @@ class TCPSocketWrapper extends BaseSocketWrapper {
   }
 
   @override
-  void prepareDisconnect() {
+  void prepareDisconnect() {}
-    _ignoreSocketClosure = true;
-  }
 }