fix(connectivity_plus): prevent Linux client leaks

packages/connectivity_plus/connectivity_plus/lib/src/connectivity_plus_linux.dart

@@ -21,13 +21,15 @@ class ConnectivityPlusLinuxPlugin extends ConnectivityPlatform {
   @override
   Future<List<ConnectivityResult>> checkConnectivity() async {
     final client = createClient();
-    await client.connect();
-    final connectivity = _getConnectivity(client);
-    await client.close();
-    return connectivity;
+    try {
+      await client.connect();
+      return _getConnectivity(client);
+    } finally {
+      await client.close();
+    }
   }
 
-  NetworkManagerClient? _client;
+  _NetworkManagerClientSession? _clientSession;
   StreamController<List<ConnectivityResult>>? _controller;
 
   /// Returns a Stream of ConnectivityResults changes.
@@ -69,26 +71,69 @@ class ConnectivityPlusLinuxPlugin extends ConnectivityPlatform {
   }
 
   Future<void> _startListenConnectivity() async {
-    _client ??= createClient();
-    await _client!.connect();
-    _addConnectivity(_client!);
-    _client!.propertiesChanged.listen((properties) {
-      if (properties.contains('Connectivity')) {
-        _addConnectivity(_client!);
+    final session =
+        _clientSession ??= _NetworkManagerClientSession(createClient());
+    try {
+      await session.connected;
+
+      if (!identical(_clientSession, session)) {
+        return;
+      }
+
+      final client = session.client;
+      _addConnectivity(client);
+      session.propertiesChangedSubscription =
+          client.propertiesChanged.listen((properties) {
+        if (identical(_clientSession, session) &&
+            properties.contains('Connectivity')) {
+          _addConnectivity(client);
+        }
+      });
+    } catch (_) {
+      if (identical(_clientSession, session)) {
+        _clientSession = null;
       }
-    });
+      await session.close();
+      rethrow;
+    }
   }
 
   void _addConnectivity(NetworkManagerClient client) {
     _controller!.add(_getConnectivity(client));
   }
 
   Future<void> _stopListenConnectivity() async {
-    await _client?.close();
-    _client = null;
+    final session = _clientSession;
+    _clientSession = null;
+    await session?.close();
   }
 
   @visibleForTesting
   // ignore: prefer_function_declarations_over_variables
   NetworkManagerClientFactory createClient = () => NetworkManagerClient();
 }
+
+class _NetworkManagerClientSession {
+  _NetworkManagerClientSession(this.client) : connected = client.connect();
+
+  final NetworkManagerClient client;
+  final Future<void> connected;
+  StreamSubscription<List<String>>? propertiesChangedSubscription;
+
+  Future<void>? _closeFuture;
+
+  Future<void> close() => _closeFuture ??= _close();
+
+  Future<void> _close() async {
+    try {
+      await connected;
+    } catch (_) {
+      // Connection errors are reported by the listener setup.
+    }
+    try {
+      await propertiesChangedSubscription?.cancel();
+    } finally {
+      await client.close();
+    }
+  }
+}

packages/connectivity_plus/connectivity_plus/test/connectivity_plus_linux_test.dart

@@ -1,3 +1,5 @@
+import 'dart:async';
+
 import 'package:connectivity_plus/src/connectivity_plus_linux.dart';
 import 'package:connectivity_plus_platform_interface/connectivity_plus_platform_interface.dart';
 import 'package:flutter_test/flutter_test.dart';
@@ -101,6 +103,19 @@
         completion(equals([ConnectivityResult.none])));
   });
 
+  test('checkConnectivity closes the client after an error', () async {
+    final linux = ConnectivityPlusLinuxPlugin();
+    final client = MockNetworkManagerClient();
+    when(client.connect()).thenAnswer((_) async {});
+    when(client.close()).thenAnswer((_) async {});
+    when(client.connectivity).thenThrow(StateError('Failed to read state'));
+    linux.createClient = () => client;
+
+    await expectLater(linux.checkConnectivity(), throwsStateError);
+
+    verify(client.close()).called(1);
+  });
+
   test('connectivity changes', () {
     final linux = ConnectivityPlusLinuxPlugin();
     linux.createClient = () {
@@ -122,4 +137,101 @@
           [ConnectivityResult.none]
         ]));
   });
+
+  test('canceling the last listener cancels the property subscription',
+      () async {
+    final linux = ConnectivityPlusLinuxPlugin();
+    final client = MockNetworkManagerClient();
+    final propertiesChanged = StreamController<List<String>>.broadcast();
+    final clientClosed = Completer<void>();
+    addTearDown(propertiesChanged.close);
+    when(client.connect()).thenAnswer((_) async {});
+    when(client.close()).thenAnswer((_) async {
+      clientClosed.complete();
+    });
+    when(client.connectivity).thenReturn(NetworkManagerConnectivityState.full);
+    when(client.primaryConnectionType).thenReturn('wireless');
+    when(client.propertiesChanged).thenAnswer((_) => propertiesChanged.stream);
+    linux.createClient = () => client;
+
+    final initialEvent = Completer<void>();
+    final subscription = linux.onConnectivityChanged.listen((_) {
+      initialEvent.complete();
+    });
+    await initialEvent.future;
+    expect(propertiesChanged.hasListener, isTrue);
+
+    await subscription.cancel();
+    await clientClosed.future;
+
+    expect(propertiesChanged.hasListener, isFalse);
+    verify(client.close()).called(1);
+  });
+
+  test('canceling while connecting does not reuse the closing client',
+      () async {
+    final linux = ConnectivityPlusLinuxPlugin();
+    final clients = <MockNetworkManagerClient>[];
+    final connectCompleters = <Completer<void>>[];
+    final closeCompleters = <Completer<void>>[];
+    final closeCalledCompleters = <Completer<void>>[];
+    linux.createClient = () {
+      final client = MockNetworkManagerClient();
+      final connectCompleter = Completer<void>();
+      final closeCompleter = Completer<void>();
+      final closeCalledCompleter = Completer<void>();
+      when(client.connect()).thenAnswer((_) => connectCompleter.future);
+      when(client.close()).thenAnswer((_) {
+        closeCalledCompleter.complete();
+        return closeCompleter.future;
+      });
+      when(client.connectivity)
+          .thenReturn(NetworkManagerConnectivityState.full);
+      when(client.primaryConnectionType).thenReturn('wireless');
+      when(client.propertiesChanged)
+          .thenAnswer((_) => Stream<List<String>>.empty());
+      clients.add(client);
+      connectCompleters.add(connectCompleter);
+      closeCompleters.add(closeCompleter);
+      closeCalledCompleters.add(closeCalledCompleter);
+      return client;
+    };
+
+    final errors = <Object>[];
+    await runZonedGuarded(
+      () async {
+        final firstSubscription = linux.onConnectivityChanged.listen((_) {});
+        await firstSubscription.cancel();
+
+        final secondSubscription = linux.onConnectivityChanged.listen((_) {});
+
+        closeCompleters.first.complete();
+
+        for (final completer in connectCompleters) {
+          completer.complete();
+        }
+        await Future.wait(
+          connectCompleters.map((completer) => completer.future),
+        );
+
+        await secondSubscription.cancel();
+        for (final completer in closeCompleters) {
+          if (!completer.isCompleted) {
+            completer.complete();
+          }
+        }
+        await Future.wait(
+          closeCalledCompleters.map((completer) => completer.future),
+        );
+      },
+      (error, stackTrace) {
+        errors.add(error);
+      },
+    );
+
+    expect(errors, isEmpty);
+    expect(clients, hasLength(2));
+    verify(clients.first.close()).called(1);
+    verify(clients.last.close()).called(1);
+  });
 }