diff --git a/lib/services/network_storage_service.dart b/lib/services/network_storage_service.dart index 10292dc2..dff52bb5 100644 --- a/lib/services/network_storage_service.dart +++ b/lib/services/network_storage_service.dart @@ -206,6 +206,17 @@ class NetworkStorageService { Future? _server; Future _mutation = Future.value(); final _routes = {}; + final _readClients = + < + ({ + String id, + String address, + String username, + String password, + String? certificate, + }), + HttpClient + >{}; static String newId() { final random = Random.secure(); @@ -249,6 +260,7 @@ class NetworkStorageService { next.add(connection); await _write(jsonEncode(next.map((c) => c.toJson()).toList())); _connections = List.unmodifiable(next); + _closeReadClients(connection.id); }); Future remove(String id) => _mutate(() async { @@ -256,6 +268,7 @@ class NetworkStorageService { await _write(jsonEncode(next.map((c) => c.toJson()).toList())); _connections = List.unmodifiable(next); _routes.removeWhere((_, route) => route.id == id); + _closeReadClients(id); }); Future connection(String id) async => @@ -331,6 +344,35 @@ class NetworkStorageService { ..idleTimeout = timeout ..autoUncompress = false; + HttpClient _readClient(NetworkConnection connection) { + // Pool only read requests with the same credentials and certificate pin. + // Uploads retain their own scoped client and cancellation lifecycle. + final key = ( + id: connection.id, + address: connection.address, + username: connection.username, + password: connection.password, + certificate: connection.trustedCertificateSha256, + ); + final client = + _readClients.remove(key) ?? (_client()..maxConnectionsPerHost = 6); + while (_readClients.length >= 8) { + _readClients.remove(_readClients.keys.first)!.close(); + } + _readClients[key] = client; + return client; + } + + void _closeReadClients(String id) { + for (final key in _readClients.keys.where((key) => key.id == id).toList()) { + // Let in-flight reads finish; new requests use the updated configuration. + _readClients.remove(key)!.close(); + } + } + + Future _discardResponse(HttpClientResponse response) => + response.listen((_) {}).cancel(); + Future _request( HttpClient client, NetworkConnection c, @@ -363,7 +405,18 @@ class NetworkStorageService { }; final HttpClientRequest request; try { - request = await client.openUrl(method, uri).timeout(timeout); + var expired = false; + final opening = client.openUrl(method, uri).then((request) { + if (expired) request.abort(); + return request; + }); + request = await opening.timeout( + timeout, + onTimeout: () { + expired = true; + throw TimeoutException('Network connection timed out'); + }, + ); } on HandshakeException { if (rejectedCertificate != null) throw rejectedCertificate!; rethrow; @@ -408,7 +461,13 @@ class NetworkStorageService { } final HttpClientResponse response; try { - response = await request.close().timeout(timeout); + response = await request.close().timeout( + timeout, + onTimeout: () { + request.abort(); + throw TimeoutException('Network response timed out'); + }, + ); } on HandshakeException { if (rejectedCertificate != null) throw rejectedCertificate!; rethrow; @@ -416,6 +475,7 @@ class NetworkStorageService { if ([301, 302, 307, 308].contains(response.statusCode)) { // Never replay a mutating request to an unselected path. if (!['GET', 'HEAD', 'PROPFIND'].contains(method)) { + await _discardResponse(response); throw const HttpException('Write destination redirected'); } final location = response.headers.value(HttpHeaders.locationHeader); @@ -429,6 +489,7 @@ class NetworkStorageService { next.path != uri.path && next.path != '${uri.path}/') || redirects >= 3) { + await _discardResponse(response); throw const HttpException('Unsupported server redirect'); } await response.drain().timeout(timeout); @@ -479,44 +540,48 @@ class NetworkStorageService { } return prepareSmbEntries(files, path); } - final client = _client(); - try { - if (c.protocol == NetworkProtocol.http && - NetworkEntry('', target.pathSegments.lastOrNull ?? '').audio) { - final response = await _request( - client, - c, - 'GET', - target, - range: 'bytes=0-0', - ); - if (response.statusCode != 200 && response.statusCode != 206) { - throw HttpException('Server returned ${response.statusCode}'); - } - return [NetworkEntry(path, target.pathSegments.last)]; - } + final client = _readClient(c); + if (c.protocol == NetworkProtocol.http && + NetworkEntry('', target.pathSegments.lastOrNull ?? '').audio) { final response = await _request( client, c, - c.protocol == NetworkProtocol.webdav ? 'PROPFIND' : 'GET', + 'GET', target, - dav: c.protocol == NetworkProtocol.webdav, - directoryListing: true, + range: 'bytes=0-0', ); - if (response.statusCode < 200 || response.statusCode >= 300) { + if (response.statusCode != 200 && response.statusCode != 206) { + await _discardResponse(response); throw HttpException('Server returned ${response.statusCode}'); } - final bytes = BytesBuilder(); - await for (final chunk in response.timeout(timeout)) { - if (bytes.length + chunk.length > 4 * 1024 * 1024) { - throw const HttpException('Directory listing is too large'); - } - bytes.add(chunk); + // Do not download a whole track when a server ignores the range. + if (response.statusCode == 206 && response.contentLength == 1) { + await response.drain().timeout(timeout); + } else { + await _discardResponse(response); } - return await parseListingInBackground(c, path, bytes.takeBytes()); - } finally { - client.close(force: true); + return [NetworkEntry(path, target.pathSegments.last)]; } + final response = await _request( + client, + c, + c.protocol == NetworkProtocol.webdav ? 'PROPFIND' : 'GET', + target, + dav: c.protocol == NetworkProtocol.webdav, + directoryListing: true, + ); + if (response.statusCode < 200 || response.statusCode >= 300) { + await _discardResponse(response); + throw HttpException('Server returned ${response.statusCode}'); + } + final bytes = BytesBuilder(); + await for (final chunk in response.timeout(timeout)) { + if (bytes.length + chunk.length > 4 * 1024 * 1024) { + throw const HttpException('Directory listing is too large'); + } + bytes.add(chunk); + } + return await parseListingInBackground(c, path, bytes.takeBytes()); } /// Large SMB listings arrive from native I/O as plain stat objects. Transfer @@ -689,7 +754,6 @@ class NetworkStorageService { } Future _serve(HttpRequest request) async { - HttpClient? client; NetworkSmbClient? smb; try { final route = _routes[request.uri.path.replaceFirst('/', '')]; @@ -735,7 +799,7 @@ class NetworkStorageService { ); } } else { - client = _client(); + final client = _readClient(c); final response = await _request( client, c, @@ -758,6 +822,8 @@ class NetworkStorageService { } if (request.method == 'GET') { await request.response.addStream(response.timeout(timeout)); + } else { + await response.drain().timeout(timeout); } } } catch (_) { @@ -765,7 +831,6 @@ class NetworkStorageService { request.response.statusCode = HttpStatus.badGateway; } catch (_) {} } finally { - client?.close(force: true); try { await smb?.close(); } catch (_) {} @@ -779,6 +844,10 @@ class NetworkStorageService { final server = _server; _server = null; _routes.clear(); + for (final client in _readClients.values) { + client.close(force: true); + } + _readClients.clear(); if (server != null) await (await server).close(force: true); } } diff --git a/test/network_storage_pool_test.dart b/test/network_storage_pool_test.dart new file mode 100644 index 00000000..84034dcf --- /dev/null +++ b/test/network_storage_pool_test.dart @@ -0,0 +1,104 @@ +import 'dart:convert'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:spotiflac_android/services/network_storage_service.dart'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + HttpOverrides.global = null; + + test( + 'WebDAV listings and ranges reuse connections; saved credentials rotate the pool', + () async { + final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + addTearDown(() => server.close(force: true)); + final ports = {}; + final auth = []; + var active = 0; + var maximum = 0; + var failListing = false; + server.listen((request) async { + ports.add(request.connectionInfo!.remotePort); + auth.add(request.headers.value('authorization')); + await request.drain(); + if (request.method == 'PROPFIND') { + request.response.statusCode = failListing ? 500 : 207; + request.response.write( + failListing + ? 'offline' + : '/music/song.flac' + '4' + 'HTTP/1.1 200 OK', + ); + } else { + expect(request.headers.value('range'), 'bytes=0-3'); + active++; + if (active > maximum) maximum = active; + await Future.delayed(const Duration(milliseconds: 15)); + request.response.statusCode = 206; + request.response.headers.set('Content-Range', 'bytes 0-3/4'); + request.response.contentLength = 4; + request.response.add([1, 2, 3, 4]); + active--; + } + await request.response.close(); + }); + final storage = NetworkStorageService( + read: () async => null, + write: (_) async {}, + ); + addTearDown(storage.dispose); + NetworkConnection connection(String password) => NetworkConnection( + id: 'pool', + name: 'Pool fixture', + protocol: NetworkProtocol.webdav, + address: 'http://127.0.0.1:${server.port}/music/', + username: 'user', + password: password, + ); + final original = connection('first'); + await storage.save(original); + final client = HttpClient(); + addTearDown(() => client.close(force: true)); + final proxy = await storage.resolve( + NetworkStorageService.source('pool', 'song.flac'), + ); + Future read() async { + final request = await client.getUrl(Uri.parse(proxy)); + request.headers.set('Range', 'bytes=0-3'); + final response = await request.close(); + expect(response.statusCode, 206); + expect(response.headers.value('content-range'), 'bytes 0-3/4'); + expect(await response.fold>([], (a, b) => a..addAll(b)), [ + 1, + 2, + 3, + 4, + ]); + } + + await storage.list(original); + await read(); + await storage.list(original); + expect(ports, hasLength(1)); + await Future.wait(List.generate(9, (_) => read())); + expect(maximum, greaterThan(1)); + expect(maximum, lessThanOrEqualTo(6)); + expect(ports.length, lessThanOrEqualTo(6)); + expect(auth.toSet(), { + 'Basic ${base64Encode(utf8.encode('user:first'))}', + }); + failListing = true; + await expectLater(storage.list(original), throwsA(isA())); + failListing = false; + expect(await storage.list(original), hasLength(1)); + final previous = ports.toSet(); + final updated = connection('second'); + await storage.save(updated); + await read(); + expect(auth.last, 'Basic ${base64Encode(utf8.encode('user:second'))}'); + expect(ports.difference(previous), hasLength(1)); + }, + ); +}