fix: better connection selection

This commit is contained in:
edde746
2025-11-15 00:42:45 +01:00
parent 831552d5b8
commit e6b699c976
10 changed files with 884 additions and 122 deletions
+6
View File
@@ -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
+63 -1
View File
@@ -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<void> 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<String>? prioritizedEndpoints,
Future<void> 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<void> updateEndpointPreferences(
List<String> 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<bool> testConnection() async {
try {
@@ -1209,4 +1256,19 @@ class PlexClient {
Future<void> analyzeLibrary(String sectionId) async {
await _dio.get('/library/sections/$sectionId/analyze');
}
Future<void> _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);
}
}
}
@@ -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<String> urls) {
_setEndpoints(urls);
}
late List<String> _endpoints;
int _currentIndex = 0;
List<String> 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<String> urls, {
String? currentBaseUrl,
}) {
_setEndpoints(urls);
if (currentBaseUrl != null) {
final index = _endpoints.indexOf(currentBaseUrl);
_currentIndex = index >= 0 ? index : 0;
} else {
_currentIndex = 0;
}
}
void _setEndpoints(List<String> urls) {
final sanitized = <String>[];
final seen = <String>{};
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<void> Function(String newBaseUrl) onEndpointSwitch,
}) : _dio = dio,
_onEndpointSwitch = onEndpointSwitch;
final Dio _dio;
final EndpointFailoverManager endpointManager;
final Future<void> 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<Response<dynamic>> _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<dynamic>(
requestOptions.path,
data: requestOptions.data,
queryParameters: requestOptions.queryParameters,
options: options,
cancelToken: requestOptions.cancelToken,
onSendProgress: requestOptions.onSendProgress,
onReceiveProgress: requestOptions.onReceiveProgress,
);
}
}
+362 -59
View File
@@ -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<PlexConnection> findBestWorkingConnection() async* {
if (connections.isEmpty) return;
Stream<PlexConnection> 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<String>? excludeUrls,
}) {
final seen = <String>{};
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<String> prioritizedEndpointUrls({String? preferredFirst}) {
final urls = <String>[];
final exclude = <String>{};
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<PlexConnection?> 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';
+245 -62
View File
@@ -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<List<ConnectivityResult>>?
_connectivitySubscription;
static Future<void>? _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<void> _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<void> _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',
);
}
}
}
}
@@ -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"))
+24
View File
@@ -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:
+1
View File
@@ -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
@@ -6,6 +6,7 @@
#include "generated_plugin_registrant.h"
#include <connectivity_plus/connectivity_plus_windows_plugin.h>
#include <hotkey_manager_windows/hotkey_manager_windows_plugin_c_api.h>
#include <media_kit_libs_windows_video/media_kit_libs_windows_video_plugin_c_api.h>
#include <media_kit_video/media_kit_video_plugin_c_api.h>
@@ -16,6 +17,8 @@
#include <window_manager/window_manager_plugin.h>
void RegisterPlugins(flutter::PluginRegistry* registry) {
ConnectivityPlusWindowsPluginRegisterWithRegistrar(
registry->GetRegistrarForPlugin("ConnectivityPlusWindowsPlugin"));
HotkeyManagerWindowsPluginCApiRegisterWithRegistrar(
registry->GetRegistrarForPlugin("HotkeyManagerWindowsPluginCApi"));
MediaKitLibsWindowsVideoPluginCApiRegisterWithRegistrar(
+1
View File
@@ -3,6 +3,7 @@
#
list(APPEND FLUTTER_PLUGIN_LIST
connectivity_plus
hotkey_manager_windows
media_kit_libs_windows_video
media_kit_video