diff --git a/lib/main.dart b/lib/main.dart index b0706b899..35c204050 100644 --- a/lib/main.dart +++ b/lib/main.dart @@ -1561,15 +1561,15 @@ class _MainAppState extends State with WidgetsBindingObserver { Provider(create: (_) => ProfileConnectionRegistry(_appDatabase)), Provider( create: (context) { - // start() resolves StorageService internally — the singleton was - // already initialised eagerly during boot, so the await is a - // microtask hop in practice. + // Hydrate the disk cache eagerly for profile resolution. Live + // refresh is started only after MainScreen has settled the + // startup offline decision. final service = PlexHomeService( connections: context.read(), profileConnections: context.read(), storage: context.read(), ); - unawaited(service.start()); + unawaited(service.hydrate()); return service; }, dispose: (_, s) => s.dispose(), diff --git a/lib/profiles/active_profile_provider.dart b/lib/profiles/active_profile_provider.dart index 146b97db6..cd550c0f7 100644 --- a/lib/profiles/active_profile_provider.dart +++ b/lib/profiles/active_profile_provider.dart @@ -231,8 +231,8 @@ class ActiveProfileProvider extends ChangeNotifier with DisposableChangeNotifier Future _reloadSnapshot() async { _storage ??= await StorageService.getInstance(); // Hydrate the Plex Home cache before we read it — `_plexHome.current` - // is only populated after start() finishes its disk-cache load. - await _plexHome.start(); + // is only populated after the disk-cache load finishes. + await _plexHome.hydrate(); _localProfiles = await _registry.list(); final initialConns = await _connections.list(); diff --git a/lib/profiles/plex_home_service.dart b/lib/profiles/plex_home_service.dart index a28124d95..a2301c3f4 100644 --- a/lib/profiles/plex_home_service.dart +++ b/lib/profiles/plex_home_service.dart @@ -38,6 +38,7 @@ class PlexHomeService { final _controller = StreamController>>.broadcast(); StreamSubscription>? _connSub; Timer? _refreshTimer; + Future? _hydrateFuture; Future? _startFuture; bool _started = false; final Map _refreshGenerations = {}; @@ -56,7 +57,7 @@ class PlexHomeService { /// Emits the current snapshot immediately on subscribe, then forwards /// every change from [_controller]. Without the seed emission, late /// subscribers (e.g. the profiles management screen, which mounts long - /// after [start] fires its initial `_emit`) sit on `ConnectionState.waiting` + /// after [hydrate] fires its initial `_emit`) sit on `ConnectionState.waiting` /// forever — `combineLatest` upstream of them never fills its slot for /// this stream and the UI shows a perpetual spinner. Stream>> get stream { @@ -74,6 +75,28 @@ class PlexHomeService { return ctrl.stream; } + /// Populate the in-memory snapshot from disk without starting live work. + /// + /// Safe during the startup offline decision: this does not subscribe to + /// connection changes, install the refresh timer, or issue a refresh. + Future hydrate() { + if (_disposed) return Future.value(); + final pending = _hydrateFuture; + if (pending != null) return pending; + + final epoch = _lifecycleEpoch; + final future = _hydrate(epoch).catchError((Object error, StackTrace stackTrace) { + _hydrateFuture = null; + Error.throwWithStackTrace(error, stackTrace); + }); + _hydrateFuture = future; + return future; + } + + /// Start observing connections and refreshing Plex Home users. + /// + /// Hydration always completes first so [_onChange] cannot observe a + /// half-initialized cache or storage handle. Future start() { if (_disposed || _started) return Future.value(); final pending = _startFuture; @@ -90,11 +113,11 @@ class PlexHomeService { /// Re-read per-connection Plex Home user caches from storage. /// - /// This is used after boot-time legacy migration. The service is started + /// This is used after boot-time legacy migration. The service is hydrated /// before [ConnectionBootstrap] runs, so it may have already missed the /// copied `plex_home_users_{connectionId}` cache and new connection row. Future reloadFromStorage() async { - await start(); + await hydrate(); final epoch = _lifecycleEpoch; if (!_isLifecycleCurrent(epoch)) return; _storage ??= await StorageService.getInstance(); @@ -134,10 +157,11 @@ class PlexHomeService { if (changed && _isLifecycleCurrent(epoch)) _emit(); } - Future _start(int epoch) async { + Future _hydrate(int epoch) async { _storage ??= await StorageService.getInstance(); if (!_isLifecycleCurrent(epoch)) return; await _reloadStorageCacheIfNeeded(_storage!); + if (!_isLifecycleCurrent(epoch)) return; final initial = await _connections.list(); if (!_isLifecycleCurrent(epoch)) return; @@ -154,8 +178,12 @@ class PlexHomeService { } } _emit(); + } + Future _start(int epoch) async { + await hydrate(); if (!_isLifecycleCurrent(epoch)) return; + _connSub = _connections.watchConnections().listen(_onChange); _refreshTimer = Timer.periodic(_refreshInterval, (_) => unawaited(_refreshAll())); @@ -428,14 +456,14 @@ class PlexHomeService { /// [connectionId]. Use this instead of [materializeFirstPlexHome] in /// multi-account flows that already know which Plex account is active. Future materializePlexHomeForConnection(String connectionId) async { - await start(); + await hydrate(); return materializePlexHome(connectionId); } /// Convenience wrapper: materialize the home for the first Plex account /// in [ConnectionRegistry] (the only one most users have). Future materializeFirstPlexHome() async { - await start(); + await hydrate(); final all = await _connections.list(); final first = all.whereType().firstOrNull; if (first == null) return null; @@ -485,6 +513,7 @@ class PlexHomeService { _connSub = null; final pendingCommits = _commitBarriers.values.toList(); if (pendingCommits.isNotEmpty) await Future.wait(pendingCommits); + _hydrateFuture = null; _startFuture = null; if (!_controller.isClosed) await _controller.close(); _started = false; diff --git a/lib/screens/main_screen.dart b/lib/screens/main_screen.dart index 347f7ba67..357cc3399 100644 --- a/lib/screens/main_screen.dart +++ b/lib/screens/main_screen.dart @@ -23,6 +23,9 @@ import '../widgets/auth_error_banner.dart'; import '../widgets/app_icon.dart'; import '../utils/provider_extensions.dart'; import '../utils/platform_detector.dart'; +import '../utils/platform_http_client_stub.dart' + if (dart.library.io) '../utils/platform_http_client_io.dart' + show warmUpPlatformHttpClient; import '../utils/snackbar_helper.dart'; import '../utils/update_dialog.dart'; import '../utils/video_player_navigation.dart'; @@ -558,7 +561,18 @@ class _MainScreenState extends State _hadProfiles = activeProfile.profiles.isNotEmpty; activeProfile.addListener(_onActiveProfileChanged); _plexHomeService = context.read(); - unawaited(_plexHomeService!.start()); + // `start()` is the live/network entry point: it installs the connection + // watch, the refresh timer and an initial `_refreshAll()`. Hydration + // already happened from the provider's `create:`, so an offline launch + // keeps its cached Plex Home users without reaching the network. + // `_handleOfflineStatusChanged` starts it if we come online later. + if (!_isOffline) unawaited(_plexHomeService!.start()); + // Cronet's first `CronetEngine.build()` costs ~460 ms of synchronous JNI + // work (Play services Dynamite + GMS HTTP flags) and used to land between + // `database_ready` and `credentials_loaded`, i.e. squarely on the path to + // this screen. Android clients start on the tuned IOClient and swap to + // Cronet once this completes. + unawaited(warmUpPlatformHttpClient()); final manager = context.read().serverManager; // Read the binder so the Provider's `lazy: false` create has fired // for sure; start only in online mode so explicit startup offline does @@ -1332,6 +1346,11 @@ class _MainScreenState extends State // Ensure profile settings are warmed when coming back online if (!_isOffline) { + // A launch that started offline skipped PlexHomeService's live/network + // side, so pick it up now that we have a network. `start()` is idempotent, + // so repeated offline/online transitions are free. + final plexHome = _plexHomeService ??= context.read(); + unawaited(plexHome.start()); unawaited(() async { final mp = context.read(); final binder = context.read(); diff --git a/lib/screens/profile/borrow_connection_screen.dart b/lib/screens/profile/borrow_connection_screen.dart index 7cad4ff58..e9d78c9a0 100644 --- a/lib/screens/profile/borrow_connection_screen.dart +++ b/lib/screens/profile/borrow_connection_screen.dart @@ -81,7 +81,10 @@ class _BorrowConnectionScreenState extends State { final profileRegistry = context.read(); final plexHome = context.read(); - await plexHome.start(); + // Cache-only: this picker reads `plexHome.current` immediately and must + // not start live refresh, which would reach the network from a screen the + // user can open while offline. + await plexHome.hydrate(); final results = await Future.wait([ pcRegistry.listAll(), connRegistry.list(), diff --git a/lib/utils/managed_http_client.dart b/lib/utils/managed_http_client.dart index 7bc8170d5..9b9beb1bb 100644 --- a/lib/utils/managed_http_client.dart +++ b/lib/utils/managed_http_client.dart @@ -4,6 +4,17 @@ import 'package:http/http.dart' as http; import 'app_logger.dart'; +/// [http.Client] that can abort and drain its active requests before closing. +/// +/// Shutdown paths that must not outrun in-flight native callbacks (per-server +/// failover, server removal, app exit) await this instead of the +/// fire-and-forget [http.Client.close]. Composite clients that only delegate +/// to [ManagedHttpClient]s (`AndroidPlatformHttpClient`) implement it so the +/// awaited drain survives the extra layer. +abstract interface class GracefulHttpClient implements http.Client { + Future closeGracefully({Duration drainTimeout}); +} + /// [http.Client] wrapper that owns native-client shutdown semantics. /// /// `package:http` clients define closing with active requests as undefined. For @@ -13,7 +24,7 @@ import 'app_logger.dart'; /// active requests during shutdown, and only closes the inner client once the /// active set has drained — or, when [forceCloseOnDrainTimeout] opts in, /// force-closes a drain-resistant inner client instead of leaking its sockets. -class ManagedHttpClient extends http.BaseClient { +class ManagedHttpClient extends http.BaseClient implements GracefulHttpClient { ManagedHttpClient(this._inner, {required this.debugLabel, this.forceCloseOnDrainTimeout = false}) { _instances.add(this); } @@ -61,6 +72,7 @@ class ManagedHttpClient extends http.BaseClient { } } + @override Future closeGracefully({Duration drainTimeout = const Duration(seconds: 2)}) { _closing = true; if (_innerClosed) return Future.value(); diff --git a/lib/utils/media_server_http_client.dart b/lib/utils/media_server_http_client.dart index b6e124aaa..8dc2b3f54 100644 --- a/lib/utils/media_server_http_client.dart +++ b/lib/utils/media_server_http_client.dart @@ -237,8 +237,8 @@ class MediaServerHttpClient { Future closeGracefully({Duration drainTimeout = const Duration(seconds: 2)}) async { _closing = true; _abortActiveRequests(); - if (_client case final ManagedHttpClient managed) { - await managed.closeGracefully(drainTimeout: drainTimeout); + if (_client case final GracefulHttpClient graceful) { + await graceful.closeGracefully(drainTimeout: drainTimeout); } else { _client.close(); } diff --git a/lib/utils/platform_http_client_io.dart b/lib/utils/platform_http_client_io.dart index 56a019526..961512db4 100644 --- a/lib/utils/platform_http_client_io.dart +++ b/lib/utils/platform_http_client_io.dart @@ -13,22 +13,110 @@ import 'media_server_timeouts.dart'; /// Shared Cronet engine so all clients reuse the same connection pool. CronetEngine? _sharedEngine; bool _cronetBroken = false; +Future? _cronetWarmUp; + +const String _androidCronetLabel = 'CronetClient'; +const String _androidIoLabel = 'IOClient (Android fallback)'; const bool _tvosBuild = bool.fromEnvironment('TVOS_BUILD'); -bool _loggedPlatformClient = false; +final Set _loggedPlatformClients = {}; void _logPlatformClient(String platform, String client) { - if (_loggedPlatformClient) return; - _loggedPlatformClient = true; + if (!_loggedPlatformClients.add(client)) return; appLogger.i('Platform HTTP client', error: {'platform': platform, 'client': client}); } +/// Builds the shared Cronet engine, off the cold-start critical path. +/// +/// [CronetEngine.build] goes straight to `org.chromium.net.CronetEngine.Builder`, +/// which enumerates every registered `CronetProvider` and calls `isEnabled()` on +/// each one. `play-services-cronet` — pulled in transitively by +/// `media3-datasource-cronet` — answers that call by installing the Play services +/// Dynamite module and resolving GMS HTTP flags. `package:cronet_http` exposes no +/// way to pick a provider, so the only lever left in Dart is *when* that cost is +/// paid. Call this once the first screen is up. +/// +/// Never throws: a failed build latches Android onto the tuned IOClient. +Future warmUpPlatformHttpClient() => _cronetWarmUp ??= _buildSharedCronetEngine(); + +Future _buildSharedCronetEngine() async { + if (!Platform.isAndroid || _cronetBroken || _sharedEngine != null) return; + // Yield first: callers warm up from a post-frame callback, and the build is + // synchronous JNI work that must not land inside that frame. + await Future.delayed(Duration.zero); + try { + _sharedEngine = CronetEngine.build( + cacheMode: CacheMode.memory, + cacheMaxSize: 2 * 1024 * 1024, + enableBrotli: true, + enableHttp2: true, + ); + } catch (e, st) { + _cronetBroken = true; + _sharedEngine = null; + appLogger.w('CronetEngine build failed, staying on IOClient', error: e, stackTrace: st); + } +} + +/// Android client that starts on the tuned IOClient and swaps to Cronet as soon +/// as [warmUpPlatformHttpClient] has built the shared engine. +/// +/// Clients here are long-lived (one `MediaServerHttpClient` per server, created +/// in a constructor initializer), so resolving a delegate once at construction +/// would pin the primary media-server traffic to HTTP/1.1 forever. The delegate +/// is therefore resolved per request. Each delegate is a [ManagedHttpClient] in +/// its own right, keeping the existing shutdown semantics, and neither is +/// constructed until a request actually needs it. +class AndroidPlatformHttpClient extends http.BaseClient implements GracefulHttpClient { + ManagedHttpClient? _cronet; + ManagedHttpClient? _io; + bool _closed = false; + + @override + Future send(http.BaseRequest request) { + if (_closed) { + throw http.ClientException('HTTP client is closing', request.url); + } + return _delegate().send(request); + } + + ManagedHttpClient _delegate() { + final engine = _sharedEngine; + if (engine != null) { + final cronet = _cronet; + if (cronet != null) return cronet; + _logPlatformClient('android', _androidCronetLabel); + return _cronet = ManagedHttpClient(CronetClient.fromCronetEngine(engine), debugLabel: _androidCronetLabel); + } + final io = _io; + if (io != null) return io; + _logPlatformClient('android', _androidIoLabel); + return _io = _createIoClient(_androidIoLabel, tuned: true); + } + + @override + void close() { + _closed = true; + _cronet?.close(); + _io?.close(); + } + + @override + Future closeGracefully({Duration drainTimeout = const Duration(seconds: 2)}) async { + _closed = true; + await Future.wait([ + if (_cronet case final cronet?) cronet.closeGracefully(drainTimeout: drainTimeout), + if (_io case final io?) io.closeGracefully(drainTimeout: drainTimeout), + ], eagerError: false); + } +} + /// dart:io leaves TCP connects unbounded (Darwin retries SYNs for ~75 s) and /// `package:http` cannot abort a request whose connection is still being /// established, so every IOClient gets an explicit connect bound and /// permission to force-close after a failed drain (#1972). -http.Client _createIoClient(String debugLabel, {bool tuned = false}) { +ManagedHttpClient _createIoClient(String debugLabel, {bool tuned = false}) { final httpClient = HttpClient()..connectionTimeout = MediaServerTimeouts.connect; if (tuned) { // Plex home loads fan out many HTTP/1.1 calls; keep connections warm. @@ -41,24 +129,7 @@ http.Client _createIoClient(String debugLabel, {bool tuned = false}) { http.Client createPlatformClient() { if (Platform.isAndroid) { - if (!_cronetBroken) { - try { - _sharedEngine ??= CronetEngine.build( - cacheMode: CacheMode.memory, - cacheMaxSize: 2 * 1024 * 1024, - enableBrotli: true, - enableHttp2: true, - ); - _logPlatformClient('android', 'CronetClient'); - return ManagedHttpClient(CronetClient.fromCronetEngine(_sharedEngine!), debugLabel: 'CronetClient'); - } catch (e, st) { - _cronetBroken = true; - _sharedEngine = null; - appLogger.w('CronetClient init failed, falling back to IOClient', error: e, stackTrace: st); - } - } - _logPlatformClient('android', 'IOClient (Android fallback)'); - return _createIoClient('IOClient (Android fallback)', tuned: true); + return AndroidPlatformHttpClient(); } if (Platform.isIOS && _tvosBuild) { _logPlatformClient('tvos', 'IOClient (tvOS tuned)'); diff --git a/lib/utils/platform_http_client_stub.dart b/lib/utils/platform_http_client_stub.dart index 513718e8b..90bfe4424 100644 --- a/lib/utils/platform_http_client_stub.dart +++ b/lib/utils/platform_http_client_stub.dart @@ -5,3 +5,6 @@ import 'package:http/http.dart' as http; http.Client createPlatformClient() => throw UnsupportedError('No platform HTTP client available'); http.Client createPlexApiClient() => throw UnsupportedError('No platform HTTP client available'); + +/// No-op: only the `dart:io` implementation has an engine to warm. +Future warmUpPlatformHttpClient() async {} diff --git a/test/profiles/plex_home_service_test.dart b/test/profiles/plex_home_service_test.dart index d0c30ec27..afa4b3d27 100644 --- a/test/profiles/plex_home_service_test.dart +++ b/test/profiles/plex_home_service_test.dart @@ -248,6 +248,82 @@ void main() { expect(service.current[acct.id]!.first.uuid, 'seeded-uuid'); }); + // `start()` used to be the only entry point, and it was called from a + // provider `create:` — so it began network refresh before SetupScreen had + // decided whether the app should go straight offline. `hydrate()` is the + // disk-only half, safe to call before that decision. + test('hydrate loads the cache without issuing any network fetch', () async { + await storage.savePlexHomeUsersCache('plex.hydrate', [_user('cached-uuid').toJson()]); + final acct = _account('plex.hydrate'); + await connections.upsert(acct); + + final fetcher = _QueuedFetcher(); + addTearDown(fetcher.close); + service = PlexHomeService( + connections: connections, + profileConnections: profileConnections, + storage: storage, + plexHomeUserFetcher: fetcher.call, + ); + + await service.hydrate(); + // Give any stray unawaited refresh a chance to reach the fetcher. + await Future.delayed(Duration.zero); + + expect(service.current[acct.id]!.single.uuid, 'cached-uuid'); + expect(fetcher.requests, isEmpty, reason: 'hydration must not touch the network'); + }); + + test('hydrate still satisfies the stream replay contract', () async { + // `stream` re-emits on listen so a late listener behind a combineLatest + // does not sit on ConnectionState.waiting forever. That has to hold even + // when the live/network side never starts. + await storage.savePlexHomeUsersCache('plex.replay', [_user('replay-uuid').toJson()]); + final acct = _account('plex.replay'); + await connections.upsert(acct); + + final fetcher = _QueuedFetcher(); + addTearDown(fetcher.close); + service = PlexHomeService( + connections: connections, + profileConnections: profileConnections, + storage: storage, + plexHomeUserFetcher: fetcher.call, + ); + + await service.hydrate(); + + final first = await service.stream.first; + expect(first[acct.id]!.single.uuid, 'replay-uuid'); + expect(fetcher.requests, isEmpty); + }); + + test('concurrent hydrate calls share one pass and start implies hydration', () async { + await storage.savePlexHomeUsersCache('plex.coalesce', [_user('coalesce-uuid').toJson()]); + final acct = _account('plex.coalesce'); + await connections.upsert(acct); + + final fetcher = _QueuedFetcher(); + addTearDown(fetcher.close); + service = PlexHomeService( + connections: connections, + profileConnections: profileConnections, + storage: storage, + plexHomeUserFetcher: fetcher.call, + ); + + // Provider create and a picker opening at the same time. + await Future.wait([service.hydrate(), service.hydrate(), service.hydrate()]); + expect(service.current[acct.id]!.single.uuid, 'coalesce-uuid'); + expect(fetcher.requests, isEmpty); + + // start() is self-sufficient: it implies hydration and then goes live. + await service.start(); + await fetcher.waitForCount(1); + expect(fetcher.requests, hasLength(1), reason: 'the live side does refresh'); + expect(service.current[acct.id]!.single.uuid, 'coalesce-uuid'); + }); + test('reloadFromStorage picks up caches written after startup', () async { final refreshBlocker = Completer>(); service = PlexHomeService( diff --git a/test/screens/profile/borrow_connection_screen_test.dart b/test/screens/profile/borrow_connection_screen_test.dart index a39565087..f3720ebdc 100644 --- a/test/screens/profile/borrow_connection_screen_test.dart +++ b/test/screens/profile/borrow_connection_screen_test.dart @@ -64,10 +64,10 @@ void main() { ), ); - expect(plexHome.starts, hasLength(1)); + expect(plexHome.hydrations, hasLength(1)); expect(find.byType(CircularProgressIndicator), findsOneWidget); - plexHome.starts.first.completeError(StateError('load failed')); + plexHome.hydrations.first.completeError(StateError('load failed')); await tester.pumpAndSettle(); expect(find.text(t.profiles.borrowLoadFailed), findsOneWidget); @@ -77,11 +77,11 @@ void main() { await tester.tap(find.text(t.common.retry)); await tester.pump(); - expect(plexHome.starts, hasLength(2)); + expect(plexHome.hydrations, hasLength(2)); expect(find.byType(CircularProgressIndicator), findsOneWidget); expect(find.text(t.profiles.borrowLoadFailed), findsNothing); - plexHome.starts.last.complete(); + plexHome.hydrations.last.complete(); await tester.pumpAndSettle(); expect(find.text(t.profiles.borrowEmpty), findsOneWidget); @@ -92,12 +92,15 @@ void main() { class _ControlledPlexHomeService extends PlexHomeService { _ControlledPlexHomeService({required super.connections, required super.profileConnections, required super.storage}); - final Queue> starts = Queue(); + /// The picker reads `current` straight after awaiting, so it needs the disk + /// cache only and deliberately does not start live refresh — gate the same + /// call the screen makes. + final Queue> hydrations = Queue(); @override - Future start() { + Future hydrate() { final completer = Completer(); - starts.add(completer); + hydrations.add(completer); return completer.future; } } diff --git a/test/utils/media_server_http_client_shutdown_test.dart b/test/utils/media_server_http_client_shutdown_test.dart index c8e834f0e..698b5691e 100644 --- a/test/utils/media_server_http_client_shutdown_test.dart +++ b/test/utils/media_server_http_client_shutdown_test.dart @@ -6,6 +6,7 @@ import 'package:http/http.dart' as http; import 'package:plezy/exceptions/media_server_exceptions.dart'; import 'package:plezy/utils/managed_http_client.dart'; import 'package:plezy/utils/media_server_http_client.dart'; +import 'package:plezy/utils/platform_http_client_io.dart'; void main() { group('MediaServerHttpClient shutdown', () { @@ -110,6 +111,32 @@ void main() { expect(File(filePath).existsSync(), isFalse, reason: 'a cancelled download must not commit the final file'); expect(File('$filePath.download').existsSync(), isFalse, reason: 'the partial temp file must be cleaned up'); }); + + test('closeGracefully awaits a delegating GracefulHttpClient drain', () async { + final transport = _ManualDrainClient(); + final client = MediaServerHttpClient(client: transport, baseUrl: 'https://example.test/'); + + var closed = false; + final closing = client.closeGracefully().then((_) => closed = true); + await Future.delayed(Duration.zero); + expect(transport.drainRequested, isTrue); + expect(closed, isFalse, reason: 'must await the delegate drain, not fall back to fire-and-forget close()'); + + transport.finishDrain(); + await closing; + expect(closed, isTrue); + }); + + test('AndroidPlatformHttpClient keeps the graceful-drain contract and rejects sends once closing', () async { + final client = AndroidPlatformHttpClient(); + expect(client, isA(), reason: 'per-server awaited drains dispatch on GracefulHttpClient'); + + await client.closeGracefully(drainTimeout: Duration.zero); + expect( + () => client.send(http.Request('GET', Uri.parse('https://example.test/library/sections'))), + throwsA(isA()), + ); + }); }); } @@ -150,3 +177,24 @@ class _HangingBodyClient extends http.BaseClient { await body.close(); } } + +/// Delegating client (the AndroidPlatformHttpClient shape) whose drain +/// completes only when the test says so. +class _ManualDrainClient extends http.BaseClient implements GracefulHttpClient { + final _drained = Completer(); + bool drainRequested = false; + + @override + Future send(http.BaseRequest request) => throw UnimplementedError(); + + @override + Future closeGracefully({Duration drainTimeout = const Duration(seconds: 2)}) { + drainRequested = true; + return _drained.future; + } + + void finishDrain() => _drained.complete(); + + @override + void close() {} +}