From 33e977872e0a319372096033589e010a46436f81 Mon Sep 17 00:00:00 2001 From: edde746 <86283021+edde746@users.noreply.github.com> Date: Tue, 27 Jan 2026 17:45:10 +0100 Subject: [PATCH] fix: consume full connection stream and upgrade to HTTPS early --- lib/services/multi_server_manager.dart | 78 ++++++++++++++++++++------ lib/services/plex_auth_service.dart | 12 +++- 2 files changed, 72 insertions(+), 18 deletions(-) diff --git a/lib/services/multi_server_manager.dart b/lib/services/multi_server_manager.dart index 469920a3..53829cef 100644 --- a/lib/services/multi_server_manager.dart +++ b/lib/services/multi_server_manager.dart @@ -71,23 +71,20 @@ class MultiServerManager { Future _createClientForServer({required PlexServer server, required String clientIdentifier}) async { final serverId = server.clientIdentifier; - // Find best working connection - PlexConnection? workingConnection; - await for (final connection in server.findBestWorkingConnection()) { - workingConnection = connection; - break; - } - - if (workingConnection == null) { - throw Exception('No working connection found'); - } - - final baseUrl = workingConnection.uri; - // Get storage and load cached endpoint for this server final storage = await StorageService.getInstance(); final cachedEndpoint = storage.getServerEndpoint(serverId); + // Find best working connection, passing cached endpoint for fast-path + final streamIterator = StreamIterator(server.findBestWorkingConnection(preferredUri: cachedEndpoint)); + + if (!await streamIterator.moveNext()) { + throw Exception('No working connection found'); + } + + final workingConnection = streamIterator.current; + final baseUrl = workingConnection.uri; + // Create PlexClient with failover support final prioritizedEndpoints = server.prioritizedEndpointUrls(preferredFirst: cachedEndpoint ?? baseUrl); final config = await PlexConfig.create( @@ -110,9 +107,50 @@ class MultiServerManager { // Save the initial endpoint await storage.saveServerEndpoint(serverId, baseUrl); + // Drain remaining stream values in background to apply better connections + _drainOptimizationStream(streamIterator, client: client, server: server, storage: storage); + return client; } + /// Continues draining the connection optimization stream in the background, + /// switching the client to any better endpoint found. + void _drainOptimizationStream( + StreamIterator streamIterator, { + required PlexClient client, + required PlexServer server, + required StorageService storage, + }) { + final serverId = server.clientIdentifier; + + () async { + try { + while (await streamIterator.moveNext()) { + final connection = streamIterator.current; + final newUrl = connection.uri; + + if (newUrl == client.config.baseUrl) { + appLogger.d('Background optimization confirmed current endpoint for ${server.name}'); + continue; + } + + appLogger.i( + 'Background optimization found better endpoint for ${server.name}', + error: {'from': client.config.baseUrl, 'to': newUrl, 'type': connection.displayType}, + ); + + await storage.saveServerEndpoint(serverId, newUrl); + final newEndpoints = server.prioritizedEndpointUrls(preferredFirst: newUrl); + await client.updateEndpointPreferences(newEndpoints, switchToFirst: true); + } + } catch (e, stackTrace) { + appLogger.w('Background connection optimization failed for ${server.name}', error: e, stackTrace: stackTrace); + } finally { + await streamIterator.cancel(); + } + }(); + } + /// Connect to all available servers in parallel /// Returns the number of successfully connected servers Future connectToAllServers( @@ -333,11 +371,12 @@ class MultiServerManager { Future _reoptimizeServer({required String serverId, required PlexServer server, required String reason}) async { final storage = await StorageService.getInstance(); final client = _clients[serverId]; + final cachedEndpoint = storage.getServerEndpoint(serverId); try { appLogger.d('Starting connection optimization for ${server.name}', error: {'reason': reason}); - await for (final connection in server.findBestWorkingConnection()) { + await for (final connection in server.findBestWorkingConnection(preferredUri: cachedEndpoint)) { final newUrl = connection.uri; // Check if this is actually a better connection than current @@ -349,9 +388,14 @@ class MultiServerManager { // Save the new endpoint await storage.saveServerEndpoint(serverId, newUrl); - // If client has endpoint failover, it will automatically switch - // Otherwise, we might need to recreate the client (but failover should handle it) - appLogger.i('Updated optimal endpoint for ${server.name}: $newUrl', error: {'type': connection.displayType}); + // Actively switch the running client to the better endpoint + if (client != null) { + final newEndpoints = server.prioritizedEndpointUrls(preferredFirst: newUrl); + await client.updateEndpointPreferences(newEndpoints, switchToFirst: true); + appLogger.i('Switched ${server.name} to better endpoint: $newUrl', error: {'type': connection.displayType}); + } else { + appLogger.i('Updated optimal endpoint for ${server.name}: $newUrl', error: {'type': connection.displayType}); + } } } catch (e, stackTrace) { appLogger.w('Connection optimization failed for ${server.name}', error: e, stackTrace: stackTrace); diff --git a/lib/services/plex_auth_service.dart b/lib/services/plex_auth_service.dart index 984ee17b..fc069396 100644 --- a/lib/services/plex_auth_service.dart +++ b/lib/services/plex_auth_service.dart @@ -433,8 +433,18 @@ class PlexServer { ); } - final firstConnection = _updateConnectionUrl(firstCandidate.connection, firstCandidate.url); + // Attempt HTTPS upgrade on the Phase 1 winner before emitting + final upgradedFirstCandidate = await _upgradeCandidateToHttpsIfPossible(firstCandidate); + final emitCandidate = upgradedFirstCandidate ?? firstCandidate; + + final firstConnection = _updateConnectionUrl(emitCandidate.connection, emitCandidate.url); yield firstConnection; + if (upgradedFirstCandidate != null && upgradedFirstCandidate.url != firstCandidate.url) { + appLogger.i( + 'Phase 1 winner upgraded to HTTPS', + error: {'from': firstCandidate.url, 'to': upgradedFirstCandidate.url}, + ); + } appLogger.d( 'Emitted first working connection, continuing latency tests in background', error: {'uri': firstConnection.uri},