From e6b699c9760522f6f9cbe060fd2033dff998f0df Mon Sep 17 00:00:00 2001 From: edde746 <86283021+edde746@users.noreply.github.com> Date: Sat, 15 Nov 2025 00:42:45 +0100 Subject: [PATCH] fix: better connection selection --- ios/Podfile.lock | 6 + lib/client/plex_client.dart | 64 ++- .../endpoint_failover_interceptor.dart | 177 ++++++++ lib/services/plex_auth_service.dart | 421 +++++++++++++++--- lib/services/server_connection_service.dart | 307 ++++++++++--- macos/Flutter/GeneratedPluginRegistrant.swift | 2 + pubspec.lock | 24 + pubspec.yaml | 1 + .../flutter/generated_plugin_registrant.cc | 3 + windows/flutter/generated_plugins.cmake | 1 + 10 files changed, 884 insertions(+), 122 deletions(-) create mode 100644 lib/network/endpoint_failover_interceptor.dart diff --git a/ios/Podfile.lock b/ios/Podfile.lock index 70cb3503..7f1d716c 100644 --- a/ios/Podfile.lock +++ b/ios/Podfile.lock @@ -1,4 +1,6 @@ PODS: + - connectivity_plus (0.0.1): + - Flutter - Flutter (1.0.0) - media_kit_libs_ios_video (1.0.4): - Flutter @@ -26,6 +28,7 @@ PODS: - Flutter DEPENDENCIES: + - connectivity_plus (from `.symlinks/plugins/connectivity_plus/ios`) - Flutter (from `Flutter`) - media_kit_libs_ios_video (from `.symlinks/plugins/media_kit_libs_ios_video/ios`) - media_kit_video (from `.symlinks/plugins/media_kit_video/ios`) @@ -39,6 +42,8 @@ DEPENDENCIES: - wakelock_plus (from `.symlinks/plugins/wakelock_plus/ios`) EXTERNAL SOURCES: + connectivity_plus: + :path: ".symlinks/plugins/connectivity_plus/ios" Flutter: :path: Flutter media_kit_libs_ios_video: @@ -63,6 +68,7 @@ EXTERNAL SOURCES: :path: ".symlinks/plugins/wakelock_plus/ios" SPEC CHECKSUMS: + connectivity_plus: cb623214f4e1f6ef8fe7403d580fdad517d2f7dd Flutter: cabc95a1d2626b1b06e7179b784ebcf0c0cde467 media_kit_libs_ios_video: 5a18affdb97d1f5d466dc79988b13eff6c5e2854 media_kit_video: 6235abf1d299037d23692ad47119b89e05346b23 diff --git a/lib/client/plex_client.dart b/lib/client/plex_client.dart index 83a6a930..b01e62f3 100644 --- a/lib/client/plex_client.dart +++ b/lib/client/plex_client.dart @@ -12,7 +12,9 @@ import '../models/plex_media_version.dart'; import '../models/plex_metadata.dart'; import '../models/plex_sort.dart'; import '../models/plex_video_playback_data.dart'; +import '../network/endpoint_failover_interceptor.dart'; import '../utils/app_logger.dart'; +import '../utils/log_redaction_manager.dart'; /// Result of testing a connection, including success status and latency class ConnectionTestResult { @@ -25,6 +27,8 @@ class ConnectionTestResult { class PlexClient { PlexConfig config; late final Dio _dio; + final EndpointFailoverManager? _endpointManager; + final Future Function(String newBaseUrl)? _onEndpointChanged; /// Custom response decoder that handles malformed UTF-8 gracefully static String _lenientUtf8Decoder( @@ -35,7 +39,18 @@ class PlexClient { return utf8.decode(responseBytes, allowMalformed: true); } - PlexClient(this.config) { + PlexClient( + this.config, { + List? prioritizedEndpoints, + Future Function(String newBaseUrl)? onEndpointChanged, + }) : _endpointManager = (prioritizedEndpoints != null && + prioritizedEndpoints.isNotEmpty) + ? EndpointFailoverManager(prioritizedEndpoints) + : null, + _onEndpointChanged = onEndpointChanged { + LogRedactionManager.registerServerUrl(config.baseUrl); + LogRedactionManager.registerToken(config.token); + _dio = Dio( BaseOptions( baseUrl: config.baseUrl, @@ -59,6 +74,16 @@ class PlexClient { responseHeader: false, ), ); + + if (_endpointManager != null) { + _dio.interceptors.add( + EndpointFailoverInterceptor( + dio: _dio, + endpointManager: _endpointManager, + onEndpointSwitch: _handleEndpointSwitch, + ), + ); + } } /// Update the token used by this client @@ -66,9 +91,31 @@ class PlexClient { // Update both the Dio headers and the config to ensure consistency _dio.options.headers['X-Plex-Token'] = newToken; config = config.copyWith(token: newToken); + LogRedactionManager.registerToken(newToken); appLogger.d('PlexClient token updated (headers and config)'); } + /// Update endpoint priority list and optionally hop to the new best endpoint. + Future updateEndpointPreferences( + List prioritizedEndpoints, { + bool switchToFirst = false, + }) async { + if (_endpointManager == null || prioritizedEndpoints.isEmpty) { + return; + } + + final targetBaseUrl = + switchToFirst ? prioritizedEndpoints.first : config.baseUrl; + _endpointManager.reset( + prioritizedEndpoints, + currentBaseUrl: targetBaseUrl, + ); + + if (switchToFirst && targetBaseUrl != config.baseUrl) { + await _handleEndpointSwitch(targetBaseUrl); + } + } + /// Test connection to server Future testConnection() async { try { @@ -1209,4 +1256,19 @@ class PlexClient { Future analyzeLibrary(String sectionId) async { await _dio.get('/library/sections/$sectionId/analyze'); } + + Future _handleEndpointSwitch(String newBaseUrl) async { + if (config.baseUrl == newBaseUrl) { + return; + } + + appLogger.i('Applying Plex endpoint switch', error: newBaseUrl); + _dio.options.baseUrl = newBaseUrl; + config = config.copyWith(baseUrl: newBaseUrl); + LogRedactionManager.registerServerUrl(newBaseUrl); + + if (_onEndpointChanged != null) { + await _onEndpointChanged(newBaseUrl); + } + } } diff --git a/lib/network/endpoint_failover_interceptor.dart b/lib/network/endpoint_failover_interceptor.dart new file mode 100644 index 00000000..71c683da --- /dev/null +++ b/lib/network/endpoint_failover_interceptor.dart @@ -0,0 +1,177 @@ +import 'package:dio/dio.dart'; + +import '../utils/app_logger.dart'; + +/// Maintains the list of endpoints we can cycle through when one fails. +class EndpointFailoverManager { + EndpointFailoverManager(List urls) { + _setEndpoints(urls); + } + + late List _endpoints; + int _currentIndex = 0; + + List get endpoints => List.unmodifiable(_endpoints); + + String get current => _endpoints[_currentIndex]; + + bool get hasFallback => _currentIndex < _endpoints.length - 1; + + /// Move to the next endpoint, returning its URL or null if exhausted. + String? moveToNext() { + if (!hasFallback) return null; + _currentIndex++; + return _endpoints[_currentIndex]; + } + + /// Replace the endpoint list and optionally set the active endpoint. + void reset( + List urls, { + String? currentBaseUrl, + }) { + _setEndpoints(urls); + if (currentBaseUrl != null) { + final index = _endpoints.indexOf(currentBaseUrl); + _currentIndex = index >= 0 ? index : 0; + } else { + _currentIndex = 0; + } + } + + void _setEndpoints(List urls) { + final sanitized = []; + final seen = {}; + for (final url in urls) { + if (url.isEmpty || seen.contains(url)) continue; + seen.add(url); + sanitized.add(url); + } + if (sanitized.isEmpty) { + throw ArgumentError('At least one endpoint is required'); + } + _endpoints = sanitized; + _currentIndex = _currentIndex.clamp(0, _endpoints.length - 1); + } +} + +/// Dio interceptor that retries failed requests on the next available endpoint. +class EndpointFailoverInterceptor extends Interceptor { + EndpointFailoverInterceptor({ + required Dio dio, + required this.endpointManager, + required Future Function(String newBaseUrl) onEndpointSwitch, + }) : _dio = dio, + _onEndpointSwitch = onEndpointSwitch; + + final Dio _dio; + final EndpointFailoverManager endpointManager; + final Future Function(String newBaseUrl) _onEndpointSwitch; + bool _isSwitching = false; + + @override + void onError(DioException err, ErrorInterceptorHandler handler) async { + if (_isSwitching || + !_shouldAttemptFailover(err) || + !endpointManager.hasFallback) { + handler.next(err); + return; + } + + final failedEndpoint = endpointManager.current; + appLogger.w( + 'Endpoint request failed, evaluating failover', + error: { + 'endpoint': failedEndpoint, + 'type': err.type.name, + 'statusCode': err.response?.statusCode, + }, + stackTrace: err.stackTrace, + ); + + final nextBaseUrl = endpointManager.moveToNext(); + if (nextBaseUrl == null) { + appLogger.w( + 'Endpoint failure but no fallback endpoints remain', + error: {'failedEndpoint': failedEndpoint}, + ); + handler.next(err); + return; + } + + _isSwitching = true; + try { + appLogger.i( + 'Switching Plex endpoint after request failure', + error: { + 'from': failedEndpoint, + 'to': nextBaseUrl, + 'path': err.requestOptions.path, + }, + ); + await _onEndpointSwitch(nextBaseUrl); + final response = await _retryRequest(err.requestOptions); + appLogger.i( + 'Endpoint failover retry succeeded', + error: {'newEndpoint': nextBaseUrl}, + ); + handler.resolve(response); + } on DioException catch (dioError) { + appLogger.w( + 'Endpoint failover retry failed', + error: { + 'newEndpoint': nextBaseUrl, + 'type': dioError.type.name, + 'statusCode': dioError.response?.statusCode, + }, + stackTrace: dioError.stackTrace, + ); + handler.next(dioError); + } catch (_) { + handler.next(err); + } finally { + _isSwitching = false; + } + } + + bool _shouldAttemptFailover(DioException error) { + if (error.type == DioExceptionType.connectionTimeout || + error.type == DioExceptionType.receiveTimeout || + error.type == DioExceptionType.sendTimeout || + error.type == DioExceptionType.connectionError) { + return true; + } + + if (error.type == DioExceptionType.badResponse) { + final statusCode = error.response?.statusCode ?? 0; + return statusCode >= 500; + } + + return false; + } + + Future> _retryRequest(RequestOptions requestOptions) { + final options = Options( + method: requestOptions.method, + headers: requestOptions.headers, + responseType: requestOptions.responseType, + contentType: requestOptions.contentType, + followRedirects: requestOptions.followRedirects, + receiveDataWhenStatusError: requestOptions.receiveDataWhenStatusError, + validateStatus: requestOptions.validateStatus, + sendTimeout: requestOptions.sendTimeout, + receiveTimeout: requestOptions.receiveTimeout, + extra: requestOptions.extra, + listFormat: requestOptions.listFormat, + ); + + return _dio.request( + requestOptions.path, + data: requestOptions.data, + queryParameters: requestOptions.queryParameters, + options: options, + cancelToken: requestOptions.cancelToken, + onSendProgress: requestOptions.onSendProgress, + onReceiveProgress: requestOptions.onReceiveProgress, + ); + } +} diff --git a/lib/services/plex_auth_service.dart b/lib/services/plex_auth_service.dart index 7c6ec57c..a380cc9c 100644 --- a/lib/services/plex_auth_service.dart +++ b/lib/services/plex_auth_service.dart @@ -6,6 +6,7 @@ import '../client/plex_client.dart'; import '../models/plex_user_profile.dart'; import '../models/plex_home.dart'; import '../models/user_switch_response.dart'; +import '../utils/app_logger.dart'; class PlexAuthService { static const String _appName = 'Plezy'; @@ -240,8 +241,14 @@ class _ConnectionCandidate { final PlexConnection connection; final String url; final bool isPlexDirectUri; + final bool isHttps; - _ConnectionCandidate(this.connection, this.url, this.isPlexDirectUri); + _ConnectionCandidate( + this.connection, + this.url, + this.isPlexDirectUri, + this.isHttps, + ); } /// Represents a Plex Media Server @@ -391,74 +398,116 @@ class PlexServer { /// Priority: local > remote > relay, then HTTPS > HTTP, then lowest latency /// Tests both plex.direct URI and direct IP for each connection /// HTTPS connections are tested first, with HTTP as fallback - Stream findBestWorkingConnection() async* { - if (connections.isEmpty) return; + Stream findBestWorkingConnection({ + String? preferredUri, + }) async* { + if (connections.isEmpty) { + appLogger.w('No connections available for server discovery'); + return; + } - // Create candidates: test both uri and directUrl for each connection - // Separate HTTPS and HTTP candidates to prioritize HTTPS first - final httpsCandidates = <_ConnectionCandidate>[]; - final httpCandidates = <_ConnectionCandidate>[]; + const preferredTimeout = Duration(seconds: 2); + const raceTimeout = Duration(seconds: 4); - for (final connection in connections) { - final uriCandidate = _ConnectionCandidate( - connection, - connection.uri, - true, - ); - final directCandidate = _ConnectionCandidate( - connection, - connection.directUrl, - false, - ); + final candidates = _buildPrioritizedCandidates(); + if (candidates.isEmpty) { + appLogger.w('No connection candidates generated for server discovery'); + return; + } - if (connection.protocol == 'https') { - httpsCandidates.add(uriCandidate); - httpsCandidates.add(directCandidate); - } else { - httpCandidates.add(uriCandidate); - httpCandidates.add(directCandidate); + final totalCandidates = candidates.length; + appLogger.d( + 'Starting server connection discovery', + error: { + 'preferred': preferredUri, + 'candidateCount': totalCandidates, + }, + ); + + _ConnectionCandidate? firstCandidate; + + // Fast-path: if we have a cached working URI, probe it with a short timeout + if (preferredUri != null) { + final cachedCandidate = _candidateForUrl(preferredUri); + if (cachedCandidate != null) { + appLogger.d( + 'Testing cached endpoint before running full race', + error: {'uri': preferredUri}, + ); + final result = await PlexClient.testConnectionWithLatency( + cachedCandidate.url, + accessToken, + timeout: preferredTimeout, + ); + + if (result.success) { + appLogger.i( + 'Cached endpoint succeeded, using immediately', + error: {'uri': preferredUri}, + ); + firstCandidate = cachedCandidate; + } else { + appLogger.w( + 'Cached endpoint failed, falling back to candidate race', + error: {'uri': preferredUri}, + ); + } } } - // Combine candidates with HTTPS first, then HTTP - final candidates = [...httpsCandidates, ...httpCandidates]; - - // Phase 1: Race to find first working connection - final completer = Completer<_ConnectionCandidate?>(); - _ConnectionCandidate? firstCandidate; - int completedTests = 0; - - // Start testing all candidates simultaneously - for (final candidate in candidates) { - PlexClient.testConnectionWithLatency(candidate.url, accessToken).then(( - result, - ) { - completedTests++; - - // If this is the first successful connection, emit it immediately - if (result.success && !completer.isCompleted) { - completer.complete(candidate); - } - - // If all tests complete without success, complete with null - if (completedTests == candidates.length && !completer.isCompleted) { - completer.complete(null); - } - }); - } - - // Wait for and emit the first successful connection - firstCandidate = await completer.future; + // If no cached candidate or it failed, race candidates to find first success if (firstCandidate == null) { - return; // No working connections found + final completer = Completer<_ConnectionCandidate?>(); + int completedTests = 0; + + appLogger.d( + 'Running connection race to find first working endpoint', + error: {'candidateCount': totalCandidates}, + ); + + for (final candidate in candidates) { + PlexClient + .testConnectionWithLatency( + candidate.url, + accessToken, + timeout: raceTimeout, + ) + .then((result) { + completedTests++; + + if (result.success && !completer.isCompleted) { + completer.complete(candidate); + } + + if (completedTests == candidates.length && !completer.isCompleted) { + completer.complete(null); + } + }); + } + + firstCandidate = await completer.future; + if (firstCandidate == null) { + appLogger.e('No working server connections after race'); + return; // No working connections found + } + appLogger.i( + 'Connection race found first working endpoint', + error: { + 'uri': firstCandidate.url, + 'type': firstCandidate.connection.displayType, + }, + ); } - // Update the connection object to use the working URL final firstConnection = _updateConnectionUrl( firstCandidate.connection, firstCandidate.url, ); yield firstConnection; + appLogger.d( + 'Emitted first working connection, continuing latency tests in background', + error: {'uri': firstConnection.uri}, + ); // Phase 2: Continue testing in background to find best connection // Test each candidate 2-3 times and average the latency @@ -480,20 +529,39 @@ class PlexServer { // If no connections succeeded, we're done if (candidateResults.isEmpty) { + appLogger.w('Latency sweep found no additional working endpoints'); return; } + appLogger.d( + 'Completed latency sweep for server connections', + error: {'successfulCandidates': candidateResults.length}, + ); + // Find the best connection considering priority, latency, and URL type final bestCandidate = _selectBestCandidateWithLatency(candidateResults); // Emit the best connection if it's different from the first one if (bestCandidate != null) { + final upgradedCandidate = + await _upgradeCandidateToHttpsIfPossible(bestCandidate) ?? + bestCandidate; + final bestConnection = _updateConnectionUrl( - bestCandidate.connection, - bestCandidate.url, + upgradedCandidate.connection, + upgradedCandidate.url, ); if (bestConnection.uri != firstConnection.uri) { + appLogger.i( + 'Latency sweep selected better endpoint', + error: {'uri': bestConnection.uri}, + ); yield bestConnection; + } else { + appLogger.d( + 'Latency sweep confirmed initial endpoint is optimal', + error: {'uri': bestConnection.uri}, + ); } } } @@ -517,6 +585,234 @@ class PlexServer { ); } + _ConnectionCandidate? _candidateForUrl(String url) { + for (final connection in connections) { + final httpUrl = connection.httpDirectUrl; + if (httpUrl == url) { + return _ConnectionCandidate(connection, httpUrl, false, false); + } + + final uri = connection.uri; + if (uri == url) { + final isHttps = uri.startsWith('https://'); + final parsedHost = Uri.tryParse(uri)?.host ?? ''; + final isPlexDirect = + parsedHost.toLowerCase().contains('plex.direct'); + return _ConnectionCandidate(connection, uri, isPlexDirect, isHttps); + } + } + return null; + } + + List<_ConnectionCandidate> _buildPrioritizedCandidates({ + Set? excludeUrls, + }) { + final seen = {}; + if (excludeUrls != null) { + seen.addAll(excludeUrls); + } + + final httpsLocal = <_ConnectionCandidate>[]; + final httpsRemote = <_ConnectionCandidate>[]; + final httpsRelay = <_ConnectionCandidate>[]; + final httpLocal = <_ConnectionCandidate>[]; + final httpRemote = <_ConnectionCandidate>[]; + final httpRelay = <_ConnectionCandidate>[]; + + List<_ConnectionCandidate> bucketFor( + PlexConnection connection, + bool isHttps, + ) { + if (isHttps) { + if (connection.relay) return httpsRelay; + if (connection.local) return httpsLocal; + return httpsRemote; + } else { + if (connection.relay) return httpRelay; + if (connection.local) return httpLocal; + return httpRemote; + } + } + + void addCandidate( + PlexConnection connection, + String url, + bool isPlexDirectUri, + bool isHttps, + ) { + if (url.isEmpty || seen.contains(url)) { + return; + } + seen.add(url); + bucketFor(connection, isHttps).add( + _ConnectionCandidate(connection, url, isPlexDirectUri, isHttps), + ); + } + + for (final connection in connections) { + addCandidate( + connection, + connection.httpDirectUrl, + false, + false, + ); + } + + return [ + ...httpsLocal, + ...httpsRemote, + ...httpsRelay, + ...httpLocal, + ...httpRemote, + ...httpRelay, + ]; + } + + List prioritizedEndpointUrls({String? preferredFirst}) { + final urls = []; + final exclude = {}; + + if (preferredFirst != null && preferredFirst.isNotEmpty) { + urls.add(preferredFirst); + exclude.add(preferredFirst); + } + + final candidates = _buildPrioritizedCandidates(excludeUrls: exclude); + urls.addAll(candidates.map((candidate) => candidate.url)); + return urls; + } + + Future<_ConnectionCandidate?> _upgradeCandidateToHttpsIfPossible( + _ConnectionCandidate candidate, + ) async { + final currentUrl = candidate.url; + if (currentUrl.startsWith('https://')) { + return null; + } + + late final String httpsUrl; + bool resultingIsPlexDirect = candidate.isPlexDirectUri; + + if (candidate.isPlexDirectUri) { + if (!currentUrl.startsWith('http://')) { + return null; + } + httpsUrl = currentUrl.replaceFirst('http://', 'https://'); + } else { + // Raw IP endpoints can't present HTTPS certificates—prefer their plex.direct alias. + final plexDirectUri = candidate.connection.uri; + if (plexDirectUri.isEmpty) { + return null; + } + + if (plexDirectUri.startsWith('https://')) { + httpsUrl = plexDirectUri; + } else if (plexDirectUri.startsWith('http://')) { + httpsUrl = plexDirectUri.replaceFirst('http://', 'https://'); + } else { + return null; + } + + final upgradedHost = Uri.tryParse(httpsUrl)?.host; + if (upgradedHost == null || + !upgradedHost.toLowerCase().endsWith('.plex.direct')) { + appLogger.d( + 'Skipping HTTPS upgrade for raw IP candidate: no plex.direct alias available', + error: {'candidate': currentUrl, 'target': httpsUrl}, + ); + return null; + } + resultingIsPlexDirect = true; + } + + if (httpsUrl == currentUrl) { + return null; + } + + appLogger.d( + 'Attempting HTTPS upgrade for candidate endpoint', + error: {'from': currentUrl, 'to': httpsUrl}, + ); + + final result = await PlexClient.testConnectionWithLatency( + httpsUrl, + accessToken, + timeout: const Duration(seconds: 4), + ); + + if (!result.success) { + appLogger.w( + 'HTTPS upgrade failed, staying on HTTP candidate', + error: {'url': currentUrl}, + ); + return null; + } + + appLogger.i( + 'HTTPS upgrade succeeded for candidate endpoint', + error: {'httpsUrl': httpsUrl}, + ); + + final httpsConnection = PlexConnection( + protocol: 'https', + address: candidate.connection.address, + port: candidate.connection.port, + uri: httpsUrl, + local: candidate.connection.local, + relay: candidate.connection.relay, + ipv6: candidate.connection.ipv6, + ); + + return _ConnectionCandidate( + httpsConnection, + httpsUrl, + resultingIsPlexDirect, + true, + ); + } + + Future upgradeConnectionToHttps( + PlexConnection current, + ) async { + if (current.uri.startsWith('https://')) { + return current; + } + + final baseConnection = _findMatchingBaseConnection(current); + if (baseConnection == null) { + return null; + } + + final candidate = _ConnectionCandidate( + baseConnection, + current.uri, + current.uri.contains('.plex.direct'), + current.uri.startsWith('https://'), + ); + final upgradedCandidate = + await _upgradeCandidateToHttpsIfPossible(candidate); + if (upgradedCandidate == null) { + return null; + } + return _updateConnectionUrl( + upgradedCandidate.connection, + upgradedCandidate.url, + ); + } + + PlexConnection? _findMatchingBaseConnection(PlexConnection connection) { + for (final base in connections) { + final sameAddress = base.address == connection.address; + final samePort = base.port == connection.port; + final sameLocal = base.local == connection.local; + final sameRelay = base.relay == connection.relay; + if (sameAddress && samePort && sameLocal && sameRelay) { + return base; + } + } + return null; + } + /// Select the best candidate considering priority, latency, and URL type preference _ConnectionCandidate? _selectBestCandidateWithLatency( Map<_ConnectionCandidate, ConnectionTestResult> results, @@ -550,8 +846,8 @@ class PlexServer { if (latencyCompare != 0) return latencyCompare; // If latencies are equal, prefer HTTPS over HTTP - final aIsHttps = a.key.connection.protocol == 'https'; - final bIsHttps = b.key.connection.protocol == 'https'; + final aIsHttps = a.key.isHttps; + final bIsHttps = b.key.isHttps; if (aIsHttps && !bIsHttps) return -1; if (!aIsHttps && bIsHttps) return 1; @@ -641,6 +937,13 @@ class PlexConnection { /// This bypasses plex.direct DNS and connects directly to the IP String get directUrl => '$protocol://$address:$port'; + /// Always return an HTTP URL that points directly at the IP/port combo. + String get httpDirectUrl { + final needsBrackets = address.contains(':') && !address.startsWith('['); + final safeAddress = needsBrackets ? '[$address]' : address; + return 'http://$safeAddress:$port'; + } + String get displayType { if (relay) return 'Relay'; if (local) return 'Local'; diff --git a/lib/services/server_connection_service.dart b/lib/services/server_connection_service.dart index 32569b25..5a3dbea2 100644 --- a/lib/services/server_connection_service.dart +++ b/lib/services/server_connection_service.dart @@ -1,3 +1,7 @@ +import 'dart:async'; + +import 'package:connectivity_plus/connectivity_plus.dart'; + import 'plex_auth_service.dart'; import 'storage_service.dart'; import '../client/plex_client.dart'; @@ -19,6 +23,12 @@ class ServerConnectionResult { /// Service for handling optimized server connections /// Implements fast-first connection with background optimization class ServerConnectionService { + static StreamSubscription>? + _connectivitySubscription; + static Future? _activeOptimization; + static PlexServer? _activeServer; + static PlexClient? _activeClient; + /// Connect to a Plex server with optimized connection testing /// /// Returns immediately with first working connection, then continues @@ -39,86 +49,115 @@ class ServerConnectionService { bool fetchUserProfile = false, void Function(String message)? onProgress, }) async { - PlexConnection? firstConnection; final storage = await StorageService.getInstance(); + final connectionStream = + server.findBestWorkingConnection().asBroadcastStream(); + PlexClient? client; + + final optimizationSubscription = connectionStream.skip(1).listen( + (connection) async { + await _handleOptimizedConnection( + connection: connection, + storage: storage, + server: server, + client: client, + reason: 'initial_latency_sweep', + ); + }, + onError: (error, stackTrace) { + appLogger.w( + 'Background connection optimization error', + error: error, + stackTrace: stackTrace, + ); + }, + ); + try { - // Listen to the connection stream for progressive connection testing - await for (final connection in server.findBestWorkingConnection()) { - if (firstConnection == null) { - // First emission - use this connection immediately - firstConnection = connection; + final connection = await connectionStream.first; - if (onProgress != null) { - onProgress('Connected to ${connection.displayType} endpoint'); - } + if (onProgress != null) { + onProgress('Connected to ${connection.displayType} endpoint'); + } - // Save server information to storage - await storage.saveServerData(server.toJson()); - await storage.saveServerUrl(connection.uri); - await storage.saveServerAccessToken(server.accessToken); + // Save server information to storage + await storage.saveServerData(server.toJson()); + await storage.saveServerUrl(connection.uri); + await storage.saveServerAccessToken(server.accessToken); - // Save plex token if provided - if (plexToken != null) { - await storage.savePlexToken(plexToken); - } + // Save plex token if provided + if (plexToken != null) { + await storage.savePlexToken(plexToken); + } - // Create client with working connection - final config = await PlexConfig.create( - baseUrl: connection.uri, - token: server.accessToken, - clientIdentifier: clientIdentifier, + // Create client with working connection + final prioritizedEndpoints = server.prioritizedEndpointUrls( + preferredFirst: connection.uri, + ); + final config = await PlexConfig.create( + baseUrl: connection.uri, + token: server.accessToken, + clientIdentifier: clientIdentifier, + ); + client = PlexClient( + config, + prioritizedEndpoints: prioritizedEndpoints, + onEndpointChanged: (newUrl) async { + await storage.saveServerUrl(newUrl); + appLogger.i( + 'Updated stored server URL after failover', + error: newUrl, ); - final client = PlexClient(config); + }, + ); - // Verify server is accessible if requested - if (verifyServer) { - try { - await client.getServerIdentity(); - } catch (e) { - appLogger.w('Server identity verification failed', error: e); - await storage.clearCredentials(); - return ServerConnectionResult( - error: 'Server is not accessible: $e', - ); - } - } - - // Fetch user profile if requested - PlexUserProfile? userProfile; - if (fetchUserProfile && plexToken != null) { - userProfile = await _fetchUserProfile(plexToken); - } - - // Return success result - // Note: Stream continues in background to find better connection + // Verify server is accessible if requested + if (verifyServer) { + try { + await client.getServerIdentity(); + } catch (e) { + await optimizationSubscription.cancel(); + appLogger.w('Server identity verification failed', error: e); + await storage.clearCredentials(); return ServerConnectionResult( - client: client, - userProfile: userProfile, - ); - } else { - // Second emission - better connection found - // Update stored connection seamlessly for future app launches - await storage.saveServerUrl(connection.uri); - appLogger.d( - 'Switched to better connection: ${connection.displayType} (${connection.uri})', + error: 'Server is not accessible: $e', ); } } - // Handle case where no connections were found - if (firstConnection == null) { - return ServerConnectionResult( - error: 'No working connections found for this server', - ); + // Fetch user profile if requested + PlexUserProfile? userProfile; + if (fetchUserProfile && plexToken != null) { + userProfile = await _fetchUserProfile(plexToken); } - // Should never reach here due to return in the stream loop + // Return success result while optimization continues in background + _activeServer = server; + _activeClient = client; + _startConnectivityMonitoring(server); + return ServerConnectionResult( - error: 'Unexpected error in connection flow', + client: client, + userProfile: userProfile, + ); + } on StateError catch (e, stackTrace) { + await optimizationSubscription.cancel(); + appLogger.e( + 'No working connections found for this server', + error: e, + stackTrace: stackTrace, + ); + return ServerConnectionResult( + error: 'No working connections found for this server', + ); + } catch (e, stackTrace) { + await optimizationSubscription.cancel(); + appLogger.e( + 'Error connecting to server', + error: e, + stackTrace: stackTrace, ); - } catch (e) { - appLogger.e('Error connecting to server', error: e); return ServerConnectionResult(error: 'Connection failed: $e'); } } @@ -148,4 +187,148 @@ class ServerConnectionService { return null; } } + + static void _startConnectivityMonitoring(PlexServer server) { + _connectivitySubscription?.cancel(); + final connectivity = Connectivity(); + _connectivitySubscription = connectivity.onConnectivityChanged.listen( + (results) { + final status = + results.isNotEmpty ? results.first : ConnectivityResult.none; + if (status == ConnectivityResult.none) { + appLogger.w( + 'Connectivity lost, pausing optimization until network returns', + ); + return; + } + appLogger.d( + 'Connectivity change detected, triggering endpoint optimization', + error: { + 'status': status.name, + 'interfaces': results.map((r) => r.name).toList(), + }, + ); + _activeServer = server; + _triggerReoptimization(reason: 'connectivity:${status.name}'); + }, + onError: (error, stackTrace) { + appLogger.w( + 'Connectivity listener error', + error: error, + stackTrace: stackTrace, + ); + }, + ); + } + + static void _triggerReoptimization({required String reason}) { + if (_activeServer == null) { + appLogger.d( + 'Optimization trigger ignored because there is no active server', + error: {'reason': reason}, + ); + return; + } + + if (_activeOptimization != null) { + appLogger.d( + 'Optimization already running, skipping new trigger', + error: {'reason': reason}, + ); + return; + } + + _activeOptimization = _runOptimization( + server: _activeServer!, + client: _activeClient, + reason: reason, + ).whenComplete(() { + _activeOptimization = null; + }); + } + + static Future _runOptimization({ + required PlexServer server, + required PlexClient? client, + required String reason, + }) async { + final storage = await StorageService.getInstance(); + try { + appLogger.d( + 'Starting background connection optimization run', + error: {'reason': reason}, + ); + await for (final connection in server.findBestWorkingConnection()) { + await _handleOptimizedConnection( + connection: connection, + storage: storage, + server: server, + client: client, + reason: reason, + ); + } + } catch (e, stackTrace) { + appLogger.w( + 'Background connection optimization failed', + error: e, + stackTrace: stackTrace, + ); + } + } + + static Future _handleOptimizedConnection({ + required PlexConnection connection, + required StorageService storage, + required PlexServer server, + required PlexClient? client, + required String reason, + }) async { + final previousUrl = storage.getServerUrl(); + final isNewEndpoint = previousUrl != connection.uri; + + await storage.saveServerUrl(connection.uri); + appLogger.d( + 'Evaluated optimized endpoint candidate', + error: { + 'uri': connection.uri, + 'displayType': connection.displayType, + 'reason': reason, + 'isNewEndpoint': isNewEndpoint, + }, + ); + + if (client != null) { + final prioritizedEndpoints = server.prioritizedEndpointUrls( + preferredFirst: connection.uri, + ); + await client.updateEndpointPreferences( + prioritizedEndpoints, + switchToFirst: isNewEndpoint, + ); + if (isNewEndpoint) { + appLogger.i( + 'Active client switched to optimized endpoint', + error: {'uri': connection.uri, 'reason': reason}, + ); + } + } else if (isNewEndpoint) { + appLogger.i( + 'Stored optimized endpoint for future sessions', + error: {'uri': connection.uri, 'reason': reason}, + ); + } + + if (isNewEndpoint && !connection.uri.startsWith('https://')) { + final upgraded = await server.upgradeConnectionToHttps(connection); + if (upgraded != null && upgraded.uri != connection.uri) { + await _handleOptimizedConnection( + connection: upgraded, + storage: storage, + server: server, + client: client, + reason: '$reason:https-upgrade', + ); + } + } + } } diff --git a/macos/Flutter/GeneratedPluginRegistrant.swift b/macos/Flutter/GeneratedPluginRegistrant.swift index aa156acf..a6e9b89c 100644 --- a/macos/Flutter/GeneratedPluginRegistrant.swift +++ b/macos/Flutter/GeneratedPluginRegistrant.swift @@ -5,6 +5,7 @@ import FlutterMacOS import Foundation +import connectivity_plus import hotkey_manager_macos import macos_window_utils import media_kit_libs_macos_video @@ -21,6 +22,7 @@ import wakelock_plus import window_manager func RegisterGeneratedPlugins(registry: FlutterPluginRegistry) { + ConnectivityPlusPlugin.register(with: registry.registrar(forPlugin: "ConnectivityPlusPlugin")) HotkeyManagerMacosPlugin.register(with: registry.registrar(forPlugin: "HotkeyManagerMacosPlugin")) MacOSWindowUtilsPlugin.register(with: registry.registrar(forPlugin: "MacOSWindowUtilsPlugin")) MediaKitLibsMacosVideoPlugin.register(with: registry.registrar(forPlugin: "MediaKitLibsMacosVideoPlugin")) diff --git a/pubspec.lock b/pubspec.lock index 106c8339..5f91f49b 100644 --- a/pubspec.lock +++ b/pubspec.lock @@ -185,6 +185,22 @@ packages: url: "https://pub.dev" source: hosted version: "1.19.1" + connectivity_plus: + dependency: "direct main" + description: + name: connectivity_plus + sha256: b5e72753cf63becce2c61fd04dfe0f1c430cc5278b53a1342dc5ad839eab29ec + url: "https://pub.dev" + source: hosted + version: "6.1.5" + connectivity_plus_platform_interface: + dependency: transitive + description: + name: connectivity_plus_platform_interface + sha256: "42657c1715d48b167930d5f34d00222ac100475f73d10162ddf43e714932f204" + url: "https://pub.dev" + source: hosted + version: "2.0.1" convert: dependency: transitive description: @@ -632,6 +648,14 @@ packages: url: "https://pub.dev" source: hosted version: "1.0.0" + nm: + dependency: transitive + description: + name: nm + sha256: "2c9aae4127bdc8993206464fcc063611e0e36e72018696cd9631023a31b24254" + url: "https://pub.dev" + source: hosted + version: "0.5.0" octo_image: dependency: transitive description: diff --git a/pubspec.yaml b/pubspec.yaml index 5e3b7f22..31042248 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -28,6 +28,7 @@ dependencies: qr_flutter: ^4.1.0 slang: ^3.31.2 slang_flutter: ^3.31.0 + connectivity_plus: ^6.0.5 os_media_controls: git: url: https://github.com/edde746/os-media-controls diff --git a/windows/flutter/generated_plugin_registrant.cc b/windows/flutter/generated_plugin_registrant.cc index 4e0884f2..e59aa9c2 100644 --- a/windows/flutter/generated_plugin_registrant.cc +++ b/windows/flutter/generated_plugin_registrant.cc @@ -6,6 +6,7 @@ #include "generated_plugin_registrant.h" +#include #include #include #include @@ -16,6 +17,8 @@ #include void RegisterPlugins(flutter::PluginRegistry* registry) { + ConnectivityPlusWindowsPluginRegisterWithRegistrar( + registry->GetRegistrarForPlugin("ConnectivityPlusWindowsPlugin")); HotkeyManagerWindowsPluginCApiRegisterWithRegistrar( registry->GetRegistrarForPlugin("HotkeyManagerWindowsPluginCApi")); MediaKitLibsWindowsVideoPluginCApiRegisterWithRegistrar( diff --git a/windows/flutter/generated_plugins.cmake b/windows/flutter/generated_plugins.cmake index adfebedf..0c647b45 100644 --- a/windows/flutter/generated_plugins.cmake +++ b/windows/flutter/generated_plugins.cmake @@ -3,6 +3,7 @@ # list(APPEND FLUTTER_PLUGIN_LIST + connectivity_plus hotkey_manager_windows media_kit_libs_windows_video media_kit_video