diff --git a/packages/connectivity_plus/connectivity_plus/lib/src/connectivity_plus_linux.dart b/packages/connectivity_plus/connectivity_plus/lib/src/connectivity_plus_linux.dart index 31b562aa24..3bfd83f454 100644 --- a/packages/connectivity_plus/connectivity_plus/lib/src/connectivity_plus_linux.dart +++ b/packages/connectivity_plus/connectivity_plus/lib/src/connectivity_plus_linux.dart @@ -21,13 +21,15 @@ class ConnectivityPlusLinuxPlugin extends ConnectivityPlatform { @override Future> 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>? _controller; /// Returns a Stream of ConnectivityResults changes. @@ -69,14 +71,31 @@ class ConnectivityPlusLinuxPlugin extends ConnectivityPlatform { } Future _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) { @@ -84,11 +103,37 @@ class ConnectivityPlusLinuxPlugin extends ConnectivityPlatform { } Future _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 connected; + StreamSubscription>? propertiesChangedSubscription; + + Future? _closeFuture; + + Future close() => _closeFuture ??= _close(); + + Future _close() async { + try { + await connected; + } catch (_) { + // Connection errors are reported by the listener setup. + } + try { + await propertiesChangedSubscription?.cancel(); + } finally { + await client.close(); + } + } +} diff --git a/packages/connectivity_plus/connectivity_plus/test/connectivity_plus_linux_test.dart b/packages/connectivity_plus/connectivity_plus/test/connectivity_plus_linux_test.dart index b7ecf03cab..48ea788dcd 100644 --- a/packages/connectivity_plus/connectivity_plus/test/connectivity_plus_linux_test.dart +++ b/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 @@ void main() { 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 @@ void main() { [ConnectivityResult.none] ])); }); + + test('canceling the last listener cancels the property subscription', + () async { + final linux = ConnectivityPlusLinuxPlugin(); + final client = MockNetworkManagerClient(); + final propertiesChanged = StreamController>.broadcast(); + final clientClosed = Completer(); + 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(); + 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 = []; + final connectCompleters = >[]; + final closeCompleters = >[]; + final closeCalledCompleters = >[]; + linux.createClient = () { + final client = MockNetworkManagerClient(); + final connectCompleter = Completer(); + final closeCompleter = Completer(); + final closeCalledCompleter = Completer(); + 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>.empty()); + clients.add(client); + connectCompleters.add(connectCompleter); + closeCompleters.add(closeCompleter); + closeCalledCompleters.add(closeCalledCompleter); + return client; + }; + + final errors = []; + 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); + }); }