yorickr/plezy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(lifecycle): abort HTTP before close

Browse Source
This commit is contained in:
edde746
2026-05-12 15:11:16 +02:00
parent bae127e459
commit 4577f26aa8
25 changed files with 1016 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/main.dart
+32 -5
View File
@@ -33,6 +33,7 @@ import 'services/settings_service.dart';
import 'utils/platform_detector.dart';
import 'services/apple_tv_remote_touch_service.dart';
import 'services/discord_rpc_service.dart';
import 'services/image_cache_service.dart';
import 'services/gamepad_service.dart';
import 'services/trakt/trakt_scrobble_service.dart';
import 'services/trakt/trakt_sync_service.dart';
@@ -66,7 +67,8 @@ import 'services/plex_api_cache.dart';
import 'database/app_database.dart';
import 'screens/video_player_screen.dart';
import 'utils/app_logger.dart';
import 'utils/media_server_http_client.dart' show httpClient;
import 'utils/managed_http_client.dart';
import 'utils/media_server_http_client.dart';
import 'utils/orientation_helper.dart';
import 'utils/watch_state_notifier.dart';
import 'i18n/strings.g.dart';
@@ -423,6 +425,7 @@ class _MainAppState extends State<MainApp> with WidgetsBindingObserver {
final Set<String> _pendingSyncKeys = <String>{};
bool _isAutoDeleteRunning = false;
bool _lastConnectivityWasWifi = false;
bool _shutdownStarted = false;
/// Last time server health probes ran from a resume event (cooldown for desktop)
DateTime _lastResumeProbe = DateTime(0);
@@ -470,13 +473,35 @@ class _MainAppState extends State<MainApp> with WidgetsBindingObserver {
_appLifecycleListener = AppLifecycleListener(
onExitRequested: () async {
httpClient.close();
await _appDatabase.close();
await _shutdownForExit();
return AppExitResponse.exit;
},
);
}
Future<void> _shutdownForExit() async {
if (_shutdownStarted) return;
_shutdownStarted = true;
_syncDebounce?.cancel();
await _watchStateSubscription?.cancel();
await _connectivitySubscription?.cancel();
_memoryCheckTimer?.cancel();
_downloadManager.dispose();
TrackerCoordinator.instance.cancelInFlight();
TraktScrobbleService.instance.cancelInFlight();
await TraktSyncService.instance.dispose();
await _serverManager.disconnectAllGracefully();
await Future.wait([
httpClient.closeGracefully(drainTimeout: const Duration(seconds: 5)),
closeArtworkHttpClientGracefully(),
], eagerError: false);
await ManagedHttpClient.closeAllGracefully();
await _appDatabase.close();
}
@override
void dispose() {
_syncDebounce?.cancel();
@@ -484,8 +509,10 @@ class _MainAppState extends State<MainApp> with WidgetsBindingObserver {
_connectivitySubscription?.cancel();
_memoryCheckTimer?.cancel();
_appLifecycleListener.dispose();
_downloadManager.dispose();
_serverManager.dispose();
if (!_shutdownStarted) {
_downloadManager.dispose();
_serverManager.dispose();
}
WidgetsBinding.instance.removeObserver(this);
super.dispose();
}
lib/media/media_server_client.dart
+4
View File
@@ -61,6 +61,10 @@ import 'server_capabilities.dart';
/// two states to different UI ("Sign in again" vs "Server offline").
enum HealthStatus { online, offline, authError }
abstract interface class GracefullyCloseable {
Future<void> closeGracefully({Duration drainTimeout});
}
abstract class MediaServerClient {
String get serverId;
String? get serverName;
lib/providers/trackers_provider.dart
+12 -6
View File
@@ -180,41 +180,47 @@ class TrackersProvider extends ChangeNotifier with DisposableChangeNotifierMixin
}
Future<MalSession> _enrichMal(MalSession raw) async {
MalClient? tmp;
try {
final tmp = MalClient(raw, onSessionInvalidated: () {});
tmp = MalClient(raw, onSessionInvalidated: () {});
final user = await tmp.getMyUser();
tmp.dispose();
final name = user?['name'] as String?;
return name != null ? raw.copyWith(username: name) : raw;
} catch (e) {
appLogger.d('MAL: getMyUser failed (non-fatal)', error: e);
return raw;
} finally {
tmp?.dispose();
}
}
Future<AnilistSession> _enrichAnilist(AnilistSession raw) async {
AnilistClient? tmp;
try {
final tmp = AnilistClient(raw, onSessionInvalidated: () {});
tmp = AnilistClient(raw, onSessionInvalidated: () {});
final name = await tmp.getViewerName();
tmp.dispose();
return name != null ? raw.copyWith(username: name) : raw;
} catch (e) {
appLogger.d('AniList: getViewerName failed (non-fatal)', error: e);
return raw;
} finally {
tmp?.dispose();
}
}
Future<SimklSession> _enrichSimkl(SimklSession raw) async {
SimklClient? tmp;
try {
final tmp = SimklClient(raw, onSessionInvalidated: () {});
tmp = SimklClient(raw, onSessionInvalidated: () {});
final user = await tmp.getUserSettings();
tmp.dispose();
final userObj = user?['user'];
final name = userObj is Map ? userObj['name'] as String? : null;
return name != null ? raw.copyWith(username: name) : raw;
} catch (e) {
appLogger.d('Simkl: getUserSettings failed (non-fatal)', error: e);
return raw;
} finally {
tmp?.dispose();
}
}
lib/providers/trakt_account_provider.dart
+9 -4
View File
@@ -77,14 +77,16 @@ class TraktAccountProvider extends ChangeNotifier with DisposableChangeNotifierM
}
Future<TraktSession> _enrichUsername(TraktSession raw) async {
TraktClient? tmp;
try {
final tmp = TraktClient(raw, onSessionInvalidated: () {});
tmp = TraktClient(raw, onSessionInvalidated: () {});
final user = await tmp.getUserSettings();
tmp.dispose();
return raw.copyWith(username: user.username);
} catch (e) {
appLogger.d('Trakt: getUserSettings failed (non-fatal)', error: e);
return raw;
} finally {
tmp?.dispose();
}
}
@@ -93,8 +95,11 @@ class TraktAccountProvider extends ChangeNotifier with DisposableChangeNotifierM
final session = _session;
if (session != null) {
final client = TraktClient(session, onSessionInvalidated: () {});
await client.revoke();
client.dispose();
try {
await client.revoke();
} finally {
client.dispose();
}
}
await _store.clear(_activeUserUuid);
_setSessionAndRebind(null);
lib/services/image_cache_service.dart
+6
View File
@@ -1,3 +1,5 @@
import 'dart:async';
// CE's public conditional export hides the IO-only httpClientFactory parameter
// behind a narrower unsupported-platform stub.
// ignore: implementation_imports
@@ -8,6 +10,10 @@ import '../utils/media_server_http_client.dart';
final _artworkHttpClient = MediaServerHttpClient(usePlexApiClient: true);
Future<void> closeArtworkHttpClientGracefully({Duration drainTimeout = const Duration(seconds: 5)}) {
return _artworkHttpClient.closeGracefully(drainTimeout: drainTimeout);
}
/// Shared cache manager for media-server image artwork. Used for both Plex and
/// Jellyfin artwork (the class name predates Jellyfin support — it's
/// backend-neutral).
lib/services/jellyfin_auth_service.dart
+17 -16
View File
@@ -82,7 +82,7 @@ class JellyfinConnectionAuthService implements ConnectionAuthService {
final normalised = _normaliseBaseUrl(baseUrl);
final client = _buildHttpClient(baseUrl: normalised);
try {
final response = await client.get('/System/Info/Public').timeout(MediaServerTimeouts.jellyfinProbe);
final response = await client.get('/System/Info/Public', timeout: MediaServerTimeouts.jellyfinProbe);
throwIfHttpError(response);
final data = response.data;
if (data is! Map<String, dynamic>) {
@@ -100,8 +100,8 @@ class JellyfinConnectionAuthService implements ConnectionAuthService {
} on MediaServerHttpException catch (e) {
throw MediaServerUrlException('Server probe failed: ${e.message}');
} on TimeoutException {
// The `.timeout(...)` above throws raw [TimeoutException]; wrap so
// callers can `catch (MediaServerUrlException)` uniformly.
// Defensive: most request timeouts are wrapped by MediaServerHttpClient,
// but keep raw timeouts surfaced uniformly if one escapes.
throw MediaServerUrlException('Server did not respond in time');
} catch (e) {
// Catch-all for transport errors that bypass the http client wrap
@@ -136,11 +136,13 @@ class JellyfinConnectionAuthService implements ConnectionAuthService {
headers: {'Authorization': authHeader, 'Content-Type': 'application/json'},
);
try {
final response = await client
.post('/Users/AuthenticateByName', body: jsonEncode({'Username': username, 'Pw': password}))
// Bound the auth POST so a hanging server can't freeze the auth
// screen indefinitely; mirrors the timeout on [probe].
.timeout(MediaServerTimeouts.jellyfinProbe);
final response = await client.post(
'/Users/AuthenticateByName',
body: jsonEncode({'Username': username, 'Pw': password}),
// Bound the auth POST so a hanging server can't freeze the auth
// screen indefinitely; mirrors the timeout on [probe].
timeout: MediaServerTimeouts.jellyfinProbe,
);
if (response.statusCode == 401 || response.statusCode == 403) {
throw MediaServerAuthException('Invalid username or password', statusCode: response.statusCode);
}
@@ -172,9 +174,8 @@ class JellyfinConnectionAuthService implements ConnectionAuthService {
isAdministrator: isAdmin,
);
} on TimeoutException {
// The auth POST's `.timeout(...)` throws raw [TimeoutException]; surface
// as a URL-level error so the auth screen shows a normal "couldn't
// reach server" message instead of a stack trace.
// Defensive: most request timeouts are wrapped by MediaServerHttpClient.
// Surface raw timeouts as a URL-level error if one escapes.
throw MediaServerUrlException('Server did not respond in time');
} on MediaServerHttpException catch (e) {
if (e.statusCode == 401 || e.statusCode == 403) {
@@ -193,7 +194,7 @@ class JellyfinConnectionAuthService implements ConnectionAuthService {
final normalised = _normaliseBaseUrl(baseUrl);
final client = _buildHttpClient(baseUrl: normalised);
try {
final response = await client.get('/QuickConnect/Enabled').timeout(MediaServerTimeouts.jellyfinProbe);
final response = await client.get('/QuickConnect/Enabled', timeout: MediaServerTimeouts.jellyfinProbe);
if (response.statusCode != 200) return false;
final data = response.data;
// The endpoint returns a bare JSON `true`/`false`, not an object.
@@ -223,9 +224,9 @@ class JellyfinConnectionAuthService implements ConnectionAuthService {
try {
// Current Jellyfin (10.7+) accepts GET; older builds required POST.
// Try GET first, fall back on 405.
var response = await client.get('/QuickConnect/Initiate').timeout(MediaServerTimeouts.jellyfinProbe);
var response = await client.get('/QuickConnect/Initiate', timeout: MediaServerTimeouts.jellyfinProbe);
if (response.statusCode == 405) {
response = await client.post('/QuickConnect/Initiate').timeout(MediaServerTimeouts.jellyfinProbe);
response = await client.post('/QuickConnect/Initiate', timeout: MediaServerTimeouts.jellyfinProbe);
}
if (response.statusCode == 401 || response.statusCode == 403) {
throw MediaServerAuthException('Quick Connect rejected by server', statusCode: response.statusCode);
@@ -375,7 +376,7 @@ class JellyfinConnectionAuthService implements ConnectionAuthService {
if (connection is! JellyfinConnection) return false;
final client = _authenticatedClient(connection);
try {
final response = await client.get('/Users/Me').timeout(MediaServerTimeouts.jellyfinProbe);
final response = await client.get('/Users/Me', timeout: MediaServerTimeouts.jellyfinProbe);
return response.statusCode == 200;
} on MediaServerHttpException catch (e) {
if (e.statusCode == 401 || e.statusCode == 403) return false;
@@ -401,7 +402,7 @@ class JellyfinConnectionAuthService implements ConnectionAuthService {
final client = _authenticatedClient(connection);
try {
// Best-effort: server may already have invalidated the session.
await client.post('/Sessions/Logout').timeout(MediaServerTimeouts.jellyfinSignOut);
await client.post('/Sessions/Logout', timeout: MediaServerTimeouts.jellyfinSignOut);
} catch (e) {
appLogger.d('JellyfinConnectionAuthService: signOut best-effort failed: $e');
} finally {
lib/services/jellyfin_client.dart
+6 -2
View File
@@ -85,7 +85,7 @@ class JellyfinClient
_JellyfinFileInfoMethods,
_JellyfinLiveTvMethods,
_JellyfinImageDownloadMethods
implements MediaServerClient, ScopedMediaServerClient {
implements MediaServerClient, ScopedMediaServerClient, GracefullyCloseable {
JellyfinClient._({
required JellyfinConnection connection,
required MediaServerHttpClient http,
@@ -229,6 +229,10 @@ class JellyfinClient
@override
void close() => _http.close();
@override
Future<void> closeGracefully({Duration drainTimeout = const Duration(seconds: 2)}) =>
_http.closeGracefully(drainTimeout: drainTimeout);
/// Reachable *and* token-valid. We probe `/Users/Me` (auth-required)
/// rather than `/System/Info/Public` so a revoked token surfaces as
/// unhealthy on the very next sweep, instead of waiting for the first
@@ -244,7 +248,7 @@ class JellyfinClient
@override
Future<HealthStatus> checkHealth() async {
try {
final response = await _http.get('/Users/Me').timeout(const Duration(seconds: 8));
final response = await _http.get('/Users/Me', timeout: const Duration(seconds: 8));
final ok = response.statusCode >= 200 && response.statusCode < 300;
if (ok) {
final data = response.data;
lib/services/multi_server_manager.dart
+57 -19
View File
@@ -310,11 +310,12 @@ class MultiServerManager {
final client = _jellyfinByCompoundId.remove(compoundId);
_jellyfinHealthByCompoundId.remove(compoundId);
if (client != null && closed.add(client)) {
client.close();
_closeClient(client);
}
}
} else {
_clients.remove(serverId)?.close();
final client = _clients.remove(serverId);
if (client != null) _closeClient(client);
}
_plexServers.remove(serverId);
_serverStatus.remove(serverId);
@@ -323,6 +324,25 @@ class MultiServerManager {
appLogger.i('Removed server: $serverId');
}
void _closeClient(MediaServerClient client) {
if (client case final GracefullyCloseable graceful) {
unawaited(graceful.closeGracefully());
} else {
client.close();
}
}
Future<void> _closeClientGracefully(
MediaServerClient client, {
Duration drainTimeout = const Duration(seconds: 2),
}) async {
if (client case final GracefullyCloseable graceful) {
await graceful.closeGracefully(drainTimeout: drainTimeout);
} else {
client.close();
}
}
/// Connect every server attached to a Plex account in parallel. Each
/// account has its own `clientIdentifier` (registered as a separate
/// device on plex.tv), and we keep that mapping per-server in
@@ -349,7 +369,8 @@ class MultiServerManager {
server: server,
clientIdentifier: connection.clientIdentifier,
).namedTimeout(timeout, operation: 'connect to ${server.name}');
_clients[serverId]?.close();
final oldClient = _clients[serverId];
if (oldClient != null) _closeClient(oldClient);
_clients[serverId] = client;
_serverStatus[serverId] = true;
onServerStatus?.call(serverId, true);
@@ -409,7 +430,8 @@ class MultiServerManager {
server: server,
clientIdentifier: connection.clientIdentifier,
).namedTimeout(timeout, operation: 'connect to ${server.name}');
_clients[serverId]?.close();
final oldClient = _clients[serverId];
if (oldClient != null) _closeClient(oldClient);
_clients[serverId] = client;
_serverStatus[serverId] = true;
_authErrorServers.remove(serverId);
@@ -433,7 +455,8 @@ class MultiServerManager {
void removePlexAccount(PlexAccountConnection connection) {
for (final server in connection.servers) {
final id = server.clientIdentifier;
_clients.remove(id)?.close();
final client = _clients.remove(id);
if (client != null) _closeClient(client);
_plexServers.remove(id);
_serverStatus.remove(id);
_authErrorServers.remove(id);
@@ -464,7 +487,8 @@ class MultiServerManager {
// Replace any prior client for this exact compound id (re-add of the
// same user — e.g., token refresh or settings re-add).
_jellyfinByCompoundId[compoundId]?.close();
final oldClient = _jellyfinByCompoundId[compoundId];
if (oldClient != null) _closeClient(oldClient);
_jellyfinByCompoundId[compoundId] = client;
// Bind this user as the active client for its machine. A previously
@@ -521,7 +545,7 @@ class MultiServerManager {
final machineId = connection.serverMachineId;
final client = _jellyfinByCompoundId.remove(compoundId);
_jellyfinHealthByCompoundId.remove(compoundId);
client?.close();
if (client != null) _closeClient(client);
if (_activeJellyfinMachine[machineId] == compoundId) {
_activeJellyfinMachine.remove(machineId);
_clients.remove(machineId);
@@ -766,7 +790,8 @@ class MultiServerManager {
appLogger.d('Attempting reconnection for ${server.name}');
final client = await _createClientForServer(server: server, clientIdentifier: clientId);
_clients[serverId]?.close();
final oldClient = _clients[serverId];
if (oldClient != null) _closeClient(oldClient);
_clients[serverId] = client;
updateServerStatus(serverId, true);
appLogger.i('Successfully reconnected to ${server.name}');
@@ -910,6 +935,22 @@ class MultiServerManager {
/// Disconnect all servers
void disconnectAll() {
appLogger.i('Disconnecting all servers');
final clients = _detachAllClients();
for (final client in clients) {
_closeClient(client);
}
}
Future<void> disconnectAllGracefully({Duration drainTimeout = const Duration(seconds: 5)}) async {
appLogger.i('Gracefully disconnecting all servers');
final clients = _detachAllClients();
await Future.wait(
clients.map((client) => _closeClientGracefully(client, drainTimeout: drainTimeout)),
eagerError: false,
);
}
Set<MediaServerClient> _detachAllClients() {
_stopNetworkMonitoring();
for (final timer in _reconnectDebounce.values) {
timer.cancel();
@@ -917,15 +958,7 @@ class MultiServerManager {
_reconnectDebounce.clear();
_activeHealthCheck = null;
_activeReconnect = null;
final activeClients = _clients.values.toSet();
for (final client in _clients.values) {
client.close();
}
for (final client in _jellyfinByCompoundId.values) {
if (!activeClients.contains(client)) {
client.close();
}
}
final clients = <MediaServerClient>{..._clients.values, ..._jellyfinByCompoundId.values};
_clients.clear();
_jellyfinByCompoundId.clear();
_activeJellyfinMachine.clear();
@@ -935,12 +968,17 @@ class MultiServerManager {
_authErrorServers.clear();
_clientIdByServer.clear();
_activeOptimizations.clear();
_statusController.add({});
if (!_statusController.isClosed) {
_statusController.add({});
}
return clients;
}
/// Dispose resources
void dispose() {
disconnectAll();
_statusController.close();
if (!_statusController.isClosed) {
_statusController.close();
}
}
}
lib/services/plex_client.dart
+33 -22
View File
@@ -134,7 +134,9 @@ class ConnectionTestResult {
ConnectionTestResult({required this.success, required this.latencyMs, this.error, this.transcoderVideo});
}
class PlexClient with MediaServerCacheMixin, _PlexLiveTvClientMethods implements MediaServerClient {
class PlexClient
with MediaServerCacheMixin, _PlexLiveTvClientMethods
implements MediaServerClient, GracefullyCloseable {
@override
PlexConfig config;
@@ -303,6 +305,11 @@ class PlexClient with MediaServerCacheMixin, _PlexLiveTvClientMethods implements
_http.close();
}
@override
Future<void> closeGracefully({Duration drainTimeout = const Duration(seconds: 2)}) {
return _http.closeGracefully(drainTimeout: drainTimeout);
}
bool _failoverSwitching = false;
/// Execute a GET request with endpoint failover retry. On timeout/connection
@@ -2578,30 +2585,34 @@ class PlexClient with MediaServerCacheMixin, _PlexLiveTvClientMethods implements
receiveTimeout: MediaServerTimeouts.receive,
defaultHeaders: const {'Accept-Language': 'en', 'Accept': 'application/json'},
);
final decisionUrl = '${config.baseUrl}/video/:/transcode/universal/decision?$queryString';
final decisionResponse = await decisionClient.get(decisionUrl);
try {
final decisionUrl = '${config.baseUrl}/video/:/transcode/universal/decision?$queryString';
final decisionResponse = await decisionClient.get(decisionUrl);
final decisionBody = decisionResponse.data?.toString() ?? '<empty>';
appLogger.i(
'Transcode decision [${decisionResponse.statusCode}] body: '
'${decisionBody.length > 2000 ? '${decisionBody.substring(0, 2000)}…' : decisionBody}',
);
final decisionBody = decisionResponse.data?.toString() ?? '<empty>';
appLogger.i(
'Transcode decision [${decisionResponse.statusCode}] body: '
'${decisionBody.length > 2000 ? '${decisionBody.substring(0, 2000)}…' : decisionBody}',
);
if (decisionResponse.statusCode != 200) {
appLogger.w('Transcode decision returned ${decisionResponse.statusCode}');
return (startPath: null, outcome: TranscodeDecisionOutcome.failed);
if (decisionResponse.statusCode != 200) {
appLogger.w('Transcode decision returned ${decisionResponse.statusCode}');
return (startPath: null, outcome: TranscodeDecisionOutcome.failed);
}
final outcome = _parseTranscodeDecisionOutcome(decisionResponse.data, isOriginal: isOriginal);
if (outcome == TranscodeDecisionOutcome.failed) {
return (startPath: null, outcome: outcome);
}
final startParams = Map<String, String>.from(allParams)..remove('X-Plex-Token');
final startQuery = startParams.entries.map((e) => '${_plexEncode(e.key)}=${_plexEncode(e.value)}').join('&');
// `.m3u8` tells the server to return an HLS manifest.
return (startPath: '/video/:/transcode/universal/start.m3u8?$startQuery', outcome: outcome);
} finally {
decisionClient.close();
}
final outcome = _parseTranscodeDecisionOutcome(decisionResponse.data, isOriginal: isOriginal);
if (outcome == TranscodeDecisionOutcome.failed) {
return (startPath: null, outcome: outcome);
}
final startParams = Map<String, String>.from(allParams)..remove('X-Plex-Token');
final startQuery = startParams.entries.map((e) => '${_plexEncode(e.key)}=${_plexEncode(e.value)}').join('&');
// `.m3u8` tells the server to return an HLS manifest.
return (startPath: '/video/:/transcode/universal/start.m3u8?$startQuery', outcome: outcome);
} catch (e, st) {
appLogger.e('Failed to build transcode start path', error: e, stackTrace: st);
return (startPath: null, outcome: TranscodeDecisionOutcome.failed);
lib/services/plex_client/parts/live_tv.dart
+26 -22
View File
@@ -1009,30 +1009,34 @@ mixin _PlexLiveTvClientMethods on MediaServerCacheMixin {
receiveTimeout: MediaServerTimeouts.receive,
defaultHeaders: {'Accept-Language': 'en'},
);
final decisionUrl = '${config.baseUrl}/video/:/transcode/universal/decision?$queryString';
final decisionResponse = await decisionClient.get(decisionUrl);
try {
final decisionUrl = '${config.baseUrl}/video/:/transcode/universal/decision?$queryString';
final decisionResponse = await decisionClient.get(decisionUrl);
if (decisionResponse.statusCode != 200) {
appLogger.w('Decision returned ${decisionResponse.statusCode}');
return null;
if (decisionResponse.statusCode != 200) {
appLogger.w('Decision returned ${decisionResponse.statusCode}');
return null;
}
// Log decision response for diagnostics (the web client parses this XML
// to extract generalDecisionCode, mdeDecisionCode, transcodeDecisionCode).
final decisionBody = decisionResponse.data?.toString() ?? '';
if (decisionBody.isNotEmpty) {
appLogger.d(
'Decision response: ${decisionBody.length > 500 ? '${decisionBody.substring(0, 500)}...' : decisionBody}',
);
}
// Token is added by the caller via .withPlexToken()
final startParams = Map<String, String>.from(allParams)..remove('X-Plex-Token');
final startQuery = startParams.entries
.map((e) => '${Uri.encodeComponent(e.key)}=${Uri.encodeComponent(e.value)}')
.join('&');
return '/video/:/transcode/universal/start?$startQuery';
} finally {
decisionClient.close();
}
// Log decision response for diagnostics (the web client parses this XML
// to extract generalDecisionCode, mdeDecisionCode, transcodeDecisionCode).
final decisionBody = decisionResponse.data?.toString() ?? '';
if (decisionBody.isNotEmpty) {
appLogger.d(
'Decision response: ${decisionBody.length > 500 ? '${decisionBody.substring(0, 500)}...' : decisionBody}',
);
}
// Token is added by the caller via .withPlexToken()
final startParams = Map<String, String>.from(allParams)..remove('X-Plex-Token');
final startQuery = startParams.entries
.map((e) => '${Uri.encodeComponent(e.key)}=${Uri.encodeComponent(e.value)}')
.join('&');
return '/video/:/transcode/universal/start?$startQuery';
} catch (e, st) {
appLogger.e('Failed to build live stream path', error: e, stackTrace: st);
return null;
lib/services/trackers/anilist/anilist_client.dart
+10 -1
View File
@@ -3,6 +3,7 @@ import 'dart:convert';
import 'package:http/http.dart' as http;
import '../../../utils/abortable_http_request.dart';
import '../../../utils/app_logger.dart';
import '../../../utils/platform_http_client_stub.dart'
if (dart.library.io) '../../../utils/platform_http_client_io.dart'
@@ -54,7 +55,15 @@ class AnilistClient {
final body = json.encode({'query': query, 'variables': ?variables});
final sw = Stopwatch()..start();
final res = await _http.post(uri, headers: headers, body: body).timeout(TrackerConstants.requestTimeout);
final res = await sendAbortableHttpRequest(
_http,
'POST',
uri,
headers: headers,
body: body,
timeout: TrackerConstants.requestTimeout,
operation: 'AniList POST ${uri.path}',
);
sw.stop();
appLogger.d('AniList POST ${uri.path} → ${res.statusCode} (${sw.elapsedMilliseconds}ms)');
lib/services/trackers/fribb_mapping_store.dart
+17 -6
View File
@@ -7,6 +7,7 @@ import 'package:path/path.dart' as p;
import 'package:path_provider/path_provider.dart';
import '../../models/trackers/fribb_mapping_row.dart';
import '../base_shared_preferences_service.dart';
import '../../utils/abortable_http_request.dart';
import '../../utils/app_logger.dart';
import '../../utils/platform_http_client_stub.dart'
if (dart.library.io) '../../utils/platform_http_client_io.dart'
@@ -99,9 +100,14 @@ class FribbMappingStore {
Future<String?> _download() async {
final client = platform.createPlatformClient();
try {
final res = await client
.get(Uri.parse(_sourceUrl), headers: const {'Accept': 'application/json'})
.timeout(_requestTimeout);
final res = await sendAbortableHttpRequest(
client,
'GET',
Uri.parse(_sourceUrl),
headers: const {'Accept': 'application/json'},
timeout: _requestTimeout,
operation: 'Fribb mapping download',
);
if (res.statusCode != 200) {
appLogger.d('Fribb: download returned HTTP ${res.statusCode}');
return null;
@@ -154,9 +160,14 @@ class FribbMappingStore {
final etag = prefs.getString(_prefsEtagKey);
final client = platform.createPlatformClient();
try {
final res = await client
.get(Uri.parse(_sourceUrl), headers: {'If-None-Match': ?etag, 'Accept': 'application/json'})
.timeout(_requestTimeout);
final res = await sendAbortableHttpRequest(
client,
'GET',
Uri.parse(_sourceUrl),
headers: {'If-None-Match': ?etag, 'Accept': 'application/json'},
timeout: _requestTimeout,
operation: 'Fribb mapping refresh',
);
await prefs.setInt(_prefsLastCheckKey, now);
if (res.statusCode == 304) {
lib/services/trackers/mal/mal_auth_service.dart
+9 -10
View File
@@ -2,6 +2,7 @@ import 'dart:convert';
import 'package:http/http.dart' as http;
import '../../../utils/abortable_http_request.dart';
import '../../../utils/app_logger.dart';
import '../../../utils/platform_http_client_stub.dart'
if (dart.library.io) '../../../utils/platform_http_client_io.dart'
@@ -35,16 +36,14 @@ class MalAuthService extends OAuthProxyAuthServiceBase<MalSession> {
}
Future<MalSession> refresh(MalSession current) async {
final res = await _http
.post(
Uri.parse(MalConstants.tokenUrl),
body: {
'client_id': MalConstants.clientId,
'grant_type': 'refresh_token',
'refresh_token': current.refreshToken,
},
)
.timeout(TrackerConstants.requestTimeout);
final res = await sendAbortableHttpRequest(
_http,
'POST',
Uri.parse(MalConstants.tokenUrl),
body: {'client_id': MalConstants.clientId, 'grant_type': 'refresh_token', 'refresh_token': current.refreshToken},
timeout: TrackerConstants.requestTimeout,
operation: 'MAL token refresh',
);
if (res.statusCode != 200) {
appLogger.w('MAL: refresh failed (${res.statusCode}): ${res.body}');
lib/services/trackers/mal/mal_client.dart
+11 -6
View File
@@ -3,6 +3,7 @@ import 'dart:convert';
import 'package:http/http.dart' as http;
import '../../../utils/abortable_http_request.dart';
import '../../../utils/app_logger.dart';
import '../../../utils/platform_http_client_stub.dart'
if (dart.library.io) '../../../utils/platform_http_client_io.dart'
@@ -131,13 +132,17 @@ class MalClient {
final sw = Stopwatch()..start();
final res = await switch (method) {
'GET' => _http.get(uri, headers: headers),
'POST' => _http.post(uri, headers: headers, body: encoded),
'PATCH' => _http.patch(uri, headers: headers, body: encoded),
'PUT' => _http.put(uri, headers: headers, body: encoded),
'DELETE' => _http.delete(uri, headers: headers),
'GET' || 'POST' || 'PATCH' || 'PUT' || 'DELETE' => sendAbortableHttpRequest(
_http,
method,
uri,
headers: headers,
body: encoded,
timeout: TrackerConstants.requestTimeout,
operation: 'MAL $method ${uri.path}',
),
_ => throw ArgumentError('Unsupported HTTP method: $method'),
}.timeout(TrackerConstants.requestTimeout);
};
sw.stop();
appLogger.d('MAL $method ${uri.path} → ${res.statusCode} (${sw.elapsedMilliseconds}ms)');
return res;
lib/services/trackers/oauth_proxy_client.dart
+18 -8
View File
@@ -3,6 +3,7 @@ import 'dart:convert';
import 'package:http/http.dart' as http;
import '../../utils/abortable_http_request.dart';
import '../../utils/app_logger.dart';
import '../../utils/platform_http_client_stub.dart'
if (dart.library.io) '../../utils/platform_http_client_io.dart'
@@ -29,13 +30,15 @@ class OAuthProxyClient {
/// POST /auth/start — register a new session. Returns a handle including the
/// URL to display as a QR code for the phone scan.
Future<OAuthProxyStart> start(String service) async {
final res = await _http
.post(
Uri.parse('$baseUrl/auth/start'),
headers: {'Content-Type': 'application/json'},
body: json.encode({'service': service}),
)
.timeout(TrackerConstants.authRequestTimeout);
final res = await sendAbortableHttpRequest(
_http,
'POST',
Uri.parse('$baseUrl/auth/start'),
headers: {'Content-Type': 'application/json'},
body: json.encode({'service': service}),
timeout: TrackerConstants.authRequestTimeout,
operation: 'OAuth proxy start',
);
if (res.statusCode != 200) {
throw OAuthProxyException('start failed: HTTP ${res.statusCode}: ${res.body}');
}
@@ -65,7 +68,14 @@ class OAuthProxyClient {
final Object? raced;
try {
raced = await Future.any<Object?>([
_http.get(uri).timeout(TrackerConstants.oauthProxyPollTimeout),
sendAbortableHttpRequest(
_http,
'GET',
uri,
timeout: TrackerConstants.oauthProxyPollTimeout,
abortTrigger: onCancel,
operation: 'OAuth proxy poll',
),
?cancelFuture,
]);
} on TimeoutException {
lib/services/trackers/simkl/simkl_auth_service.dart
+17 -6
View File
@@ -3,6 +3,7 @@ import 'dart:convert';
import 'package:http/http.dart' as http;
import '../../../models/trackers/device_code.dart';
import '../../../utils/abortable_http_request.dart';
import '../../../utils/app_logger.dart';
import '../device_code_auth_service.dart';
import '../oauth_proxy_client.dart';
@@ -24,9 +25,14 @@ class SimklAuthService extends DeviceCodeAuthServiceBase<SimklSession> {
final uri = Uri.parse(SimklConstants.pinUrl).replace(
queryParameters: {'client_id': SimklConstants.clientId, 'redirect': '${OAuthProxyClient.baseUrl}/auth/done'},
);
final res = await httpClient
.get(uri, headers: SimklConstants.headers())
.timeout(TrackerConstants.authRequestTimeout);
final res = await sendAbortableHttpRequest(
httpClient,
'GET',
uri,
headers: SimklConstants.headers(),
timeout: TrackerConstants.authRequestTimeout,
operation: 'Simkl PIN request',
);
if (res.statusCode != 200) {
throw DeviceCodeAuthFlowException('Simkl PIN request failed: HTTP ${res.statusCode}: ${res.body}');
}
@@ -49,9 +55,14 @@ class SimklAuthService extends DeviceCodeAuthServiceBase<SimklSession> {
).replace(queryParameters: {'client_id': SimklConstants.clientId});
final http.Response res;
try {
res = await httpClient
.get(pollUri, headers: SimklConstants.headers())
.timeout(TrackerConstants.authRequestTimeout);
res = await sendAbortableHttpRequest(
httpClient,
'GET',
pollUri,
headers: SimklConstants.headers(),
timeout: TrackerConstants.authRequestTimeout,
operation: 'Simkl PIN poll',
);
} catch (e) {
appLogger.d('Simkl device-code poll error (transient)', error: e);
return const DevicePollPending();
lib/services/trackers/simkl/simkl_client.dart
+11 -3
View File
@@ -3,6 +3,7 @@ import 'dart:convert';
import 'package:http/http.dart' as http;
import '../../../utils/abortable_http_request.dart';
import '../../../utils/app_logger.dart';
import '../../../utils/platform_http_client_stub.dart'
if (dart.library.io) '../../../utils/platform_http_client_io.dart'
@@ -45,10 +46,17 @@ class SimklClient {
final sw = Stopwatch()..start();
final res = await switch (method) {
'GET' => _http.get(uri, headers: headers),
'POST' => _http.post(uri, headers: headers, body: encoded),
'GET' || 'POST' => sendAbortableHttpRequest(
_http,
method,
uri,
headers: headers,
body: encoded,
timeout: TrackerConstants.requestTimeout,
operation: 'Simkl $method ${uri.path}',
),
_ => throw ArgumentError('Unsupported HTTP method: $method'),
}.timeout(TrackerConstants.requestTimeout);
};
sw.stop();
appLogger.d('Simkl $method ${uri.path} → ${res.statusCode} (${sw.elapsedMilliseconds}ms)');
lib/services/trakt/trakt_auth_service.dart
+23 -14
View File
@@ -3,6 +3,7 @@ import 'dart:convert';
import 'package:http/http.dart' as http;
import '../../models/trackers/device_code.dart';
import '../../utils/abortable_http_request.dart';
import '../../utils/app_logger.dart';
import '../trackers/device_code_auth_service.dart';
import '../trackers/tracker_constants.dart';
@@ -20,9 +21,15 @@ class TraktAuthService extends DeviceCodeAuthServiceBase<TraktSession> {
Future<DeviceCode> createDeviceCode() async {
final uri = Uri.parse(TraktConstants.deviceCodeUrl);
final sw = Stopwatch()..start();
final res = await httpClient
.post(uri, headers: TraktConstants.headers(), body: json.encode({'client_id': TraktConstants.clientId}))
.timeout(TrackerConstants.authRequestTimeout);
final res = await sendAbortableHttpRequest(
httpClient,
'POST',
uri,
headers: TraktConstants.headers(),
body: json.encode({'client_id': TraktConstants.clientId}),
timeout: TrackerConstants.authRequestTimeout,
operation: 'Trakt device code request',
);
sw.stop();
appLogger.d('Trakt POST ${uri.path} → ${res.statusCode} (${sw.elapsedMilliseconds}ms)');
@@ -48,17 +55,19 @@ class TraktAuthService extends DeviceCodeAuthServiceBase<TraktSession> {
final tokenUri = Uri.parse(TraktConstants.deviceTokenUrl);
final http.Response res;
try {
res = await httpClient
.post(
tokenUri,
headers: TraktConstants.headers(),
body: json.encode({
'code': code.deviceCode,
'client_id': TraktConstants.clientId,
'client_secret': TraktConstants.clientSecret,
}),
)
.timeout(TrackerConstants.authRequestTimeout);
res = await sendAbortableHttpRequest(
httpClient,
'POST',
tokenUri,
headers: TraktConstants.headers(),
body: json.encode({
'code': code.deviceCode,
'client_id': TraktConstants.clientId,
'client_secret': TraktConstants.clientSecret,
}),
timeout: TrackerConstants.authRequestTimeout,
operation: 'Trakt device token poll',
);
appLogger.d('Trakt POST ${tokenUri.path} → ${res.statusCode}');
} catch (e) {
appLogger.d('Trakt device-code poll error (transient)', error: e);
lib/services/trakt/trakt_client.dart
+39 -28
View File
@@ -5,6 +5,7 @@ import 'package:http/http.dart' as http;
import '../../models/trakt/trakt_scrobble_request.dart';
import '../../models/trakt/trakt_user.dart';
import '../../utils/abortable_http_request.dart';
import '../../utils/app_logger.dart';
import '../../utils/platform_http_client_stub.dart'
if (dart.library.io) '../../utils/platform_http_client_io.dart'
@@ -64,18 +65,21 @@ class TraktClient {
Future<TraktSession> _doRefresh() async {
appLogger.d('Trakt: refreshing access token');
final res = await _http
.post(
Uri.parse(TraktConstants.tokenUrl),
headers: TraktConstants.headers(),
body: json.encode({
'refresh_token': _session.refreshToken,
'client_id': TraktConstants.clientId,
'client_secret': TraktConstants.clientSecret,
'grant_type': 'refresh_token',
}),
)
.timeout(TrackerConstants.refreshTimeout);
final tokenUri = Uri.parse(TraktConstants.tokenUrl);
final res = await sendAbortableHttpRequest(
_http,
'POST',
tokenUri,
headers: TraktConstants.headers(),
body: json.encode({
'refresh_token': _session.refreshToken,
'client_id': TraktConstants.clientId,
'client_secret': TraktConstants.clientSecret,
'grant_type': 'refresh_token',
}),
timeout: TrackerConstants.refreshTimeout,
operation: 'Trakt token refresh',
);
if (res.statusCode == 200) {
final body = json.decode(res.body) as Map<String, dynamic>;
@@ -91,17 +95,19 @@ class TraktClient {
/// Revoke the access token at Trakt. Best-effort; swallows network errors.
Future<void> revoke() async {
try {
await _http
.post(
Uri.parse(TraktConstants.revokeUrl),
headers: TraktConstants.headers(),
body: json.encode({
'token': _session.accessToken,
'client_id': TraktConstants.clientId,
'client_secret': TraktConstants.clientSecret,
}),
)
.timeout(TrackerConstants.revokeTimeout);
await sendAbortableHttpRequest(
_http,
'POST',
Uri.parse(TraktConstants.revokeUrl),
headers: TraktConstants.headers(),
body: json.encode({
'token': _session.accessToken,
'client_id': TraktConstants.clientId,
'client_secret': TraktConstants.clientSecret,
}),
timeout: TrackerConstants.revokeTimeout,
operation: 'Trakt token revoke',
);
} catch (e) {
appLogger.d('Trakt: revoke failed (non-fatal)', error: e);
}
@@ -152,12 +158,17 @@ class TraktClient {
final sw = Stopwatch()..start();
final res = await switch (method) {
'GET' => _http.get(uri, headers: headers),
'POST' => _http.post(uri, headers: headers, body: encoded),
'PUT' => _http.put(uri, headers: headers, body: encoded),
'DELETE' => _http.delete(uri, headers: headers),
'GET' || 'POST' || 'PUT' || 'DELETE' => sendAbortableHttpRequest(
_http,
method,
uri,
headers: headers,
body: encoded,
timeout: TrackerConstants.requestTimeout,
operation: 'Trakt $method ${uri.path}',
),
_ => throw ArgumentError('Unsupported HTTP method: $method'),
}.timeout(TrackerConstants.requestTimeout);
};
sw.stop();
appLogger.d('Trakt $method ${uri.path} → ${res.statusCode} (${sw.elapsedMilliseconds}ms)');
lib/utils/abortable_http_request.dart
+59
View File
@@ -0,0 +1,59 @@
import 'dart:async';
import 'dart:convert';
import 'package:http/http.dart' as http;
Future<http.Response> sendAbortableHttpRequest(
http.Client client,
String method,
Uri uri, {
Map<String, String>? headers,
Object? body,
Encoding? encoding,
Duration? timeout,
Future<void>? abortTrigger,
String? operation,
}) {
final abort = Completer<void>();
void abortRequest() {
if (!abort.isCompleted) abort.complete();
}
if (abortTrigger != null) {
unawaited(abortTrigger.whenComplete(abortRequest));
}
final request = http.AbortableRequest(method, uri, abortTrigger: abort.future);
if (headers != null) request.headers.addAll(headers);
if (encoding != null) request.encoding = encoding;
if (body != null) _setBody(request, body);
final future = client.send(request).then(http.Response.fromStream);
if (timeout == null) return future.whenComplete(abortRequest);
return future
.timeout(
timeout,
onTimeout: () {
abortRequest();
throw TimeoutException('${operation ?? '$method ${uri.path}'} timed out', timeout);
},
)
.whenComplete(abortRequest);
}
void _setBody(http.Request request, Object body) {
if (body is String) {
request.body = body;
return;
}
if (body is List<int>) {
request.bodyBytes = body;
return;
}
if (body is Map) {
request.bodyFields = body.cast<String, String>();
return;
}
throw ArgumentError('Invalid request body "$body".');
}
lib/utils/managed_http_client.dart
+268
View File
@@ -0,0 +1,268 @@
import 'dart:async';
import 'package:http/http.dart' as http;
import 'app_logger.dart';
/// [http.Client] wrapper that owns native-client shutdown semantics.
///
/// `package:http` clients define closing with active requests as undefined. For
/// platform clients backed by native callbacks, especially CupertinoClient,
/// closing at the wrong time can leave callbacks racing a torn-down Dart bridge.
/// This wrapper tracks requests until their response stream finishes, aborts
/// active requests during shutdown, and only closes the inner client once the
/// active set has drained.
class ManagedHttpClient extends http.BaseClient {
ManagedHttpClient(this._inner, {required this.debugLabel}) {
_instances.add(this);
}
static final Set<ManagedHttpClient> _instances = <ManagedHttpClient>{};
static Future<void> closeAllGracefully({Duration drainTimeout = const Duration(seconds: 5)}) async {
await Future.wait(
_instances.toList().map((client) => client.closeGracefully(drainTimeout: drainTimeout)),
eagerError: false,
);
}
final http.Client _inner;
final String debugLabel;
final Set<_TrackedRequest> _active = <_TrackedRequest>{};
bool _closing = false;
bool _innerClosed = false;
Future<void>? _closeFuture;
@override
Future<http.StreamedResponse> send(http.BaseRequest request) async {
if (_closing) {
throw http.ClientException('HTTP client is closing', request.url);
}
final tracked = _TrackedRequest(request.url);
_active.add(tracked);
try {
final abortableRequest = _wrapRequest(request, tracked.abortTrigger);
final response = await _inner.send(abortableRequest);
return _wrapResponse(response, tracked);
} catch (_) {
_complete(tracked);
rethrow;
}
}
Future<void> closeGracefully({Duration drainTimeout = const Duration(seconds: 2)}) {
_closing = true;
if (_innerClosed) return Future<void>.value();
final existing = _closeFuture;
if (existing != null) return existing;
final future = _closeGracefully(drainTimeout);
_closeFuture = future;
unawaited(
future.then<void>(
(_) {
if (!_innerClosed && identical(_closeFuture, future)) {
_closeFuture = null;
}
},
onError: (Object _, StackTrace _) {
if (!_innerClosed && identical(_closeFuture, future)) {
_closeFuture = null;
}
},
),
);
return future;
}
@override
void close() {
unawaited(closeGracefully());
}
Future<void> _closeGracefully(Duration drainTimeout) async {
await _abortActive();
if (_active.isNotEmpty) {
try {
await Future.wait(_active.map((request) => request.done), eagerError: false).timeout(drainTimeout);
} on TimeoutException {
appLogger.w('HTTP client drain timed out', error: {'client': debugLabel, 'activeRequests': _active.length});
}
}
_tryCloseInner();
if (!_innerClosed) {
appLogger.w(
'HTTP client close deferred until active requests finish',
error: {'client': debugLabel, 'activeRequests': _active.length},
);
}
}
Future<void> _abortActive() async {
await Future.wait(_active.toList().map((request) => request.cancel()), eagerError: false);
}
http.BaseRequest _wrapRequest(http.BaseRequest request, Future<void> managedAbortTrigger) {
final requestAbortTrigger = request is http.Abortable ? request.abortTrigger : null;
final abortTrigger = requestAbortTrigger == null
? managedAbortTrigger
: Future.any<void>([managedAbortTrigger, requestAbortTrigger]);
final body = request.finalize();
final abortable = http.AbortableStreamedRequest(request.method, request.url, abortTrigger: abortTrigger)
..headers.addAll(request.headers)
..followRedirects = request.followRedirects
..maxRedirects = request.maxRedirects
..persistentConnection = request.persistentConnection
..contentLength = request.contentLength;
unawaited(
body.pipe(abortable.sink).catchError((Object e, StackTrace st) {
appLogger.d('HTTP request body pipe failed', error: e, stackTrace: st);
}),
);
return abortable;
}
http.StreamedResponse _wrapResponse(http.StreamedResponse response, _TrackedRequest tracked) {
late final StreamController<List<int>> controller;
StreamSubscription<List<int>>? subscription;
var subscribed = false;
var cancelledBeforeListen = false;
Future<void> cancelResponse() async {
if (tracked.isDone) return;
tracked.abort();
cancelledBeforeListen = !subscribed;
if (subscribed) {
await subscription?.cancel();
} else {
final cancelSubscription = response.stream.listen(null, onError: (_) {});
await cancelSubscription.cancel();
}
unawaited(controller.close());
_complete(tracked);
}
controller = StreamController<List<int>>(
sync: true,
onListen: () {
if (cancelledBeforeListen) {
unawaited(controller.close());
return;
}
subscribed = true;
subscription = response.stream.listen(
controller.add,
onError: controller.addError,
onDone: () {
_complete(tracked);
unawaited(controller.close());
},
);
},
onPause: () => subscription?.pause(),
onResume: () => subscription?.resume(),
onCancel: () async {
tracked.abort();
await subscription?.cancel();
_complete(tracked);
},
);
tracked.cancelResponse = cancelResponse;
if (response case http.BaseResponseWithUrl(:final url)) {
return _ManagedStreamedResponseWithUrl(
controller.stream,
response.statusCode,
url: url,
contentLength: response.contentLength,
request: response.request,
headers: response.headers,
isRedirect: response.isRedirect,
persistentConnection: response.persistentConnection,
reasonPhrase: response.reasonPhrase,
);
}
return http.StreamedResponse(
controller.stream,
response.statusCode,
contentLength: response.contentLength,
request: response.request,
headers: response.headers,
isRedirect: response.isRedirect,
persistentConnection: response.persistentConnection,
reasonPhrase: response.reasonPhrase,
);
}
void _complete(_TrackedRequest tracked) {
if (!_active.remove(tracked)) return;
tracked.complete();
if (_closing && _active.isEmpty) {
_tryCloseInner();
}
}
void _tryCloseInner() {
if (_innerClosed || _active.isNotEmpty) return;
try {
_inner.close();
_innerClosed = true;
_instances.remove(this);
} catch (e, st) {
appLogger.w('HTTP client close failed', error: e, stackTrace: st);
}
}
}
class _ManagedStreamedResponseWithUrl extends http.StreamedResponse implements http.BaseResponseWithUrl {
_ManagedStreamedResponseWithUrl(
super.stream,
super.statusCode, {
required this.url,
super.contentLength,
super.request,
super.headers,
super.isRedirect,
super.persistentConnection,
super.reasonPhrase,
});
@override
final Uri url;
}
class _TrackedRequest {
_TrackedRequest(this.url);
final Uri url;
final Completer<void> _abortCompleter = Completer<void>();
final Completer<void> _doneCompleter = Completer<void>();
Future<void> get abortTrigger => _abortCompleter.future;
Future<void> get done => _doneCompleter.future;
bool get isDone => _doneCompleter.isCompleted;
Future<void> Function()? cancelResponse;
void abort() {
if (!_abortCompleter.isCompleted) _abortCompleter.complete();
}
Future<void> cancel() async {
abort();
await cancelResponse?.call();
}
void complete() {
if (!_doneCompleter.isCompleted) _doneCompleter.complete();
}
}
lib/utils/media_server_http_client.dart
+116 -23
View File
@@ -9,6 +9,7 @@ import 'app_logger.dart';
import 'future_extensions.dart';
import 'isolate_helper.dart';
import 'log_redaction_manager.dart';
import 'managed_http_client.dart';
import '../exceptions/media_server_exceptions.dart';
// Platform-specific imports are conditional
@@ -63,6 +64,8 @@ class AbortController {
/// timeouts, logging, and optional endpoint failover.
class MediaServerHttpClient {
final http.Client _client;
final Set<AbortController> _activeAborts = <AbortController>{};
bool _closing = false;
MediaServerHttpClient({
http.Client? client,
@@ -135,58 +138,114 @@ class MediaServerHttpClient {
}) => _send('DELETE', path, queryParameters: queryParameters, headers: headers, timeout: timeout, abort: abort);
/// Fetch raw bytes (e.g. images, BIF files, subtitles).
Future<Uint8List> getBytes(String url, {Map<String, String>? headers, Duration? timeout}) async {
Future<Uint8List> getBytes(
String url, {
Map<String, String>? headers,
Duration? timeout,
AbortController? abort,
}) async {
if (_closing) {
throw MediaServerHttpException(type: MediaServerHttpErrorType.cancelled, message: 'HTTP client is closing');
}
final uri = _isAbsoluteUrl(url) ? Uri.parse(url) : _buildUri(url, null);
final request = http.Request('GET', uri);
final requestAbort = AbortController();
_activeAborts.add(requestAbort);
final request = http.AbortableRequest('GET', uri, abortTrigger: _abortTrigger(requestAbort, abort));
request.headers.addAll({...defaultHeaders, ...?headers});
final sw = Stopwatch()..start();
try {
final streamed = await _client
.send(request)
.namedTimeout(timeout ?? connectTimeout, operation: 'GET ${uri.path} connect');
final streamed = await _withAbortOnTimeout(
_client.send(request),
timeout ?? connectTimeout,
operation: 'GET ${uri.path} connect',
abort: requestAbort,
);
final bytes = await streamed.stream.toBytes().namedTimeout(
final bytes = await _withAbortOnTimeout(
streamed.stream.toBytes(),
timeout ?? receiveTimeout,
operation: 'GET ${uri.path} receive',
abort: requestAbort,
);
sw.stop();
_logResponse('GET', uri, streamed.statusCode, sw.elapsedMilliseconds);
return bytes;
} catch (e) {
requestAbort.abort();
sw.stop();
throw MediaServerHttpException.from(e, uri: uri);
} finally {
_activeAborts.remove(requestAbort);
}
}
/// Stream-download a URL directly into a file.
Future<void> downloadFile(String url, String filePath, {Map<String, String>? headers, Duration? timeout}) async {
Future<void> downloadFile(
String url,
String filePath, {
Map<String, String>? headers,
Duration? timeout,
AbortController? abort,
}) async {
if (_closing) {
throw MediaServerHttpException(type: MediaServerHttpErrorType.cancelled, message: 'HTTP client is closing');
}
final uri = _isAbsoluteUrl(url) ? Uri.parse(url) : _buildUri(url, null);
final request = http.Request('GET', uri);
final requestAbort = AbortController();
_activeAborts.add(requestAbort);
final request = http.AbortableRequest('GET', uri, abortTrigger: _abortTrigger(requestAbort, abort));
request.headers.addAll({...defaultHeaders, ...?headers});
try {
final streamed = await _client
.send(request)
.namedTimeout(timeout ?? connectTimeout, operation: 'download ${uri.path} connect');
final streamed = await _withAbortOnTimeout(
_client.send(request),
timeout ?? connectTimeout,
operation: 'download ${uri.path} connect',
abort: requestAbort,
);
final file = File(filePath);
final sink = file.openWrite();
try {
await streamed.stream.pipe(sink);
await _withAbortOnTimeout(
streamed.stream.pipe(sink),
timeout ?? receiveTimeout,
operation: 'download ${uri.path} receive',
abort: requestAbort,
);
} finally {
await sink.close();
}
} catch (e) {
requestAbort.abort();
throw MediaServerHttpException.from(e, uri: uri);
} finally {
_activeAborts.remove(requestAbort);
}
}
/// Send a streamed request (for image cache etc).
Future<http.StreamedResponse> sendStreamed(http.BaseRequest request) => _client.send(request);
void close() => _client.close();
void close() {
_closing = true;
_abortActiveRequests();
_client.close();
}
Future<void> closeGracefully({Duration drainTimeout = const Duration(seconds: 2)}) async {
_closing = true;
_abortActiveRequests();
if (_client case final ManagedHttpClient managed) {
await managed.closeGracefully(drainTimeout: drainTimeout);
} else {
_client.close();
}
}
Future<MediaServerResponse> _send(
String method,
@@ -197,30 +256,36 @@ class MediaServerHttpClient {
Duration? timeout,
AbortController? abort,
}) async {
if (_closing) {
throw MediaServerHttpException(type: MediaServerHttpErrorType.cancelled, message: 'HTTP client is closing');
}
final uri = _isAbsoluteUrl(path)
? _appendQuery(Uri.parse(path), queryParameters)
: _buildUri(path, queryParameters);
final mergedHeaders = <String, String>{...defaultHeaders, ...?headers};
final http.Request request;
if (abort != null) {
request = http.AbortableRequest(method, uri, abortTrigger: abort.trigger);
} else {
request = http.Request(method, uri);
}
final requestAbort = AbortController();
_activeAborts.add(requestAbort);
final request = http.AbortableRequest(method, uri, abortTrigger: _abortTrigger(requestAbort, abort));
request.headers.addAll(mergedHeaders);
_setBody(request, body);
final sw = Stopwatch()..start();
try {
final streamed = await _client
.send(request)
.namedTimeout(timeout ?? connectTimeout, operation: '$method ${uri.path} connect');
final streamed = await _withAbortOnTimeout(
_client.send(request),
timeout ?? connectTimeout,
operation: '$method ${uri.path} connect',
abort: requestAbort,
);
final bytes = await streamed.stream.toBytes().namedTimeout(
final bytes = await _withAbortOnTimeout(
streamed.stream.toBytes(),
timeout ?? receiveTimeout,
operation: '$method ${uri.path} receive',
abort: requestAbort,
);
sw.stop();
@@ -246,8 +311,36 @@ class MediaServerHttpClient {
requestUri: uri,
);
} catch (e) {
requestAbort.abort();
sw.stop();
throw MediaServerHttpException.from(e, uri: uri);
} finally {
_activeAborts.remove(requestAbort);
}
}
void _abortActiveRequests() {
for (final abort in _activeAborts.toList()) {
abort.abort();
}
}
Future<void> _abortTrigger(AbortController owned, AbortController? external) {
final externalTrigger = external?.trigger;
return externalTrigger == null ? owned.trigger : Future.any<void>([owned.trigger, externalTrigger]);
}
Future<T> _withAbortOnTimeout<T>(
Future<T> future,
Duration timeLimit, {
required String operation,
required AbortController abort,
}) async {
try {
return await future.namedTimeout(timeLimit, operation: operation);
} on TimeoutException {
abort.abort();
rethrow;
}
}
lib/utils/platform_http_client_io.dart
+14 -10
View File
@@ -7,6 +7,7 @@ import 'package:http/io_client.dart';
import 'package:win_http/win_http.dart';
import 'app_logger.dart';
import 'managed_http_client.dart';
/// Shared Cronet engine so all clients reuse the same connection pool.
CronetEngine? _sharedEngine;
@@ -28,7 +29,7 @@ http.Client createPlatformClient() {
enableHttp2: true,
);
_logPlatformClient('android', 'CronetClient');
return CronetClient.fromCronetEngine(_sharedEngine!);
return ManagedHttpClient(CronetClient.fromCronetEngine(_sharedEngine!), debugLabel: 'CronetClient');
}
if (Platform.isIOS || Platform.isMacOS) {
// cupertino_http relies on the objective_c FFI dylib, which isn't
@@ -36,35 +37,38 @@ http.Client createPlatformClient() {
try {
final client = CupertinoClient.defaultSessionConfiguration();
_logPlatformClient(Platform.isIOS ? 'ios' : 'macos', 'CupertinoClient');
return client;
return ManagedHttpClient(client, debugLabel: 'CupertinoClient');
} catch (e, st) {
appLogger.w('CupertinoClient init failed, falling back to IOClient', error: e, stackTrace: st);
_logPlatformClient(Platform.isIOS ? 'ios' : 'macos', 'IOClient (fallback)');
return IOClient();
return ManagedHttpClient(IOClient(), debugLabel: 'IOClient (fallback)');
}
}
if (Platform.isWindows) {
try {
final client = WinHttpClient.defaultConfiguration();
_logPlatformClient('windows', 'WinHttpClient');
return client;
return ManagedHttpClient(client, debugLabel: 'WinHttpClient');
} catch (e, st) {
appLogger.w('WinHttpClient init failed, falling back to IOClient', error: e, stackTrace: st);
_logPlatformClient('windows', 'IOClient (fallback)');
return IOClient();
return ManagedHttpClient(IOClient(), debugLabel: 'IOClient (fallback)');
}
}
_logPlatformClient(Platform.operatingSystem, 'IOClient');
return IOClient();
return ManagedHttpClient(IOClient(), debugLabel: 'IOClient');
}
http.Client createPlexApiClient() {
if (Platform.isLinux) {
_logPlatformClient('linux', 'IOClient (Plex API tuned)');
return IOClient(
HttpClient()
..maxConnectionsPerHost = 12
..idleTimeout = const Duration(seconds: 90),
return ManagedHttpClient(
IOClient(
HttpClient()
..maxConnectionsPerHost = 12
..idleTimeout = const Duration(seconds: 90),
),
debugLabel: 'IOClient (Plex API tuned)',
);
}
return createPlatformClient();
lib/watch_together/screens/watch_together_screen.dart
+3 -2
View File
@@ -97,8 +97,8 @@ class _NotInSessionViewState extends State<_NotInSessionView> with MountedSetSta
String? get _plexDisplayName => context.read<ActiveProfileProvider>().active?.displayName;
Future<void> _checkHealth() async {
final client = HttpClient();
try {
final client = HttpClient();
client.connectionTimeout = const Duration(seconds: 5);
final request = await client.getUrl(Uri.parse(WatchTogetherPeerService.healthUrlFor(_customRelayUrl)));
final response = await request.close().namedTimeout(
@@ -106,13 +106,14 @@ class _NotInSessionViewState extends State<_NotInSessionView> with MountedSetSta
operation: 'WatchTogether health check',
);
final body = await response.transform(const SystemEncoding().decoder).join();
client.close();
if (!mounted) return;
setState(() => _healthOk = response.statusCode == 200 && body.trim() == 'ok');
} catch (e) {
appLogger.w('Watch Together health check failed', error: e);
if (!mounted) return;
setState(() => _healthOk = false);
} finally {
client.close(force: true);
}
}
test/utils/http_lifecycle_test.dart
+199
View File
@@ -0,0 +1,199 @@
import 'dart:async';
import 'package:flutter_test/flutter_test.dart';
import 'package:http/http.dart' as http;
import 'package:plezy/utils/abortable_http_request.dart';
import 'package:plezy/utils/managed_http_client.dart';
void main() {
group('ManagedHttpClient', () {
test('preserves headers added by request finalization', () async {
final inner = _CapturingClient();
final client = ManagedHttpClient(inner, debugLabel: 'test');
addTearDown(() => client.closeGracefully(drainTimeout: Duration.zero));
final request = http.MultipartRequest('POST', Uri.parse('https://example.test/upload'))
..fields['title'] = 'Movie';
final response = await client.send(request);
await response.stream.drain<void>();
expect(inner.headers?['content-type'], startsWith('multipart/form-data; boundary='));
expect(inner.body, isNotEmpty);
});
test('closeGracefully aborts and cancels an active streamed response before closing inner client', () async {
final inner = _StreamingClient();
final client = ManagedHttpClient(inner, debugLabel: 'test');
addTearDown(() async {
await client.closeGracefully(drainTimeout: Duration.zero);
await inner.dispose();
});
final response = await client.send(http.Request('GET', Uri.parse('https://example.test/slow')));
await client.closeGracefully(drainTimeout: const Duration(milliseconds: 100));
expect(inner.closeCount, 1);
expect(inner.responseCancelled, isTrue);
await expectLater(inner.abortTrigger, completes);
await expectLater(response.stream.toList(), completion(isEmpty));
});
test('closeGracefully can retry after a drain timeout', () async {
final inner = _DeferredSendClient();
final client = ManagedHttpClient(inner, debugLabel: 'test');
addTearDown(() => client.closeGracefully(drainTimeout: Duration.zero));
final responseFuture = client.send(http.Request('GET', Uri.parse('https://example.test/slow')));
await Future<void>.delayed(Duration.zero);
await client.closeGracefully(drainTimeout: const Duration(milliseconds: 1));
expect(inner.closeCount, 0);
var retryCompleted = false;
final retryClose = client.closeGracefully(drainTimeout: const Duration(seconds: 1));
unawaited(retryClose.whenComplete(() => retryCompleted = true));
await Future<void>.delayed(const Duration(milliseconds: 10));
expect(retryCompleted, isFalse);
inner.completeWithEmptyResponse();
final response = await responseFuture;
await response.stream.drain<void>();
await retryClose;
expect(inner.closeCount, 1);
});
test('preserves final response URL metadata', () async {
final finalUrl = Uri.parse('https://example.test/final');
final inner = _UrlResponseClient(finalUrl);
final client = ManagedHttpClient(inner, debugLabel: 'test');
addTearDown(() => client.closeGracefully(drainTimeout: Duration.zero));
final response = await client.send(http.Request('GET', Uri.parse('https://example.test/start')));
expect(response, isA<http.BaseResponseWithUrl>().having((r) => r.url, 'url', finalUrl));
await response.stream.drain<void>();
});
});
group('sendAbortableHttpRequest', () {
test('aborts the underlying request when its timeout fires', () async {
final inner = _HangingClient();
addTearDown(inner.close);
await expectLater(
sendAbortableHttpRequest(
inner,
'GET',
Uri.parse('https://example.test/slow'),
timeout: const Duration(milliseconds: 1),
operation: 'slow request',
),
throwsA(isA<TimeoutException>().having((e) => e.message, 'message', 'slow request timed out')),
);
await expectLater(inner.abortTrigger, completes);
});
});
}
class _CapturingClient extends http.BaseClient {
Map<String, String>? headers;
List<int>? body;
@override
Future<http.StreamedResponse> send(http.BaseRequest request) async {
headers = Map.of(request.headers);
body = await request.finalize().toBytes();
return http.StreamedResponse(const Stream<List<int>>.empty(), 200, request: request);
}
}
class _StreamingClient extends http.BaseClient {
_StreamingClient() {
responseController = StreamController<List<int>>(
onCancel: () {
responseCancelled = true;
},
);
}
late final StreamController<List<int>> responseController;
Future<void>? abortTrigger;
var closeCount = 0;
var responseCancelled = false;
@override
Future<http.StreamedResponse> send(http.BaseRequest request) async {
abortTrigger = (request as http.Abortable).abortTrigger;
return http.StreamedResponse(responseController.stream, 200, request: request);
}
@override
void close() {
closeCount += 1;
}
Future<void> dispose() async {
await responseController.close();
}
}
class _DeferredSendClient extends http.BaseClient {
final _response = Completer<http.StreamedResponse>();
http.BaseRequest? request;
Future<void>? abortTrigger;
var closeCount = 0;
@override
Future<http.StreamedResponse> send(http.BaseRequest request) {
this.request = request;
abortTrigger = (request as http.Abortable).abortTrigger;
return _response.future;
}
void completeWithEmptyResponse() {
_response.complete(http.StreamedResponse(const Stream<List<int>>.empty(), 200, request: request));
}
@override
void close() {
closeCount += 1;
}
}
class _UrlResponseClient extends http.BaseClient {
_UrlResponseClient(this.url);
final Uri url;
@override
Future<http.StreamedResponse> send(http.BaseRequest request) async {
return _ResponseWithUrl(const Stream<List<int>>.empty(), 200, url: url, request: request);
}
}
class _ResponseWithUrl extends http.StreamedResponse implements http.BaseResponseWithUrl {
_ResponseWithUrl(super.stream, super.statusCode, {required this.url, super.request});
@override
final Uri url;
}
class _HangingClient extends http.BaseClient {
final _response = Completer<http.StreamedResponse>();
Future<void>? abortTrigger;
var closed = false;
@override
Future<http.StreamedResponse> send(http.BaseRequest request) {
abortTrigger = (request as http.Abortable).abortTrigger;
return _response.future;
}
@override
void close() {
closed = true;
}
}