packages/moxxmpp/lib/src/connection.dart

@@ -198,6 +198,9 @@ 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;
@@ -425,6 +428,7 @@ class XmppConnection {
     }
 
     await _setConnectionState(XmppConnectionState.error);
+    await _resetIsConnectionRunning();
     await _reconnectionPolicy.onFailure();
   }
 
@@ -433,9 +437,9 @@ class XmppConnection {
     if (event is XmppSocketErrorEvent) {
       await handleError(SocketError(event));
     } else if (event is XmppSocketClosureEvent) {
-      if (!event.expected) {
-        _log.fine('Received unexpected XmppSocketClosureEvent. Reconnecting...');
-        await handleError(SocketError(XmppSocketErrorEvent(event)));
+      if (_socketClosureTriggersReconnect) {
+        _log.fine('Received XmppSocketClosureEvent. Reconnecting...');
+        await _reconnectionPolicy.onFailure();
       } else {
         _log.fine('Received XmppSocketClosureEvent. No reconnection attempt since _socketClosureTriggersReconnect is false...');
       }
@@ -846,7 +850,6 @@ 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;
     }
@@ -870,8 +873,6 @@ 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);
@@ -895,8 +896,6 @@ class XmppConnection {
         _log.finest('Negotiations done!');
 
         _updateRoutingState(RoutingState.handleStanzas);
-        await _reconnectionPolicy.onSuccess();
-        await _resetIsConnectionRunning();
         await _onNegotiationsDone();
       } else {
         _log.finest('Picking new negotiator...');
@@ -913,8 +912,6 @@ 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;
     }
@@ -1026,6 +1023,7 @@ class XmppConnection {
 
   Future<void> _disconnect({required XmppConnectionState state, bool triggeredByUser = true}) async {
     _reconnectionPolicy.setShouldReconnect(false);
+    _socketClosureTriggersReconnect = false;
 
     if (triggeredByUser) {
       getPresenceManager().sendUnavailablePresence();
@@ -1081,6 +1079,7 @@ class XmppConnection {
     }
 
     await _reconnectionPolicy.reset();
+    _socketClosureTriggersReconnect = true;
     await _sendEvent(ConnectingEvent());
 
     final smManager = getStreamManagementManager();

packages/moxxmpp/lib/src/socket.dart

@@ -5,17 +5,13 @@ 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 {
-  XmppSocketClosureEvent(this.expected);
-
-  /// Indicate that the socket did not close unexpectedly.
-  final bool expected;
-}
+class XmppSocketClosureEvent extends XmppSocketEvent {}
 
 /// 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,18 +185,9 @@ class StreamManagementManager extends XmppManagerBase {
       _disableStreamManagement();
       _streamResumed = false;
     } else if (event is ConnectionStateChangedEvent) {
-      switch (event.state) {
-      case XmppConnectionState.connected:
+      if (event.state == 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,31 +10,22 @@ import 'package:moxxmpp_socket_tcp/src/rfc_2782.dart';
 
 /// TCP socket implementation for XmppConnection
 class TCPSocketWrapper extends BaseSocketWrapper {
-  TCPSocketWrapper(this._logData);
-
-  /// The underlying Socket/SecureSocket instance.
+  TCPSocketWrapper(this._logData)
+  : _log = Logger('TCPSocketWrapper'),
+    _dataStream = StreamController.broadcast(),
+    _eventStream = StreamController.broadcast(),
+    _secure = false,
+    _ignoreSocketClosure = false;
   Socket? _socket;
-
-  /// Indicates that we expect a socket closure.
-  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.
+  bool _ignoreSocketClosure;
+  final StreamController<String> _dataStream;
+  final StreamController<XmppSocketEvent> _eventStream;
   StreamSubscription<dynamic>? _socketSubscription;
 
-  /// Logger
-  final Logger _log = Logger('TCPSocketWrapper');
-
-  /// Flag to indicate if incoming and outgoing data should get logged.
+  final Logger _log;
   final bool _logData;
 
-  /// Indiacted whether the connection is secure.
-  bool _secure = false;
+  bool _secure;
 
   @override
   bool isSecure() => _secure;
@@ -81,6 +72,7 @@ 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
@@ -97,11 +89,13 @@ 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;
@@ -124,16 +118,19 @@ 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;
       }
     }
@@ -147,6 +144,7 @@ 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,
@@ -156,6 +154,7 @@ class TCPSocketWrapper extends BaseSocketWrapper {
       return true;
     } on Exception catch(e) {
       _log.finest('Failure! $e');
+      _ignoreSocketClosure = false;
       return false;
     }
   }
@@ -181,10 +180,9 @@ class TCPSocketWrapper extends BaseSocketWrapper {
       return false;
     }
 
-    try {
-      // The socket is closed during the entire process
-      _expectSocketClosure = true;
+    _ignoreSocketClosure = true;
     
+    try {
       _socket = await SecureSocket.secure(
         _socket!,
         supportedProtocols: const [ xmppClientALPNId ],
@@ -192,10 +190,12 @@ 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,14 +226,15 @@ class TCPSocketWrapper extends BaseSocketWrapper {
     );
     // ignore: implicit_dynamic_parameter
     _socket!.done.then((_) {
-      _eventStream.add(XmppSocketClosureEvent(_expectSocketClosure));
-      _expectSocketClosure = false;
+      if (!_ignoreSocketClosure) {
+        _eventStream.add(XmppSocketClosureEvent());
+      }
     });
   }
   
   @override
   Future<bool> connect(String domain, { String? host, int? port }) async {
-    _expectSocketClosure = false;
+    _ignoreSocketClosure = false;
     _secure = false;
 
     // Connection order:
@@ -266,8 +267,6 @@ class TCPSocketWrapper extends BaseSocketWrapper {
 
   @override
   void close() {
-    _expectSocketClosure = true;
-
     if (_socketSubscription != null) {
       _log.finest('Closing socket subscription');
       _socketSubscription!.cancel();
@@ -278,11 +277,13 @@ class TCPSocketWrapper extends BaseSocketWrapper {
       return;
     }
 
+    _ignoreSocketClosure = true;
     try {
       _socket!.close();
     } catch(e) {
       _log.warning('Closing socket threw exception: $e');
     }
+    _ignoreSocketClosure = false;
   }
 
   @override
@@ -315,5 +316,7 @@ class TCPSocketWrapper extends BaseSocketWrapper {
   }
 
   @override
-  void prepareDisconnect() {}
+  void prepareDisconnect() {
+    _ignoreSocketClosure = true;
+  }
 }