diff --git a/CHANGELOG.md b/CHANGELOG.md index b029465..678bf66 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,21 @@ +## 0.2.0 + +This release is designed to preserve existing behavior while making caching and streaming faster and more robust. + +### Improvements + +* Improved streaming performance and efficiency when serving cached content. +* Improved reliability when resuming partially downloaded files. +* Improved handling of interrupted downloads, cache validation, and source changes. +* Improved range request handling and cache integrity checks. +* Improved cache lifecycle and cleanup behavior across platforms. + +### Fixes + +* Fixed edge cases that could cause incomplete or outdated cached data to be served. +* Fixed several issues involving partial downloads, interrupted connections, and cache file transitions. +* Fixed assorted range request and local cache server edge cases. + ## 0.1.0 This release significantly simplifies cache management. A new `getCacheUrl` API automates the full lifecycle of cache streams, eliminating the need to create or manage `HttpCacheStream` instances for most integrations. diff --git a/benchmarker/README.md b/benchmarker/README.md index 753c734..5f2024f 100644 --- a/benchmarker/README.md +++ b/benchmarker/README.md @@ -145,8 +145,4 @@ flutter test The suite covers request distribution and statistics, drives real worker isolates against a local origin server, and runs all three benchmark types -end-to-end through a real `HttpCacheManager`. - -> When testing against a **loopback** origin, address it as `localhost` rather -> than `127.0.0.1`: `http_cache_stream` rejects source URLs whose host matches -> the cache server's own host. +end-to-end through a real `HttpCacheManager`. \ No newline at end of file diff --git a/benchmarker/lib/src/benchmark/benchmark_config.dart b/benchmarker/lib/src/benchmark/benchmark_config.dart index a79d47c..f0e1436 100644 --- a/benchmarker/lib/src/benchmark/benchmark_config.dart +++ b/benchmarker/lib/src/benchmark/benchmark_config.dart @@ -194,8 +194,7 @@ class RangePlan { ); @override - String toString() => - 'RangePlan(bytes $start-$end, window $windowSize, ' + String toString() => 'RangePlan(bytes $start-$end, window $windowSize, ' 'sequential: $isSequential)'; } @@ -248,7 +247,10 @@ class BenchmarkConfig { required int? totalRequests, }) { final uri = Uri.tryParse(url.trim()); - if (url.trim().isEmpty || uri == null || !uri.hasScheme || uri.host.isEmpty) { + if (url.trim().isEmpty || + uri == null || + !uri.hasScheme || + uri.host.isEmpty) { return 'Enter a valid absolute source URL.'; } if (uri.scheme != 'http' && uri.scheme != 'https') { diff --git a/benchmarker/lib/src/benchmark/source_probe.dart b/benchmarker/lib/src/benchmark/source_probe.dart index b8e3f37..778051a 100644 --- a/benchmarker/lib/src/benchmark/source_probe.dart +++ b/benchmarker/lib/src/benchmark/source_probe.dart @@ -28,7 +28,8 @@ Future probeSource( try { try { final response = await httpClient.head(url).timeout(timeout); - if (_isSuccess(response.statusCode) && (response.contentLength ?? 0) > 0) { + if (_isSuccess(response.statusCode) && + (response.contentLength ?? 0) > 0) { return SourceInfo( contentLength: response.contentLength, acceptsRanges: _advertisesRanges(response.headers), diff --git a/benchmarker/lib/src/ui/widgets/config_panel.dart b/benchmarker/lib/src/ui/widgets/config_panel.dart index 097a774..fbd74cc 100644 --- a/benchmarker/lib/src/ui/widgets/config_panel.dart +++ b/benchmarker/lib/src/ui/widgets/config_panel.dart @@ -263,8 +263,9 @@ class _RangeSelector extends StatelessWidget { ], selected: {form.rangeMode}, showSelectedIcon: false, - onSelectionChanged: - isBusy ? null : (selection) => form.rangeMode = selection.first, + onSelectionChanged: isBusy + ? null + : (selection) => form.rangeMode = selection.first, ), ), ), diff --git a/benchmarker/lib/src/ui/widgets/stats_panel.dart b/benchmarker/lib/src/ui/widgets/stats_panel.dart index 139b0e3..44c6cb0 100644 --- a/benchmarker/lib/src/ui/widgets/stats_panel.dart +++ b/benchmarker/lib/src/ui/widgets/stats_panel.dart @@ -351,8 +351,7 @@ class _TileGrid extends StatelessWidget { spacing: spacing, runSpacing: spacing, children: [ - for (final tile in tiles) - SizedBox(width: tileWidth, child: tile), + for (final tile in tiles) SizedBox(width: tileWidth, child: tile), ], ); }, @@ -426,10 +425,9 @@ class _TimingTable extends StatelessWidget { child: Text( text, textAlign: leading ? TextAlign.left : TextAlign.right, - style: (header - ? theme.textTheme.labelSmall - : theme.textTheme.bodySmall) - ?.copyWith( + style: + (header ? theme.textTheme.labelSmall : theme.textTheme.bodySmall) + ?.copyWith( color: header ? theme.colorScheme.onSurfaceVariant : null, fontFeatures: const [FontFeature.tabularFigures()], ), diff --git a/benchmarker/lib/src/util/formatting.dart b/benchmarker/lib/src/util/formatting.dart index 7fae8d9..3bc204f 100644 --- a/benchmarker/lib/src/util/formatting.dart +++ b/benchmarker/lib/src/util/formatting.dart @@ -57,5 +57,4 @@ String formatTimestamp(DateTime time) { '${formatClockTime(local)}'; } -String _pad(int value, [int width = 2]) => - value.toString().padLeft(width, '0'); +String _pad(int value, [int width = 2]) => value.toString().padLeft(width, '0'); diff --git a/benchmarker/test/benchmark_config_test.dart b/benchmarker/test/benchmark_config_test.dart index 2e87cc7..6b1fa52 100644 --- a/benchmarker/test/benchmark_config_test.dart +++ b/benchmarker/test/benchmark_config_test.dart @@ -31,8 +31,8 @@ void main() { test('always sums to the total request count', () { for (var concurrency = 1; concurrency <= 16; concurrency++) { for (var total = concurrency; total <= 200; total += 7) { - final distribution = - _config(concurrency: concurrency, total: total).requestDistribution(); + final distribution = _config(concurrency: concurrency, total: total) + .requestDistribution(); expect(distribution, hasLength(concurrency)); expect(distribution.reduce((a, b) => a + b), total); expect(distribution.every((count) => count > 0), isTrue); @@ -100,7 +100,8 @@ void main() { test('windows start where the previous one ended', () { const requestCount = 7; - final plan = RangePlan.sequential(const ByteRange(4096, 20479), requestCount); + final plan = + RangePlan.sequential(const ByteRange(4096, 20479), requestCount); // Every window but the last starts directly after its predecessor; the // last slides back to end on the final byte, so it may overlap. @@ -138,7 +139,8 @@ void main() { test('the windows cover the whole selected range', () { for (final requestCount in [1, 2, 3, 7, 16, 100]) { - final plan = RangePlan.sequential(const ByteRange(500, 1499), requestCount); + final plan = + RangePlan.sequential(const ByteRange(500, 1499), requestCount); expect(plan.windowFor(0).start, 500); expect(plan.windowFor(requestCount - 1).end, 1499); } diff --git a/benchmarker/test/benchmark_controller_test.dart b/benchmarker/test/benchmark_controller_test.dart index ecccfa2..57292f2 100644 --- a/benchmarker/test/benchmark_controller_test.dart +++ b/benchmarker/test/benchmark_controller_test.dart @@ -58,10 +58,7 @@ void main() { await request.response.close(); } }()); - // Addressed as `localhost` rather than the bound `127.0.0.1` so the source - // host differs from the cache server's host; otherwise http_cache_stream - // treats the source URL as an already-encoded cache URL. - sourceUrl = Uri.parse('http://localhost:${origin.port}/payload.bin'); + sourceUrl = Uri.parse('http://127.0.0.1:${origin.port}/payload.bin'); cacheDir = await Directory.systemTemp.createTemp('benchmarker_test'); await HttpCacheManager.init( @@ -213,10 +210,12 @@ void main() { // Every window is the same size and together they cover the payload once. expect(stats.totalBytes, payload.length); expect(stats.avgBytesPerRequest, plan.windowSize.toDouble()); - expect(receivedRanges..sort(), [ - for (var sequence = 0; sequence < 8; sequence++) - plan.windowFor(sequence).header, - ]..sort()); + expect( + receivedRanges..sort(), + [ + for (var sequence = 0; sequence < 8; sequence++) + plan.windowFor(sequence).header, + ]..sort()); expect( controller.logs.map((entry) => entry.message), contains(contains('Sequential windows: 8 ×')), diff --git a/benchmarker/test/benchmark_report_test.dart b/benchmarker/test/benchmark_report_test.dart index 82a282c..4d2c1a5 100644 --- a/benchmarker/test/benchmark_report_test.dart +++ b/benchmarker/test/benchmark_report_test.dart @@ -108,14 +108,16 @@ void main() { expect(json['run_id'], 3); expect(json['source_url'], 'https://example.com/file.bin'); - expect(json['target_url'], 'http://127.0.0.1:4612/https/example.com/f.bin'); + expect( + json['target_url'], 'http://127.0.0.1:4612/https/example.com/f.bin'); expect(json['cache_type'], 'preCached'); expect(json['cache_type_label'], 'Pre-cached'); expect(json['status'], 'Finished'); expect(json['build_mode'], 'release'); expect(json['started_at'], _startedAt.toIso8601String()); expect(json['ended_at'], _endedAt.toIso8601String()); - expect(json['wall_duration_us'], const Duration(seconds: 5).inMicroseconds); + expect( + json['wall_duration_us'], const Duration(seconds: 5).inMicroseconds); expect(json['concurrency'], 2); expect(json['http_client'], kHttpClientOptions.first.label); @@ -155,8 +157,8 @@ void main() { }); test('renders without a config', () { - final json = jsonDecode(buildJsonReport(_result())) - as Map; + final json = + jsonDecode(buildJsonReport(_result())) as Map; expect(json['source_url'], isNull); expect((json['requests']! as Map)['completed'], 4); diff --git a/benchmarker/test/config_panel_test.dart b/benchmarker/test/config_panel_test.dart index ca2527e..7798bd1 100644 --- a/benchmarker/test/config_panel_test.dart +++ b/benchmarker/test/config_panel_test.dart @@ -45,7 +45,8 @@ void main() { (tester) async { await pumpPanel(tester); - RangeSlider slider() => tester.widget(find.byType(RangeSlider)); + RangeSlider slider() => + tester.widget(find.byType(RangeSlider)); ButtonSegment segment(RangeMode mode) => tester .widget>( find.byType(SegmentedButton), diff --git a/lib/http_cache_stream.dart b/lib/http_cache_stream.dart index dc907d9..82d5e44 100644 --- a/lib/http_cache_stream.dart +++ b/lib/http_cache_stream.dart @@ -25,6 +25,7 @@ export 'src/models/cache_files/cache_files.dart'; export 'src/models/cache_state/cache_state.dart'; export 'src/models/exceptions/http_exceptions.dart'; export 'src/models/exceptions/invalid_cache_exceptions.dart'; +export 'src/models/exceptions/partial_cache_feed_exceptions.dart'; export 'src/models/exceptions/state_errors.dart'; export 'src/models/exceptions/stream_response_exceptions.dart'; export 'src/models/http_range/http_range.dart'; diff --git a/lib/src/cache_manager/http_cache_manager.dart b/lib/src/cache_manager/http_cache_manager.dart index 1384cbf..ac17630 100644 --- a/lib/src/cache_manager/http_cache_manager.dart +++ b/lib/src/cache_manager/http_cache_manager.dart @@ -72,7 +72,9 @@ class HttpCacheManager { ///Remove when stream is disposed cacheStream.future.onComplete(() { - _streams.remove(requestKey); + if (identical(_streams[requestKey], cacheStream)) { + _streams.remove(requestKey); + } }); if (_onStreamCreated case final streamCreatedCallback?) { diff --git a/lib/src/cache_server/keep_alive_server.dart b/lib/src/cache_server/keep_alive_server.dart index f79a102..e7d7c6f 100644 --- a/lib/src/cache_server/keep_alive_server.dart +++ b/lib/src/cache_server/keep_alive_server.dart @@ -67,7 +67,6 @@ class KeepAliveServer { if (_closed) return; final prevServer = _server; - _serverSubscription?.cancel(); _server = await HttpServer.bind(address, port, shared: true); _forwardEvents(_server); diff --git a/lib/src/cache_server/local_cache_server.dart b/lib/src/cache_server/local_cache_server.dart index 6396f24..1047d93 100644 --- a/lib/src/cache_server/local_cache_server.dart +++ b/lib/src/cache_server/local_cache_server.dart @@ -80,11 +80,17 @@ class LocalCacheServer { Uri encodeSourceUrl(Uri sourceUrl) { if (sourceUrl.host == serverUri.host) { - if (!validateCacheUrl(sourceUrl)) { - throw ArgumentError( - 'Invalid source URL: $sourceUrl. The host matches the cache server host but the URL is not a valid cache URL.'); + if (validateCacheUrl(sourceUrl)) { + return sourceUrl; // Already encoded for this server. + } + // A cache server may be assigned a different port between runs. Decode a + // URL produced by an earlier instance before encoding it for this one. + // Requiring the cache server's scheme and a different port lets regular + // source URLs hosted by another local server pass through unchanged. + if (sourceUrl.scheme == serverUri.scheme && + sourceUrl.port != serverUri.port) { + sourceUrl = decodeSourceUrl(sourceUrl) ?? sourceUrl; } - return sourceUrl; //Already encoded } final defaultPort = switch (sourceUrl.scheme) { diff --git a/lib/src/cache_stream/cache_downloader/buffered_io_sink.dart b/lib/src/cache_stream/cache_downloader/buffered_io_sink.dart index 37e5ab1..456562e 100644 --- a/lib/src/cache_stream/cache_downloader/buffered_io_sink.dart +++ b/lib/src/cache_stream/cache_downloader/buffered_io_sink.dart @@ -2,17 +2,28 @@ import 'dart:async'; import 'dart:io'; import 'dart:typed_data'; +import '../../models/exceptions/partial_cache_feed_exceptions.dart'; +import '../response_streams/partial_cache_feed.dart'; + +part 'buffered_io_sink_feed.dart'; + /// An IO sink that supports adding data while flushing to disk asynchronously. class BufferedIOSink { + //Maximum number of bytes to write in a single write operation. This prevents long writes from stalling position waiters. + static const int _maxWriteSize = 256 * 1024; // 256 KB + final File file; BufferedIOSink(this.file, int initialPosition) - : _flushedBytes = initialPosition; + : _flushedBytes = initialPosition { + _feed = BufferedIOSinkFeed._(this); + } + int _flushedBytes; final _buffer = BytesBuilder(copy: false); RandomAccessFile? _openedRAF; bool _isClosed = false; Future? _flushFuture; - final List<({int position, Completer completer})> _positionWaiters = []; + late final BufferedIOSinkFeed _feed; void add(List data) { if (_isClosed) { @@ -44,56 +55,40 @@ class BufferedIOSink { while (_buffer.isNotEmpty) { final bytes = _buffer.takeBytes(); - await raf.writeFrom(bytes, 0, bytes.length); - _flushedBytes += bytes.length; - _notifyPositionWaiters(); + for (int start = 0; start < bytes.length; start += _maxWriteSize) { + final int uncappedEnd = start + _maxWriteSize; + final int end = + uncappedEnd < bytes.length ? uncappedEnd : bytes.length; + await raf.writeFrom(bytes, start, end); + _flushedBytes += end - start; + _feed._notifyPositionWaiters(); + } } _flushFuture = null; } catch (e) { - _failPositionWaiters(e); + _feed._failPositionWaiters(e); rethrow; } }(); } - /// Returns a [Future] that completes once [flushedBytes] reaches or exceeds [minFlushedBytes]. + /// Returns a [PositionWaiter] that completes once [flushedBytes] reaches or exceeds [minFlushedBytes]. /// Completes immediately if the position is already reached. - /// Fails if the sink is closed or a flush error occurs before the position is reached. - Future waitForPosition(int minFlushedBytes, - [Duration timeout = const Duration(seconds: 30)]) { - if (_flushedBytes >= minFlushedBytes) return Future.value(); - if (_isClosed) { - return Future.error(StateError( - 'BufferedIOSink closed before reaching position $minFlushedBytes')); - } - final completer = Completer(); - _positionWaiters.add((position: minFlushedBytes, completer: completer)); - return completer.future.timeout(timeout, onTimeout: () { - _positionWaiters.removeWhere((w) => w.completer == completer); - throw TimeoutException( - 'Timeout while waiting for flushedBytes to reach $minFlushedBytes', - timeout); - }); - } + /// Fails if the sink is closed, a flush error occurs before the position is reached, or the waiter is cancelled. + PositionWaiter waitForPosition(int minFlushedBytes) => + _feed.waitForPosition(minFlushedBytes); - void _notifyPositionWaiters() { - if (_positionWaiters.isEmpty) return; - for (int i = _positionWaiters.length - 1; i >= 0; i--) { - if (_flushedBytes >= _positionWaiters[i].position) { - _positionWaiters.removeAt(i).completer.complete(); - } - } - } - - void _failPositionWaiters(Object error) { - if (_positionWaiters.isEmpty) return; - for (final w in _positionWaiters) { - w.completer.completeError(error); - } - _positionWaiters.clear(); - } - - Future close({final bool flushBuffer = true}) async { + /// Closes the sink, resolving any waiters that can no longer be satisfied. + /// + /// Set [isDone] when the producer reached the end of its content. The feed is + /// then left unfailed, so readers treat [flushedBytes] as the true end of the + /// content. When [isDone] is false the download was aborted, and the feed + /// fails with [PartialCacheAbortedException] so readers do not mistake the + /// truncated content for an end of stream. + Future close({ + final bool flushBuffer = true, + final bool isDone = false, + }) async { if (_isClosed) return; _isClosed = true; @@ -103,17 +98,23 @@ class BufferedIOSink { } await flush(); //Even if !flushBuffer, ongoing flush must complete before RAF can be closed } finally { - _failPositionWaiters(StateError('BufferedIOSink closed')); _buffer.clear(); - if (_openedRAF case final RandomAccessFile raf) { - _openedRAF = null; - await raf.close(); + try { + if (_openedRAF case final RandomAccessFile raf) { + _openedRAF = null; + await raf.close(); + } + } finally { + _feed._close( + failure: isDone ? null : PartialCacheAbortedException(_flushedBytes), + ); } } } int get bufferSize => _buffer.length; int get flushedBytes => _flushedBytes; + PartialCacheFeed get feed => _feed; bool get flushed => _buffer.isEmpty && !isFlushing; bool get isFlushing => _flushFuture != null; bool get isClosed => _isClosed; diff --git a/lib/src/cache_stream/cache_downloader/buffered_io_sink_feed.dart b/lib/src/cache_stream/cache_downloader/buffered_io_sink_feed.dart new file mode 100644 index 0000000..2f79795 --- /dev/null +++ b/lib/src/cache_stream/cache_downloader/buffered_io_sink_feed.dart @@ -0,0 +1,108 @@ +part of 'buffered_io_sink.dart'; + +/// Read-only partial-cache progress backed by a [BufferedIOSink]. +final class BufferedIOSinkFeed implements PartialCacheFeed { + final BufferedIOSink _sink; + final List<_BufferedPositionWaiter> _positionWaiters = []; + bool _isClosed = false; + Object? _failure; + + BufferedIOSinkFeed._(this._sink); + + @override + int get position => _sink.flushedBytes; + + @override + bool get isClosed => _isClosed; + + @override + Object? get failure => _failure; + + @override + PositionWaiter waitForPosition(final int minPosition) { + if (position >= minPosition) { + return PositionWaiter.reached(minPosition); + } + + final failure = _failure; + if (failure != null) { + return PositionWaiter.failed(minPosition, failure); + } + + if (isClosed) { + return PositionWaiter.failed( + minPosition, + PartialCacheFeedClosedException(minPosition), + ); + } + + final waiter = _BufferedPositionWaiter(this, minPosition); + _positionWaiters.add(waiter); + return waiter; + } + + void _close({final Object? failure}) { + if (_isClosed) return; + _isClosed = true; + _failure ??= failure; + + if (_positionWaiters.isEmpty) return; + final waiters = List<_BufferedPositionWaiter>.of(_positionWaiters); + _positionWaiters.clear(); + for (final waiter in waiters) { + waiter._completeError( + _failure ?? PartialCacheFeedClosedException(waiter.minPosition), + ); + } + } + + void _notifyPositionWaiters() { + if (_positionWaiters.isEmpty) return; + final currentPosition = position; + for (int i = _positionWaiters.length - 1; i >= 0; i--) { + if (currentPosition >= _positionWaiters[i].minPosition) { + _positionWaiters.removeAt(i)._complete(); + } + } + } + + void _failPositionWaiters(final Object error) { + _failure ??= error; + if (_positionWaiters.isEmpty) return; + final waiters = List<_BufferedPositionWaiter>.of(_positionWaiters); + _positionWaiters.clear(); + for (final waiter in waiters) { + waiter._completeError(_failure!); + } + } +} + +final class _BufferedPositionWaiter extends PositionWaiter { + final BufferedIOSinkFeed _feed; + final _completer = Completer(); + + _BufferedPositionWaiter(this._feed, super.minPosition); + + @override + Future get future => _completer.future; + + @override + bool get isCompleted => _completer.isCompleted; + + @override + void cancel() { + if (_completer.isCompleted) return; + _feed._positionWaiters.remove(this); + _completer.completeError(PositionWaiterCancelledException(minPosition)); + } + + void _complete() { + if (_completer.isCompleted) return; + _completer.complete(); + } + + void _completeError(final Object error) { + if (_completer.isCompleted) return; + _completer.completeError(error); + } +} diff --git a/lib/src/cache_stream/cache_downloader/cache_downloader.dart b/lib/src/cache_stream/cache_downloader/cache_downloader.dart index 74160f4..e03ec73 100644 --- a/lib/src/cache_stream/cache_downloader/cache_downloader.dart +++ b/lib/src/cache_stream/cache_downloader/cache_downloader.dart @@ -2,15 +2,15 @@ import 'dart:async'; import 'package:http_cache_stream/src/etc/extensions/file_extensions.dart'; +import '../../etc/extensions/future_extensions.dart'; import '../../models/cache_config/stream_cache_config.dart'; import '../../models/cache_files/cache_files.dart'; -import '../../models/exceptions/http_exceptions.dart'; import '../../models/exceptions/invalid_cache_exceptions.dart'; import '../../models/metadata/cache_metadata.dart'; import '../../models/metadata/cached_response_headers.dart'; import '../../models/stream_requests/int_range.dart'; import '../../models/stream_requests/stream_request.dart'; -import '../../models/stream_response/stream_response.dart'; +import '../../models/stream_response/partial_file_stream_response.dart'; import 'buffered_io_sink.dart'; import 'downloader.dart'; @@ -18,13 +18,15 @@ class CacheDownloader { final CacheFiles _cacheFiles; final Downloader _downloader; final BufferedIOSink _sink; - final _streamController = StreamController>.broadcast(sync: true); final _completer = Completer(); int _position; - int _pendingStreamBytes = - 0; //Bytes received but not added to stream yet. These bytes will be added within the current event loop. - CachedResponseHeaders? _cachedHeaders; bool _paused = false; + + ///Used to validate cache response when resuming a previous partial download + final CachedResponseHeaders? _resumeHeaders; + + ///Headers received and validated + CachedResponseHeaders? _validatedHeaders; CacheDownloader._( final CacheMetadata cacheMetadata, final int startPosition, @@ -32,7 +34,7 @@ class CacheDownloader { ) : _cacheFiles = cacheMetadata.cacheFiles, _position = startPosition, _sink = BufferedIOSink(cacheMetadata.partialCacheFile, startPosition), - _cachedHeaders = startPosition > 0 ? cacheMetadata.headers : null; + _resumeHeaders = startPosition > 0 ? cacheMetadata.headers : null; factory CacheDownloader.construct( final CacheMetadata cacheMetadata, @@ -66,10 +68,9 @@ class CacheDownloader { downloadRange: () => IntRange(downloadPosition), onError: (error) { onError(error); - _streamController.addError(error); }, onHeaders: (cacheHttpHeaders) { - final prevHeaders = _cachedHeaders; + final prevHeaders = _validatedHeaders ?? _resumeHeaders; if (prevHeaders != null && downloadPosition > 0 && !CachedResponseHeaders.validateCacheResponse( @@ -77,20 +78,18 @@ class CacheDownloader { throw CacheSourceChangedException(sourceUrl); } - _cachedHeaders = cacheHttpHeaders; + _validatedHeaders = cacheHttpHeaders; onHeaders(cacheHttpHeaders); onPosition( downloadPosition); //Emit current position to update progress and process queued requests }, onData: (data) { + assert(_validatedHeaders != null, + 'Bad state: No validated headers onData'); _position += data.length; _sink.add(data); - _pendingStreamBytes = data.length; onPosition( downloadPosition); //Emit current position to update progress and synchronously process queued requests - _streamController.add( - data); //Add after processing queued requests. Requests may be fulfilled from the data. - _pendingStreamBytes = 0; if (_sink.bufferSize > maxBufferSize) { _downloader @@ -111,6 +110,7 @@ class CacheDownloader { }, ); } on InvalidCacheException { + _validatedHeaders = null; rethrow; } catch (e) { onError(e); @@ -119,36 +119,26 @@ class CacheDownloader { // Post-download — flush remaining data and verify cache integrity try { await _sink.close( - flushBuffer: true); //Flushes all buffered data and closes the sink + flushBuffer: true, + isDone: _downloader + .isDone, //If the source did not end, the feed is marked as aborted so readers do not treat it as an end of stream + ); //Flushes all buffered data and closes the sink } catch (e) { onError(e); } - final partialCacheLength = (await _sink.file.stat()).size; - - InvalidCacheSizeException.validate( - sourceUrl, - partialCacheLength, - downloadPosition, - ); - final sourceLength = _cachedHeaders?.sourceLength ?? + final sourceLength = _validatedHeaders?.sourceLength ?? (_downloader.isDone ? downloadPosition : null); - if (sourceLength != null && partialCacheLength == sourceLength) { + if (sourceLength != null && downloadPosition == sourceLength) { await onComplete(sourceLength); } } finally { - if (!_completer.isCompleted) { - _completer.complete(); - } if (!_sink.isClosed) { ///The sink is not closed on invalid cache exception, so we need to close it here - _sink.close(flushBuffer: false).ignore(); + await _sink.close(flushBuffer: false).ignoreResult(); } - if (!_streamController.isClosed) { - if (!_downloader.isDone) { - _streamController.addError(DownloadStoppedException(sourceUrl)); - } - _streamController.close().ignore(); + if (!_completer.isCompleted) { + _completer.complete(); } } } @@ -175,52 +165,31 @@ class CacheDownloader { bool processRequest(final StreamRequest request) { assert(!_paused); if (request.start > downloadPosition) return false; - if (!_downloader.isActive) return false; - final headers = _cachedHeaders; + final headers = _validatedHeaders; if (headers == null) return false; - request.complete(() async { - if (request.start >= streamPosition) { - return StreamResponse.fromStream( - request.range, - headers, - _streamController.stream, - streamPosition, - _downloader.streamConfig, - ); - } - + if (_downloader.isClosed && !_downloader.isDone) { final effectiveEnd = request.end ?? headers.sourceLength; - if (effectiveEnd != null && downloadPosition >= effectiveEnd) { - await _sink.waitForPosition(effectiveEnd); - return StreamResponse.fromFile(request.range, _cacheFiles, headers); + if (effectiveEnd == null || effectiveEnd > downloadPosition) { + return false; //Downloader closed and request exceeds downloaded range, cannot fulfill request } + } - final dataStreamPosition = streamPosition; - final combinedCacheResponse = StreamResponse.combined( + request.complete( + () => PartialFileStreamResponse( request.range, - headers, _cacheFiles, - _streamController.stream, - dataStreamPosition, - _downloader.streamConfig, - ); - - try { - await _sink.waitForPosition(dataStreamPosition); - return combinedCacheResponse; - } catch (_) { - combinedCacheResponse.cancel(); - rethrow; - } - }); + headers, + _sink.feed, + ), + ); return true; } - int? get sourceLength => _cachedHeaders?.sourceLength; + int? get sourceLength => + _validatedHeaders?.sourceLength ?? _resumeHeaders?.sourceLength; int get downloadPosition => _position; - int get streamPosition => downloadPosition - _pendingStreamBytes; int get filePosition => _sink.flushedBytes; Uri get sourceUrl => _downloader.sourceUrl; bool get isClosed => _completer.isCompleted; diff --git a/lib/src/cache_stream/cache_downloader/download_response_listener.dart b/lib/src/cache_stream/cache_downloader/download_response_listener.dart index e8e79d9..8b2f4fa 100644 --- a/lib/src/cache_stream/cache_downloader/download_response_listener.dart +++ b/lib/src/cache_stream/cache_downloader/download_response_listener.dart @@ -34,7 +34,9 @@ class DownloadResponseListener { cancelOnError: true, ); _timeoutTimer.start(() { - cancel(ReadTimedOutException(sourceUrl, _timeoutTimer.duration)); + cancel(isPaused + ? DownloadPausedException(sourceUrl, _timeoutTimer.duration) + : ReadTimedOutException(sourceUrl, _timeoutTimer.duration)); }); } diff --git a/lib/src/cache_stream/cache_downloader/downloader.dart b/lib/src/cache_stream/cache_downloader/downloader.dart index 28d5752..b4f9e5f 100644 --- a/lib/src/cache_stream/cache_downloader/downloader.dart +++ b/lib/src/cache_stream/cache_downloader/downloader.dart @@ -52,7 +52,7 @@ class Downloader { final readTimeout = streamConfig.readTimeout; await _pauseCounter.onResume.timeout(readTimeout, onTimeout: () => - throw ReadTimedOutException(sourceUrl, readTimeout)); + throw DownloadPausedException(sourceUrl, readTimeout)); } checkActive(); onHeaders(downloadStream.responseHeaders); @@ -70,6 +70,8 @@ class Downloader { rethrow; } else if (!isActive) { break; + } else if (e is DownloadPausedException) { + await _pauseCounter.onResume; } else { onError(e); await (_pauseCounter.isPaused diff --git a/lib/src/cache_stream/http_cache_stream.dart b/lib/src/cache_stream/http_cache_stream.dart index 0c95e76..31f1c1a 100644 --- a/lib/src/cache_stream/http_cache_stream.dart +++ b/lib/src/cache_stream/http_cache_stream.dart @@ -2,6 +2,7 @@ import 'dart:async'; import 'dart:io'; import 'package:http_cache_stream/src/cache_stream/cache_downloader/cache_downloader.dart'; +import 'package:http_cache_stream/src/etc/extensions/future_extensions.dart'; import 'package:http_cache_stream/src/models/cache_config/stream_cache_config.dart'; import 'package:http_cache_stream/src/models/cache_files/cache_files.dart'; import 'package:http_cache_stream/src/models/metadata/cached_response_headers.dart'; @@ -20,7 +21,9 @@ import '../models/exceptions/state_errors.dart'; import '../models/exceptions/stream_response_exceptions.dart'; import '../models/metadata/cache_metadata.dart'; import '../models/stream_requests/stream_request.dart'; +import '../models/stream_response/file_stream_response.dart'; import '../models/stream_response/header_stream_response.dart'; +import '../models/stream_response/range_download_stream_response.dart'; import '../models/stream_response/stream_response.dart'; /// A stream that handles downloading, caching, and serving content. @@ -50,7 +53,7 @@ class HttpCacheStream { final _validateCacheFuture = FutureRunner(); final _initFuture = FutureRunner(); Timer? _lifeCycleTimer; //Timer for auto-disposing the stream after release - late final _fileLock = Lock(); //Lock for modifying cache files + final _fileLock = Lock(); //Lock for modifying cache files final _disposeCompleter = Completer(); //Completer for the dispose future CachedResponseHeaders? @@ -101,7 +104,7 @@ class HttpCacheStream { if (responseHeaders != null && cacheState.isComplete) { final verifiedCacheState = await refreshCacheState(); if (verifiedCacheState.isComplete) { - return StreamResponse.fromFile(range, files, responseHeaders); + return FileStreamResponse(range, files, responseHeaders); } } @@ -109,7 +112,7 @@ class HttpCacheStream { if (rangeThreshold != null && range.start >= rangeThreshold && (range.start - cachePosition) >= rangeThreshold) { - return StreamResponse.fromDownload(sourceUrl, range, config); + return RangeDownloadStreamResponse.construct(sourceUrl, range, config); } if (!isDownloading) { @@ -195,12 +198,22 @@ class HttpCacheStream { _checkDisposed(); while (true) { - if ((await refreshCacheState()).isComplete) { + final state = await refreshCacheState(); + if (state.isComplete) { return files.complete; } if (!isRetained) { throw DownloadStoppedException(sourceUrl); } + + ///The content is fully downloaded, but the cache file could not be renamed because a response stream still holds the partial cache file open. + ///There is nothing left to download; wait for it to be released, then let [refreshCacheState] rename it. + if (state.sourceLength case final int sourceLength + when state.position >= sourceLength) { + await Future.delayed(const Duration(seconds: 10)); + continue; + } + try { final downloader = _cacheDownloader = CacheDownloader.construct(metadata, config); @@ -214,16 +227,15 @@ class HttpCacheStream { } }, onComplete: (sourceLength) async { - await _fileLock.synchronized( - () => files.partial.rename(files.complete.path)); final cachedHeaders = _cachedResponseHeaders!; if (cachedHeaders.sourceLength != sourceLength || - !cachedHeaders.acceptsRangeRequests) { - _setCachedResponseHeaders( + !cachedHeaders.acceptsRangeRequests || + cachedHeaders.isCompressedOrChunked) { + await _setCachedResponseHeaders( cachedHeaders.setSourceLength(sourceLength)); } - _updateCacheState(CacheState.complete(sourceLength)); - config.handleCacheCompletion(this, files.complete); + //Handles validating and renaming partial cache to complete. + await refreshCacheState(); }, onHeaders: (responseHeaders) { _setCachedResponseHeaders(responseHeaders); @@ -236,6 +248,7 @@ class HttpCacheStream { } catch (e) { if (e is InvalidCacheException) { await _resetCache(e); + if (isRetained) continue; //Retry download after resetting cache } else { _addError(e, closeRequests: true); } @@ -293,15 +306,16 @@ class HttpCacheStream { try { final downloader = _cacheDownloader; if (downloader != null) { - await downloader.cancel(); + await downloader.cancel().ignoreResult(); if (isRetained) { return; //Stream was retained again during download cancellation } } - if (!config.savePartialCache && !cacheState.isComplete) { + + if (!config.savePartialCache && !(await refreshCacheState()).isComplete) { await resetCache(); - } else if (!config.saveMetadata && cacheState.isComplete) { - _cachedResponseHeaders = null; + } else if (!config.saveMetadata && + (await refreshCacheState()).isComplete) { await _fileLock.synchronized(() async { if (await files.metadata.exists()) { await files.metadata.delete(); @@ -344,20 +358,22 @@ class HttpCacheStream { _addError(e, closeRequests: false); } finally { if (_queuedRequests.isNotEmpty && !isDownloading && isRetained) { - download().ignore(); //Restart download to fulfill pending requests + //Restart download to fulfill pending requests + Timer.run(() => download() + .ignore()); //Use Timer.run to avoid calling download() within the lock } } }); } } - void _setCachedResponseHeaders(CachedResponseHeaders headers) { + Future _setCachedResponseHeaders(CachedResponseHeaders headers) { if (!config.saveAllHeaders) { headers = headers.essentialHeaders(); } - _cachedResponseHeaders = headers; + _cachedResponseHeaders = headers; //Set synchronously - _fileLock.synchronized(() async { + return _fileLock.synchronized(() async { try { await files.metadata.parent.create(recursive: true); await files.metadata.writeAsBytes(jsonEncodeToBytes(metadata.toJson())); @@ -368,34 +384,106 @@ class HttpCacheStream { } Future refreshCacheState() async { - CacheState state; + final state = await _fileLock.synchronized(_cacheFileState); + _updateCacheState(state); + return state; + } + + Future _cacheFileState() async { + assert(_fileLock.locked, + 'fileCacheState must be called within _fileLock.synchronized()'); + final sourceLength = _cachedResponseHeaders?.sourceLength; + if (sourceLength == null) return const CacheState.zero(); + + InvalidCacheException? cacheException; + try { - state = await metadata.cacheState(); + final completeCacheStat = await files.complete.stat(); + if (completeCacheStat.type == FileSystemEntityType.file) { + InvalidCacheSizeException.validate( + sourceUrl, completeCacheStat.size, sourceLength); + return CacheState.complete(completeCacheStat.size); + } } catch (e) { - state = const CacheState.zero(); if (e is InvalidCacheException) { - _resetCache(e).ignore(); - } else { - _addError(e, closeRequests: false); + await files.complete.delete().ignoreResult(); + cacheException = e; } + _addError(e, closeRequests: false); } - _updateCacheState(state); - return state; + + try { + final partialCacheStat = await files.partial.stat(); + + if (partialCacheStat.type == FileSystemEntityType.file) { + InvalidCacheSizeException.validate( + sourceUrl, partialCacheStat.size, sourceLength, + partial: true); + + if (partialCacheStat.size == sourceLength) { + try { + await files.partial.rename(files.complete + .path); //Rename the partial cache to the complete cache + return CacheState.complete(partialCacheStat.size); + } on FileSystemException catch (e) { + final completeCacheStat = await files.complete.stat(); + if (completeCacheStat.type == FileSystemEntityType.file && + completeCacheStat.size == sourceLength) { + return CacheState.complete(completeCacheStat + .size); //Renamed by another process, treat as complete. + } + //Rename can fail if the file is open by a response stream on Windows. + if (lastErrorOrNull is! FileSystemException) { + _addError(e, + closeRequests: + false); //Prevent spamming the error log with repeated rename failures + } + } + } + + return CacheState.incomplete(partialCacheStat.size, sourceLength); + } + } catch (e) { + if (e is InvalidCacheException) { + cacheException = e; + } + _addError(e, closeRequests: false); + } + + if (cacheException != null && _cacheDownloader?.isClosed != false) { + _cachedResponseHeaders = + null; //Reset cached headers if the cache is invalid + await files.delete(partialOnly: false).ignoreResult(); + if (_queuedRequests.isNotEmpty && !isDownloading && isRetained) { + Timer.run(() => download().ignore()); + } + return const CacheState.zero(); + } + + return CacheState.incomplete(0, sourceLength); } void _updateCacheState(final CacheState cacheState) { + final previousState = _stateController.valueOrNull; + if (!_stateController.isClosed) { _stateController.add(cacheState); } - if (cacheState.isComplete && - _queuedRequests.isNotEmpty && - headers != null) { + if (!cacheState.isComplete) return; + + if (_queuedRequests.isNotEmpty && headers != null) { _queuedRequests.processAndRemove((request) { - request.complete( - () => StreamResponse.fromFile(request.range, files, headers!)); + request + .complete(() => FileStreamResponse(request.range, files, headers!)); }); } + + ///Only the transition that created the complete cache file counts as completion. + ///A first state of [CompleteCacheState] means the file already existed on disk, so nothing was completed here. + if (previousState != null && !previousState.isComplete) { + config.handleCacheCompletion(this, files.complete); + } } void _addError(final Object error, {required final bool closeRequests}) { diff --git a/lib/src/cache_stream/response_streams/buffered_data_stream.dart b/lib/src/cache_stream/response_streams/buffered_data_stream.dart deleted file mode 100644 index fe04b63..0000000 --- a/lib/src/cache_stream/response_streams/buffered_data_stream.dart +++ /dev/null @@ -1,184 +0,0 @@ -import 'dart:async'; -import 'dart:typed_data'; - -import '../../etc/extensions/stream_extensions.dart'; -import '../../models/cache_config/stream_cache_config.dart'; -import '../../models/exceptions/stream_response_exceptions.dart'; -import '../../models/stream_response/stream_response_range.dart'; - -///A stream that buffers data while waiting for a listener. -///Immediately begins buffering data from the source stream upon creation. -///Data from the source stream is clamped to the specified range. The specified range may be beyond the current position of the source stream, but never before it. -///If the buffered data exceeds the maximum buffer size, the stream is cancelled with an exception. -class BufferedDataStream extends Stream> { - final _controller = StreamController>(sync: true); - final _buffer = BytesBuilder(copy: false); - StreamSubscription>? _dataSubscription; - - BufferedDataStream({ - required final StreamRange range, - required final Stream> dataStream, - required final int dataStreamPosition, - required final StreamCacheConfig streamConfig, - }) { - if (dataStreamPosition > range.start) { - throw RangeError( - 'BufferedDataStream: dataStreamPosition ($dataStreamPosition) cannot be greater than range.start (${range.start})'); - } - final maxBufferSize = streamConfig.maxBufferSize; - bool done = false; //If source stream is done, or range end is reached - bool ready = false; //If listener is present and not paused - - void flush() { - if (_buffer.isNotEmpty) { - _controller.add(_buffer.takeBytes()); - } - ready = !_controller.isPaused; - if (done) _close(); - } - - void onDone() { - done = true; //Mark stream as done - _dataSubscription?.cancel().ignore(); - _dataSubscription = null; - if (_buffer.isEmpty) _close(); - } - - _dataSubscription = dataStream.listen( - _rangeDataHandler( - start: range.start, - end: range.end, - sourceLength: range.sourceLength, - initPosition: dataStreamPosition, - onCompletion: () => onDone(), - onData: (data) { - if (ready) { - assert(_buffer.isEmpty, - 'BufferedDataStream: Buffer should be empty when stream has listener and is not paused'); - assert(!_controller.isPaused, - 'BufferedDataStream: Stream should not be paused when ready is true'); - _controller.add(data); - } else if (_buffer.length + data.length > maxBufferSize) { - cancel(StreamResponseExceededMaxBufferSizeException(maxBufferSize)); - } else { - _buffer.add(data); - } - }, - ), - onDone: () { - _dataSubscription = null; - onDone(); - }, - onError: (e) { - if (ready) { - flush(); //Flush data before reporting error for event synchronization - _controller - .addError(e); //Allow listener to decide how to handle the error - } else { - cancel(e); - } - }, - cancelOnError: false, - ); - - _controller.onCancel = _close; //Listener cancelled, discard buffered data - _controller.onResume = flush; //Flush buffered data when listener resumes - _controller.onPause = () { - ready = false; //Buffer data while paused - }; - _controller.onListen = () { - scheduleMicrotask( - flush); //Avoid synchronously writing to listener upon listen (listener may not be ready yet) - }; - } - - void _close([Object? error]) { - if (_controller.isClosed) return; - _dataSubscription?.cancel().ignore(); - _dataSubscription = null; - _buffer.clear(); - _controller.clearCallbacks(); - if (error != null) { - _controller.addError(error); - } - _controller.close().ignore(); - } - - ///Public API to cancel the stream and discard buffered data - void cancel([Object error = const StreamResponseCancelledException()]) { - _close(error); - } - - @override - StreamSubscription> listen( - void Function(List event)? onData, { - Function? onError, - void Function()? onDone, - bool? cancelOnError, - }) { - return _controller.stream.listen( - onData, - onError: onError, - onDone: onDone, - cancelOnError: cancelOnError, - ); - } -} - -void Function(List) _rangeDataHandler({ - required final int start, - required final int? end, - required final int? sourceLength, - required final int initPosition, - required final void Function(List) onData, - required final void Function() onCompletion, -}) { - if (end != null && (sourceLength == null || sourceLength > end)) { - int currentPosition = initPosition; - return (List data) { - final int nextPosition = currentPosition + data.length; - - if (nextPosition >= end) { - final int startOffset = - start > currentPosition ? start - currentPosition : 0; - final int endOffset = end - currentPosition; - if (data.length >= endOffset && endOffset > startOffset) { - if (data.length != endOffset || startOffset > 0) { - data = data.sublist(startOffset, endOffset); //Clamp start and end - } - onData(data); - } - - onCompletion(); //Close stream when end is reached - } else if (start > currentPosition) { - final int startOffset = start - currentPosition; - if (data.length > startOffset) { - onData(data.sublist(startOffset)); - } - } else { - onData(data); - } - - currentPosition = nextPosition; - }; - } else if (start > initPosition) { - int currentPosition = initPosition; - return (List data) { - final int nextPosition = currentPosition + data.length; - - if (start > currentPosition) { - final int startOffset = start - currentPosition; - if (startOffset >= data.length) { - currentPosition = nextPosition; - return; //Skip entire chunk - } - data = data.sublist(startOffset); //Clamp start - } - - onData(data); - currentPosition = nextPosition; - }; - } else { - return onData; - } -} diff --git a/lib/src/cache_stream/response_streams/combined_data_stream.dart b/lib/src/cache_stream/response_streams/combined_data_stream.dart deleted file mode 100644 index 9fec3ea..0000000 --- a/lib/src/cache_stream/response_streams/combined_data_stream.dart +++ /dev/null @@ -1,116 +0,0 @@ -import 'dart:async'; - -import '../../etc/extensions/stream_extensions.dart'; -import '../../models/cache_config/stream_cache_config.dart'; -import '../../models/cache_files/cache_files.dart'; -import '../../models/exceptions/stream_response_exceptions.dart'; -import '../../models/stream_requests/int_range.dart'; -import '../../models/stream_response/stream_response_range.dart'; -import 'buffered_data_stream.dart'; -import 'cache_file_stream.dart'; - -///A stream that combines data from cache file and data stream -class CombinedDataStream extends Stream> { - final CacheFileStream _fileStream; - final BufferedDataStream _dataStream; - final _controller = StreamController>(sync: true); - CombinedDataStream._(this._fileStream, this._dataStream) { - _controller.onCancel = _close; - _controller.onPause = () => _currentSubscription?.pause(); - _controller.onResume = () => _currentSubscription?.resume(); - _controller.onListen = _start; - } - - factory CombinedDataStream( - final IntRange range, - final CacheFiles cacheFiles, - final Stream> dataStream, - final int dataStreamPosition, - final int? sourceLength, - final StreamCacheConfig streamConfig, - ) { - return CombinedDataStream._( - CacheFileStream( - StreamRange.validate(range.start, dataStreamPosition, - sourceLength), //Read upto dataStreamPosition from file - cacheFiles, - ), - BufferedDataStream( - range: StreamRange.validate(dataStreamPosition, range.end, - sourceLength), //Read from dataStreamPosition to range.end from data stream - dataStream: dataStream, - dataStreamPosition: dataStreamPosition, - streamConfig: streamConfig, - ), - ); - } - - void _start() { - void subscribe( - {required final Stream> stream, - required final void Function() onDone}) { - assert(_currentSubscription == null, - 'CombinedCacheStreamResponse: subscribe: _currentSubscription should be null when subscribing to a new stream'); - _currentSubscription = stream.listen( - _controller.add, - onError: (e) { - _currentSubscription = null; - _close(e); - }, - onDone: () { - _currentSubscription = null; - onDone(); - }, - cancelOnError: true, - ); - } - - try { - subscribe( - stream: _fileStream, //Start with file stream - onDone: () { - subscribe( - stream: _dataStream, //Then switch to data stream - onDone: _close, //Close controller when done - ); - }, - ); - } catch (e) { - _close(e); - } - } - - void _close([Object? error]) { - if (_controller.isClosed) return; - _dataStream.cancel(); //Always cancel data stream to free buffered data - _currentSubscription?.cancel().ignore(); - _currentSubscription = null; - _controller.clearCallbacks(); - if (error != null) { - _controller.addError(error); - } - _controller.close().ignore(); - } - - ///Public API to cancel the stream and discard buffered data - void cancel([Object error = const StreamResponseCancelledException()]) { - _close(error); - } - - StreamSubscription>? _currentSubscription; - - @override - StreamSubscription> listen( - void Function(List event)? onData, { - Function? onError, - void Function()? onDone, - bool? cancelOnError, - }) { - return _controller.stream.listen( - onData, - onError: onError, - onDone: onDone, - cancelOnError: cancelOnError, - ); - } -} diff --git a/lib/src/cache_stream/response_streams/partial_cache_feed.dart b/lib/src/cache_stream/response_streams/partial_cache_feed.dart new file mode 100644 index 0000000..dca1f1b --- /dev/null +++ b/lib/src/cache_stream/response_streams/partial_cache_feed.dart @@ -0,0 +1,137 @@ +import 'dart:async'; + +import '../../models/exceptions/partial_cache_feed_exceptions.dart'; + +/// A read-only view of the bytes committed to a partial cache file. +abstract interface class PartialCacheFeed { + /// Creates a feed that is already closed at [finalPosition]. + const factory PartialCacheFeed.completed(final int finalPosition) = + CompletedPartialCacheFeed; + + /// The exclusive end position currently safe to read from the cache file. + int get position; + + /// Whether the producer can no longer commit additional bytes. + bool get isClosed; + + /// The error this feed ended with, or null if it is still open or reached the + /// end of its content cleanly. + /// + /// A closed feed with no [failure] means [position] is the true end of the + /// content. A closed feed with a [failure] stopped short of it, so readers + /// that do not know the content length must not treat it as an end of stream. + Object? get failure; + + /// Returns a waiter that completes once [position] reaches or exceeds + /// [minPosition]. + /// + /// The waiter is returned synchronously and may already be completed. It + /// fails if the feed fails, closes before reaching the requested position, or + /// is cancelled via [PositionWaiter.cancel]. Callers that no longer need the + /// position must cancel the waiter to release it. + PositionWaiter waitForPosition(int minPosition); +} + +/// A partial-cache feed whose final position is already known. +/// +/// This feed is closed successfully from construction. Requests at or before +/// [position] complete immediately; later requests fail because the cache file +/// cannot grow any further. +final class CompletedPartialCacheFeed implements PartialCacheFeed { + @override + final int position; + + const CompletedPartialCacheFeed(this.position) + : assert(position >= 0, 'The final position cannot be negative.'); + + @override + bool get isClosed => true; + + @override + Object? get failure => null; + + @override + PositionWaiter waitForPosition(final int minPosition) { + if (position >= minPosition) { + return PositionWaiter.reached(minPosition); + } + return PositionWaiter.failed( + minPosition, + PartialCacheFeedClosedException(minPosition), + ); + } +} + +/// A request for a [PartialCacheFeed] to reach [minPosition]. +/// +/// Returned synchronously by [PartialCacheFeed.waitForPosition], and may +/// already be completed. Await [future] to observe the result, or call [cancel] +/// to abandon a wait that is no longer needed. +abstract class PositionWaiter implements Comparable { + /// The position [PartialCacheFeed.position] must reach for [future] to + /// complete successfully. + final int minPosition; + const PositionWaiter(this.minPosition); + + /// Creates a waiter for a position that has already been reached. + factory PositionWaiter.reached(final int minPosition) = + _CompletedPositionWaiter.reached; + + /// Creates a waiter for a position that can no longer be reached. + factory PositionWaiter.failed( + final int minPosition, + final Object error, + ) = _CompletedPositionWaiter.failed; + + /// Completes once the feed reaches [minPosition]. + Future get future; + + /// Whether [future] has already completed, successfully or otherwise. + bool get isCompleted; + + /// Abandons the wait, releasing it from the feed. + void cancel(); + + @override + int compareTo(final PositionWaiter other) => + minPosition.compareTo(other.minPosition); + + @override + String toString() => + '$runtimeType(minPosition: $minPosition, isCompleted: $isCompleted)'; +} + +/// A waiter that was already resolved when it was created. +/// +/// Feed implementations can use this for positions that have already been +/// reached or can no longer be reached. +final class _CompletedPositionWaiter extends PositionWaiter { + @override + final Future future; + + _CompletedPositionWaiter.reached(super.minPosition) + : future = Future.value(); + + _CompletedPositionWaiter.failed( + super.minPosition, + final Object error, + ) : future = Future.error(error); + + @override + bool get isCompleted => true; + + @override + void cancel() {} +} + +/// Thrown when a [PositionWaiter] is cancelled before its position is reached. +class PositionWaiterCancelledException implements Exception { + /// The position that was being waited for. + final int minPosition; + const PositionWaiterCancelledException(this.minPosition); + + @override + String toString() => + 'PositionWaiterCancelledException: Cancelled while waiting for partial ' + 'cache position $minPosition'; +} diff --git a/lib/src/cache_stream/response_streams/partial_cache_file_stream.dart b/lib/src/cache_stream/response_streams/partial_cache_file_stream.dart new file mode 100644 index 0000000..fe48c53 --- /dev/null +++ b/lib/src/cache_stream/response_streams/partial_cache_file_stream.dart @@ -0,0 +1,378 @@ +import 'dart:async'; +import 'dart:io'; +import 'dart:math'; + +import '../../models/cache_files/cache_files.dart'; +import '../../models/exceptions/partial_cache_feed_exceptions.dart'; +import '../../models/stream_response/stream_response_range.dart'; +import 'partial_cache_feed.dart'; + +/// Streams committed bytes from a partial cache file while it is being saved. +/// +/// Every listener owns its file handle and read position. When it catches up to +/// the committed position exposed by [feed], it waits for more data while +/// retaining the same file handle. +/// +/// No timeout is applied while waiting for the feed to advance; the listener is +/// responsible for bounding how long it is willing to wait. +class PartialCacheFileStream extends Stream> { + final StreamRange range; + final CacheFiles cacheFiles; + final PartialCacheFeed feed; + const PartialCacheFileStream(this.range, this.cacheFiles, this.feed); + + @override + StreamSubscription> listen( + final void Function(List event)? onData, { + final Function? onError, + final void Function()? onDone, + final bool? cancelOnError, + }) { + return _PartialCacheFileReader(range, cacheFiles, feed).stream.listen( + onData, + onError: onError, + onDone: onDone, + cancelOnError: cancelOnError, + ); + } +} + +/// Reads one range of a partial cache file into a single-subscription stream. +/// +/// Reads are pipelined like dart:io's File.openRead(): after one asynchronous +/// file read completes, the next eligible read is started before the current +/// block is synchronously emitted. This allows file I/O for the next block to +/// overlap downstream processing of the current block. +/// +/// The reader never starts a read beyond [PartialCacheFeed.position]. A paused +/// listener stops further read-ahead after at most the single read that was +/// already in flight, matching File.openRead()'s bounded read-ahead behavior. +class _PartialCacheFileReader { + static const int _maxReadSize = 256 * 1024; + final CacheFiles _cacheFiles; + final PartialCacheFeed _feed; + final _controller = StreamController>(sync: true); + final int? _requestedEnd; + + int _readPosition; + RandomAccessFile? _raf; + PositionWaiter? _positionWaiter; + bool _readInProgress = false; + bool _closing = false; + final _closeCompleter = Completer(); + + _PartialCacheFileReader( + final StreamRange range, + this._cacheFiles, + this._feed, + ) : _requestedEnd = range.absoluteEnd, + _readPosition = range.start { + _controller.onListen = _start; + _controller.onResume = _pump; + _controller.onCancel = _finish; + } + + Stream> get stream => _controller.stream; + + /// If the listener is gone, either because it cancelled or because the + /// stream was closed. + bool get _isDone => _controller.isClosed || !_controller.hasListener; + + bool get _atRequestedEnd => + _requestedEnd != null && _readPosition >= _requestedEnd; + + /// Performs the one-time setup. The hot read path is callback-driven rather + /// than an async/await loop so each block can schedule the next read before + /// the current block is emitted. + Future _start() async { + RandomAccessFile? openedRaf; + + try { + if (_isDone || _closing || _atRequestedEnd) { + await _finish(); + return; + } + + // Wait for the first requested byte before opening; the cache file may + // not exist yet. + while (_readPosition >= _feed.position) { + if (_requestedEnd == null && _feed.isClosed) { + _finishAtEndOfContent(); + return; + } + + await _awaitPosition(_readPosition + 1); + if (_isDone || _closing) { + await _finish(); + return; + } + } + + openedRaf = await _openActiveCacheFile(); + if (_isDone || _closing) { + return; + } + + if (_readPosition > 0) { + await openedRaf.setPosition(_readPosition); + if (_isDone || _closing) { + return; + } + } + + _raf = openedRaf; + openedRaf = null; + _pump(); + } on PositionWaiterCancelledException { + // Cancelled while waiting for the feed; the listener is gone. + await _finish(); + } catch (e, stackTrace) { + _handleError(e, stackTrace); + } finally { + if (openedRaf != null) { + try { + await openedRaf.close(); + } catch (_) { + // Intentionally ignored. + } + } + } + } + + /// Starts the next piece of work when possible. + void _pump() { + if (_closing || _controller.isPaused) return; + if (_readInProgress || _positionWaiter != null) return; + + if (_atRequestedEnd) { + _finish().ignore(); + return; + } + + final raf = _raf; + if (raf == null) return; + + final feedPosition = _feed.position; + final committedEnd = min(feedPosition, _requestedEnd ?? feedPosition); + final availableBytes = committedEnd - _readPosition; + + if (availableBytes <= 0) { + if (_requestedEnd == null && _feed.isClosed) { + _finishAtEndOfContent(); + } else { + _waitForPosition(_readPosition + 1); + } + return; + } + + _startRead(raf, min(_maxReadSize, availableBytes)); + } + + void _startRead(final RandomAccessFile raf, final int byteCount) { + assert(!_readInProgress); + assert(byteCount > 0); + + _readInProgress = true; + raf.read(byteCount).then( + _onRead, + onError: (Object e, StackTrace stackTrace) { + _readInProgress = false; + _handleError(e, stackTrace); + }, + ); + } + + void _onRead(final List bytes) { + _readInProgress = false; + + if (_closing || _isDone) { + _finish().ignore(); + return; + } + + final raf = _raf; + if (raf == null) { + _handleError( + StateError('Partial cache file closed while a read was in progress'), + StackTrace.current, + ); + return; + } + + if (bytes.isEmpty) { + _handleError( + FileSystemException( + 'Partial cache file ended before its committed position', + raf.path, + ), + StackTrace.current, + ); + return; + } + + _readPosition += bytes.length; + + // Match dart:io File.openRead(): start the next eligible read (or register + // the next feed wait) before synchronously delivering this block. If the + // listener pauses while handling this block, at most one read is already + // in flight and no further read is started until onResume calls _pump(). + _scheduleReadAhead(); + _controller.add(bytes); + + // Handles range completion, feed closure, or a pause that occurred while + // the current block was being emitted. If read-ahead was already started, + // _pump() is a no-op because _readInProgress/a waiter is active. + _pump(); + } + + /// Schedules only work that is safe to begin before the current block is + /// emitted. End-of-stream handling is deliberately left to [_pump] after the + /// emission so the final block is never closed out before it is delivered. + void _scheduleReadAhead() { + if (_controller.isPaused || _atRequestedEnd) { + return; + } + assert(!_readInProgress); + assert(_positionWaiter?.isCompleted != false); + + final raf = _raf; + if (raf == null) return; + + final feedPosition = _feed.position; + final committedEnd = min(feedPosition, _requestedEnd ?? feedPosition); + final availableBytes = committedEnd - _readPosition; + + if (availableBytes > 0) { + _startRead(raf, min(_maxReadSize, availableBytes)); + } else if (!_feed.isClosed) { + _waitForPosition(_readPosition + 1); + } + } + + void _waitForPosition(final int minPosition) { + if (_closing || _isDone || _positionWaiter?.isCompleted == false) return; + + final waiter = _feed.waitForPosition(minPosition); + _positionWaiter = waiter; + + waiter.future.then( + (_) { + if (!identical(_positionWaiter, waiter)) return; + _positionWaiter = null; + _pump(); + }, + onError: (Object e, StackTrace stackTrace) { + if (identical(_positionWaiter, waiter)) { + _positionWaiter = null; + } + _handleError(e, stackTrace); + }, + ); + } + + Future _awaitPosition(final int minPosition) async { + assert( + _positionWaiter?.isCompleted != false, + 'A previous position waiter is still pending; only one can be awaited at a time.', + ); + assert( + !_isDone, + 'Registering a position waiter after the listener is gone; it will never be cancelled.', + ); + + final waiter = _feed.waitForPosition(minPosition); + _positionWaiter = waiter; + try { + await waiter.future; + } finally { + if (identical(_positionWaiter, waiter)) { + _positionWaiter = null; + } + } + } + + void _handleError(final Object e, final StackTrace stackTrace) { + if (_closing || _isDone) { + _finish().ignore(); + return; + } + + if (e is PositionWaiterCancelledException) { + _finish().ignore(); + return; + } + + if (e is PartialCacheFeedClosedException && _requestedEnd == null) { + _finishAtEndOfContent(); + return; + } + + _controller.addError(e, stackTrace); + _finish().ignore(); + } + + /// Ends a read with no known end position, now that the feed is closed. + /// + /// A closed feed is the only end-of-content signal available when the content + /// length is unknown, so an aborted download has to be reported as an error. + /// Returning normally would hand the listener a truncated body it would + /// accept as complete. + void _finishAtEndOfContent() { + if (_closing || _isDone) { + _finish().ignore(); + return; + } + if (_feed.failure case final Object failure) { + _controller.addError(failure); + } + + _finish().ignore(); + } + + /// Stops scheduling work and closes the file once any in-flight read has + /// completed. This avoids closing a RandomAccessFile underneath raf.read(). + Future _finish() { + if (_closeCompleter.isCompleted) return _closeCompleter.future; + + _closing = true; + _positionWaiter?.cancel(); + + if (!_readInProgress) { + _closeResources(); + } + + return _closeCompleter.future; + } + + void _closeResources() async { + if (_closeCompleter.isCompleted || _readInProgress) return; + + final raf = _raf; + _raf = null; + + try { + await raf?.close(); + } catch (_) { + // Intentionally ignored. + } finally { + _controller.close().ignore(); + if (!_closeCompleter.isCompleted) { + _closeCompleter.complete(); + } + } + } + + Future _openActiveCacheFile() async { + assert( + !_isDone, + 'The listener is gone; the read loop should not be running.', + ); + try { + return await _cacheFiles.activeCacheFile().open(mode: FileMode.read); + } on FileSystemException { + // The partial file may have been renamed after activeCacheFile() selected + // it. Resolve the active path again and retry once. + return _cacheFiles.activeCacheFile().open(mode: FileMode.read); + } + } +} diff --git a/lib/src/etc/extensions/future_extensions.dart b/lib/src/etc/extensions/future_extensions.dart index 6aa182c..40e0e78 100644 --- a/lib/src/etc/extensions/future_extensions.dart +++ b/lib/src/etc/extensions/future_extensions.dart @@ -9,4 +9,10 @@ extension FutureExtensions on Future { action(); } } + + Future ignoreResult() async { + try { + await this; + } catch (_) {} + } } diff --git a/lib/src/models/cache_config/cache_config.dart b/lib/src/models/cache_config/cache_config.dart index f5c958a..4cc42ae 100644 --- a/lib/src/models/cache_config/cache_config.dart +++ b/lib/src/models/cache_config/cache_config.dart @@ -36,7 +36,6 @@ abstract interface class CacheConfiguration { ///The maximum amount of data (in bytes) to buffer in memory. ///If an ongoing cache download is receiving data faster than it can be written to disk, and the buffer exceeds this size, the download will be paused until the buffer is flushed to disk. - ///If a response stream is receiving data faster than it can be consumed, and the buffer exceeds this size, then the stream will be cancelled with an exception. ///Default is 25MB. int get maxBufferSize; set maxBufferSize(int value); diff --git a/lib/src/models/exceptions/http_exceptions.dart b/lib/src/models/exceptions/http_exceptions.dart index b7fbd74..9212238 100644 --- a/lib/src/models/exceptions/http_exceptions.dart +++ b/lib/src/models/exceptions/http_exceptions.dart @@ -40,6 +40,21 @@ class ReadTimedOutException extends DownloadException } } +/// Thrown when a paused download does not resume within the configured timeout. +/// This exception is intentional - it prevents a paused download from hanging indefinitely. +class DownloadPausedException extends DownloadException + implements TimeoutException, http.ClientException { + @override + final Duration duration; + DownloadPausedException(Uri uri, this.duration) + : super(uri, 'Timed out after $duration'); + + @override + String toString() { + return 'DownloadPausedException: Paused download from $uri timed out after $duration'; + } +} + class HttpStatusCodeException extends DownloadException { HttpStatusCodeException(Uri url, int expected, int result) : super( diff --git a/lib/src/models/exceptions/invalid_cache_exceptions.dart b/lib/src/models/exceptions/invalid_cache_exceptions.dart index 93773a5..8baa997 100644 --- a/lib/src/models/exceptions/invalid_cache_exceptions.dart +++ b/lib/src/models/exceptions/invalid_cache_exceptions.dart @@ -77,9 +77,11 @@ class InvalidCacheSizeException extends InvalidCacheException { static void validate( final Uri url, final int size, - final int expected, - ) { + final int expected, { + final bool partial = false, + }) { if (size == expected) return; + if (partial && size < expected) return; if (expected == 0 && size == -1) { //Accept non-existent cache as valid if expected length is 0 diff --git a/lib/src/models/exceptions/partial_cache_feed_exceptions.dart b/lib/src/models/exceptions/partial_cache_feed_exceptions.dart new file mode 100644 index 0000000..0ec3285 --- /dev/null +++ b/lib/src/models/exceptions/partial_cache_feed_exceptions.dart @@ -0,0 +1,31 @@ +/// Thrown when a cleanly closed cache download cannot reach a requested +/// position. +/// +/// Readers without a known end position may interpret this as end of content. +/// Readers with a requested end must retain the error because the feed ended +/// before satisfying their range. +class PartialCacheFeedClosedException extends StateError { + /// The position the closed feed could not reach. + final int minPosition; + + PartialCacheFeedClosedException(this.minPosition) + : super( + 'Partial cache feed closed before reaching position $minPosition', + ); +} + +/// Thrown when a cache download stops before reaching the end of its +/// content, because the download that fills it was aborted. +/// +/// Distinguishes an aborted download from a clean end of content, which readers +/// that do not know the content length cannot tell apart from [position] alone. +class PartialCacheAbortedException implements Exception { + /// The position the feed stopped at. + final int position; + const PartialCacheAbortedException(this.position); + + @override + String toString() => + 'PartialCacheAbortedException: Download aborted at position $position, ' + 'before the end of the content'; +} diff --git a/lib/src/models/exceptions/stream_response_exceptions.dart b/lib/src/models/exceptions/stream_response_exceptions.dart index 9b1f369..5b38892 100644 --- a/lib/src/models/exceptions/stream_response_exceptions.dart +++ b/lib/src/models/exceptions/stream_response_exceptions.dart @@ -13,6 +13,7 @@ class StreamResponseCancelledException extends StreamResponseException { : super('StreamResponse was cancelled'); } +@Deprecated('No longer used, will be removed in future versions') class StreamResponseExceededMaxBufferSizeException extends StreamResponseException { const StreamResponseExceededMaxBufferSizeException(int maxBufferSize) diff --git a/lib/src/models/http_range/http_range.dart b/lib/src/models/http_range/http_range.dart index 3d51896..cc56430 100644 --- a/lib/src/models/http_range/http_range.dart +++ b/lib/src/models/http_range/http_range.dart @@ -36,7 +36,13 @@ abstract class HttpRange { return false; } if (previous.end != null && next.end != null) { - if (previous.end != next.end) return false; + if (previous.end != next.end) { + final sourceLength = next.sourceLength; + final isClampedToSourceEnd = sourceLength != null && + previous.end! >= sourceLength && + next.end == sourceLength - 1; + if (!isClampedToSourceEnd) return false; + } } if (previous.sourceLength != null && next.sourceLength != null) { if (previous.sourceLength != next.sourceLength) return false; diff --git a/lib/src/models/metadata/cache_metadata.dart b/lib/src/models/metadata/cache_metadata.dart index 91ac2c9..8a638bd 100644 --- a/lib/src/models/metadata/cache_metadata.dart +++ b/lib/src/models/metadata/cache_metadata.dart @@ -1,6 +1,5 @@ import 'dart:io'; -import 'package:http_cache_stream/src/etc/extensions/file_extensions.dart'; import 'package:http_cache_stream/src/models/cache_files/cache_files.dart'; import 'package:http_cache_stream/src/models/metadata/cached_response_headers.dart'; @@ -42,26 +41,37 @@ class CacheMetadata { final sourceLength = this.sourceLength; if (sourceLength == null) return const CacheState.zero(); - final completeCacheSize = await cacheFile.lengthOrNull(); - if (completeCacheSize != null) { + final completeCacheStat = await cacheFile.stat(); + if (completeCacheStat.type == FileSystemEntityType.file) { InvalidCacheSizeException.validate( - sourceUrl, completeCacheSize, sourceLength); - return CacheState.complete(completeCacheSize); + sourceUrl, completeCacheStat.size, sourceLength); + return CacheState.complete(completeCacheStat.size); } - final partialCacheSize = await partialCacheFile.lengthOrNull(); - if (partialCacheSize == null || partialCacheSize <= 0) { - return const CacheState.zero(); - } else if (partialCacheSize == sourceLength) { - await partialCacheFile.rename( - cacheFile.path); //Rename the partial cache to the complete cache - return CacheState.complete(partialCacheSize); - } else if (partialCacheSize > sourceLength) { - throw InvalidCacheSizeException( - sourceUrl, partialCacheSize, sourceLength); - } else { - return CacheState.incomplete(partialCacheSize, sourceLength); + final partialCachStat = await partialCacheFile.stat(); + if (partialCachStat.type == FileSystemEntityType.file) { + InvalidCacheSizeException.validate( + sourceUrl, partialCachStat.size, sourceLength, + partial: true); + + if (partialCachStat.size == sourceLength) { + try { + await partialCacheFile.rename(cacheFile.path); + return CacheState.complete(partialCachStat.size); + } on FileSystemException { + final completeCacheStat = await cacheFile.stat(); + if (completeCacheStat.type == FileSystemEntityType.file && + completeCacheStat.size == sourceLength) { + return CacheState.complete(completeCacheStat + .size); //Renamed by another process, treat as complete. + } + } + } + + return CacheState.incomplete(partialCachStat.size, sourceLength); } + + return CacheState.incomplete(0, sourceLength); } ///Returns true if the cache is complete. Returns false if the cache is incomplete or does not exist. diff --git a/lib/src/models/stream_requests/int_range.dart b/lib/src/models/stream_requests/int_range.dart index 649b3cf..8ef3803 100644 --- a/lib/src/models/stream_requests/int_range.dart +++ b/lib/src/models/stream_requests/int_range.dart @@ -23,7 +23,7 @@ class IntRange implements Comparable { throw RangeError.range(start, 0, max, 'start'); } if (end != null && end > max) { - throw RangeError.range(end, start, max, 'end'); + end = max; } } return IntRange(start, end); diff --git a/lib/src/models/stream_response/cache_download_stream_response.dart b/lib/src/models/stream_response/cache_download_stream_response.dart deleted file mode 100644 index e681720..0000000 --- a/lib/src/models/stream_response/cache_download_stream_response.dart +++ /dev/null @@ -1,44 +0,0 @@ -import 'dart:async'; - -import '../../cache_stream/response_streams/buffered_data_stream.dart'; -import '../cache_config/stream_cache_config.dart'; -import '../exceptions/stream_response_exceptions.dart'; -import '../metadata/cached_response_headers.dart'; -import '../stream_requests/int_range.dart'; -import 'stream_response.dart'; -import 'stream_response_range.dart'; - -/// A stream response that buffers data from the cache download stream and serves it according to the specified range. -class CacheDownloadStreamResponse extends StreamResponse { - final BufferedDataStream _stream; - CacheDownloadStreamResponse._( - super.range, super.responseHeaders, this._stream); - - factory CacheDownloadStreamResponse( - final IntRange range, - final CachedResponseHeaders responseHeaders, { - required final Stream> dataStream, - required final int dataStreamPosition, - required final StreamCacheConfig streamConfig, - }) { - return CacheDownloadStreamResponse._( - range, - responseHeaders, - BufferedDataStream( - range: StreamRange(range, responseHeaders.sourceLength), - dataStream: dataStream, - dataStreamPosition: dataStreamPosition, - streamConfig: streamConfig, - ), - ); - } - - @override - void cancel() => _stream.cancel(const StreamResponseCancelledException()); - - @override - ResponseSource get source => ResponseSource.cacheDownload; - - @override - Stream> get stream => _stream; -} diff --git a/lib/src/models/stream_response/combined_cache_stream_response.dart b/lib/src/models/stream_response/combined_cache_stream_response.dart deleted file mode 100644 index 50627ce..0000000 --- a/lib/src/models/stream_response/combined_cache_stream_response.dart +++ /dev/null @@ -1,47 +0,0 @@ -import 'dart:async'; - -import '../../cache_stream/response_streams/combined_data_stream.dart'; -import '../cache_config/stream_cache_config.dart'; -import '../cache_files/cache_files.dart'; -import '../exceptions/stream_response_exceptions.dart'; -import '../metadata/cached_response_headers.dart'; -import '../stream_requests/int_range.dart'; -import 'stream_response.dart'; - -/// A stream that combines data from the partial cache file and the cache download stream. -/// It first streams data from the partial cache file, and once the file is done, it switches to the download stream. -/// Upon initalization, immediately starts buffering data from the download stream. -class CombinedCacheStreamResponse extends StreamResponse { - final CombinedDataStream _stream; - CombinedCacheStreamResponse._( - super.range, super.responseHeaders, this._stream); - - factory CombinedCacheStreamResponse.construct( - final IntRange range, - final CachedResponseHeaders responseHeaders, - final CacheFiles cacheFiles, - final Stream> dataStream, - final int dataStreamPosition, - final StreamCacheConfig streamConfig, - ) { - final combinedDataStream = CombinedDataStream( - range, - cacheFiles, - dataStream, - dataStreamPosition, - responseHeaders.sourceLength, - streamConfig, - ); - return CombinedCacheStreamResponse._( - range, responseHeaders, combinedDataStream); - } - - @override - void cancel() => _stream.cancel(const StreamResponseCancelledException()); - - @override - ResponseSource get source => ResponseSource.combined; - - @override - Stream> get stream => _stream; -} diff --git a/lib/src/models/stream_response/file_stream_response.dart b/lib/src/models/stream_response/file_stream_response.dart index 244d850..4905fc2 100644 --- a/lib/src/models/stream_response/file_stream_response.dart +++ b/lib/src/models/stream_response/file_stream_response.dart @@ -20,7 +20,7 @@ class FileStreamResponse extends StreamResponse { StreamRange(range, responseHeaders.sourceLength); //Validate range return FileStreamResponse._( CacheFileStream(streamRange, cacheFiles), - range, + streamRange.range, responseHeaders, ); } diff --git a/lib/src/models/stream_response/partial_file_stream_response.dart b/lib/src/models/stream_response/partial_file_stream_response.dart new file mode 100644 index 0000000..e256f16 --- /dev/null +++ b/lib/src/models/stream_response/partial_file_stream_response.dart @@ -0,0 +1,52 @@ +import 'dart:async'; + +import '../../cache_stream/response_streams/partial_cache_file_stream.dart'; +import '../../cache_stream/response_streams/partial_cache_feed.dart'; +import '../cache_files/cache_files.dart'; +import '../metadata/cached_response_headers.dart'; +import '../stream_requests/int_range.dart'; +import 'stream_response.dart'; +import 'stream_response_range.dart'; + +/// A response served from a cache file that is still being written. +class PartialFileStreamResponse extends StreamResponse { + final StreamRange _streamRange; + final CacheFiles _cacheFiles; + final PartialCacheFeed _feed; + + const PartialFileStreamResponse._( + super.range, + super.responseHeaders, + this._streamRange, + this._cacheFiles, + this._feed, + ); + + factory PartialFileStreamResponse( + final IntRange range, + final CacheFiles cacheFiles, + final CachedResponseHeaders responseHeaders, + final PartialCacheFeed feed, + ) { + final streamRange = StreamRange(range, responseHeaders.sourceLength); + return PartialFileStreamResponse._( + streamRange.range, + responseHeaders, + streamRange, + cacheFiles, + feed, + ); + } + + @override + Stream> get stream => + PartialCacheFileStream(_streamRange, _cacheFiles, _feed); + + @override + ResponseSource get source => ResponseSource.partialCacheFile; + + @override + void cancel() { + // Streams are created on demand and own their subscription resources. + } +} diff --git a/lib/src/models/stream_response/range_download_stream_response.dart b/lib/src/models/stream_response/range_download_stream_response.dart index 60a06ef..7e6b906 100644 --- a/lib/src/models/stream_response/range_download_stream_response.dart +++ b/lib/src/models/stream_response/range_download_stream_response.dart @@ -18,12 +18,25 @@ class RangeDownloadStreamResponse extends StreamResponse { final StreamCacheConfig config, ) async { final downloadStream = await DownloadStream.open(url, range, config); - return RangeDownloadStreamResponse._( - range, - downloadStream.responseHeaders, - downloadStream, - config.minChunkSize, - ); + + try { + final responseHeaders = downloadStream.responseHeaders; + final responseRange = IntRange.validate( + range.start, + range.end, + responseHeaders.sourceLength, + ); + return RangeDownloadStreamResponse._( + responseRange, + responseHeaders, + downloadStream, + config.minChunkSize, + ); + } catch (_) { + // If an error occurs during range validation and construction, cancel the download stream to free resources. + downloadStream.cancel(); + rethrow; + } } @override diff --git a/lib/src/models/stream_response/stream_response.dart b/lib/src/models/stream_response/stream_response.dart index 2961a6c..072b76b 100644 --- a/lib/src/models/stream_response/stream_response.dart +++ b/lib/src/models/stream_response/stream_response.dart @@ -1,16 +1,9 @@ import 'dart:async'; -import '../cache_config/stream_cache_config.dart'; -import '../cache_files/cache_files.dart'; import '../metadata/cached_response_headers.dart'; import '../stream_requests/int_range.dart'; -import 'cache_download_stream_response.dart'; -import 'combined_cache_stream_response.dart'; -import 'file_stream_response.dart'; -import 'header_stream_response.dart'; -import 'range_download_stream_response.dart'; -/// Represents a response from the cache manager. +/// Represents a response from a [HttpCacheStream]. abstract class StreamResponse { /// The byte range of the response. final IntRange range; @@ -28,65 +21,6 @@ abstract class StreamResponse { /// The total length of the source content, if known. int? get sourceLength => sourceHeaders.sourceLength; - factory StreamResponse.headersOnly( - final IntRange range, - final CachedResponseHeaders responseHeaders, - ) { - return HeaderStreamResponse(range, responseHeaders); - } - - /// Creates a [StreamResponse] from a remote download. - static Future fromDownload( - final Uri url, - final IntRange range, - final StreamCacheConfig config, - ) { - return RangeDownloadStreamResponse.construct(url, range, config); - } - - /// Creates a [StreamResponse] from a cached file. - factory StreamResponse.fromFile( - final IntRange range, - final CacheFiles cacheFiles, - final CachedResponseHeaders responseHeaders, - ) { - return FileStreamResponse(range, cacheFiles, responseHeaders); - } - - factory StreamResponse.fromStream( - final IntRange range, - final CachedResponseHeaders headers, - final Stream> dataStream, - final int dataStreamPosition, - final StreamCacheConfig streamConfig, - ) { - return CacheDownloadStreamResponse( - range, - headers, - dataStream: dataStream, - dataStreamPosition: dataStreamPosition, - streamConfig: streamConfig, - ); - } - - factory StreamResponse.combined( - final IntRange range, - final CachedResponseHeaders headers, - final CacheFiles cacheFiles, - final Stream> dataStream, - final int dataStreamPosition, - final StreamCacheConfig streamConfig, - ) { - return CombinedCacheStreamResponse.construct( - range, - headers, - cacheFiles, - dataStream, - dataStreamPosition, - streamConfig, - ); - } - ///The length of the content in the response. This may be different from the source length. int? get contentLength { final effectiveEnd = this.effectiveEnd; @@ -133,13 +67,7 @@ enum ResponseSource { ///A stream response that is served exclusively from cached data saved to a file. cacheFile, - ///A stream response that is served exclusively from the cache download stream. - /// - ///Data from the cache download stream is buffered until a listener is added. The stream must be read to completion or cancelled to release buffered data. If you no longer need the stream, you must manually call [cancel] to avoid memory leaks. - cacheDownload, - - ///A stream response that combines [cacheFile] and [cacheDownload] sources. When a listener is added, data is streamed from the cache file first, and once the file stream is done, it switches to the cache download stream. - /// - ///Data from the cache download stream is buffered until a listener is added. The stream must be read to completion or cancelled to release buffered data. If you no longer need the stream, you must manually call [cancel] to avoid memory leaks. - combined, + /// A stream response served from committed bytes in a cache file that is + /// still being written. It waits for requested positions as needed. + partialCacheFile, } diff --git a/lib/src/models/stream_response/stream_response_range.dart b/lib/src/models/stream_response/stream_response_range.dart index ea03f63..c74a0cf 100644 --- a/lib/src/models/stream_response/stream_response_range.dart +++ b/lib/src/models/stream_response/stream_response_range.dart @@ -8,11 +8,9 @@ class StreamRange { const StreamRange._(this.range, this.sourceLength); factory StreamRange(IntRange range, int? sourceLength) { - if (sourceLength != null && range.upperBound > sourceLength) { - throw RangeError.range(range.upperBound, 0, sourceLength, 'range end'); - } - - return StreamRange._(range, sourceLength); + final validatedRange = + IntRange.validate(range.start, range.end, sourceLength); + return StreamRange._(validatedRange, sourceLength); } static StreamRange validate(int? start, int? end, int? sourceLength) { diff --git a/pubspec.yaml b/pubspec.yaml index 48f3478..4d4a1f5 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -1,6 +1,6 @@ name: http_cache_stream description: "Simultaneously download, cache, and stream remote content. Perfect for media players and any plugin that streams web content." -version: 0.1.0 +version: 0.2.0 homepage: https://github.com/Colton127/http_cache_stream repository: https://github.com/Colton127/http_cache_stream topics: diff --git a/test/e2e/dispose_test.dart b/test/e2e/dispose_test.dart index b1dff8c..04c55b6 100644 --- a/test/e2e/dispose_test.dart +++ b/test/e2e/dispose_test.dart @@ -1,4 +1,7 @@ +import 'dart:async'; + import 'package:flutter_test/flutter_test.dart'; +import 'package:http_cache_stream/http_cache_stream.dart'; import '../support/harness.dart'; @@ -10,7 +13,12 @@ void main() { // than completing in milliseconds. setUp(() async { h = CacheTestHarness(); - await h.setUp(); + await h.setUp( + configBuilder: (cacheDir) => GlobalCacheConfig( + cacheDirectory: cacheDir, + savePartialCache: false, + ), + ); }); tearDown(() => h.tearDown()); @@ -44,4 +52,38 @@ void main() { await stream.dispose(force: true); expect(stream.isDisposed, isTrue); }); + + test('dispose deletes an incomplete cache with unknown source length', + () async { + final responseCloseGate = Completer(); + addTearDown(() { + if (!responseCloseGate.isCompleted) responseCloseGate.complete(); + }); + h.origin.chunkedTransferEncoding = true; + h.origin.responseCloseGate = responseCloseGate; + + final stream = h.manager.createStream(h.origin.url('/chunked.mp3')); + stream.download().ignore(); + + // Wait until the chunked body is committed while the origin deliberately + // withholds the final chunk, keeping the download incomplete with no known + // source length. + for (var i = 0; + i < 100 && + (!stream.files.partial.existsSync() || + stream.files.partial.lengthSync() == 0); + i++) { + await Future.delayed(const Duration(milliseconds: 20)); + } + + expect(stream.headers?.sourceLength, isNull); + expect(stream.files.partial.existsSync(), isTrue); + expect(stream.files.partial.lengthSync(), greaterThan(0)); + + await stream.dispose(force: true); + responseCloseGate.complete(); + + expect(stream.files.partial.existsSync(), isFalse); + expect(stream.files.metadata.existsSync(), isFalse); + }); } diff --git a/test/e2e/e2e_headers_test.dart b/test/e2e/e2e_headers_test.dart index eb6a6ec..ab7a51d 100644 --- a/test/e2e/e2e_headers_test.dart +++ b/test/e2e/e2e_headers_test.dart @@ -53,6 +53,52 @@ void main() { await stream.dispose(); }); + test('a satisfiable range with an oversized end is clamped', () async { + final source = h.origin.url('/media/clip.mp3'); + final stream = h.manager.createStream(source); + await stream.download(); + final total = h.origin.payload.length; + final start = total - 100; + + final res = await h.fetch( + h.manager.getCacheUrl(source), + range: 'bytes=$start-${total + 1000}', + ); + expect(res.statusCode, 206); + expect(res.header('content-range'), 'bytes $start-${total - 1}/$total'); + expect(res.body.length, 100); + + await stream.dispose(); + }); + + test('an uncached satisfiable range with an oversized end is clamped', + () async { + final total = h.origin.payload.length; + final start = total - 100; + + final res = await h.fetch( + h.manager.getCacheUrl(h.origin.url('/media/uncached-clip.mp3')), + range: 'bytes=$start-${total + 1000}', + ); + expect(res.statusCode, 206); + expect(res.header('content-range'), 'bytes $start-${total - 1}/$total'); + expect(res.body.length, 100); + }); + + test('a split origin range with an oversized end is clamped', () async { + h.manager.config.rangeRequestSplitThreshold = 1; + final total = h.origin.payload.length; + final start = total - 100; + + final res = await h.fetch( + h.manager.getCacheUrl(h.origin.url('/media/split-clip.mp3')), + range: 'bytes=$start-${total + 1000}', + ); + expect(res.statusCode, 206); + expect(res.header('content-range'), 'bytes $start-${total - 1}/$total'); + expect(res.body.length, 100); + }); + test('HEAD returns headers with an empty body', () async { final cacheUrl = h.manager.getCacheUrl(h.origin.url('/media/clip.mp3')); final res = await h.fetch(cacheUrl, method: 'HEAD'); diff --git a/test/e2e/lifecycle_test.dart b/test/e2e/lifecycle_test.dart index 73b0801..4c856a7 100644 --- a/test/e2e/lifecycle_test.dart +++ b/test/e2e/lifecycle_test.dart @@ -1,3 +1,6 @@ +import 'dart:async'; +import 'dart:typed_data'; + import 'package:flutter_test/flutter_test.dart'; import 'package:http_cache_stream/http_cache_stream.dart'; @@ -11,6 +14,24 @@ const _fastLifecycle = StreamLifecycleConfig( disposeAfter: Duration(milliseconds: 150), ); +/// Starts [stream]'s download, waits until the origin has delivered its gated +/// first half, then disposes it while preserving the partial cache. +Future _abortAtHalf(HttpCacheStream stream) async { + final reachedHalf = Completer(); + final subscription = stream.cacheStateStream.listen((state) { + final progress = state.progress; + if (progress != null && progress >= 0.5 && !reachedHalf.isCompleted) { + reachedHalf.complete(progress); + } + }); + + stream.download().ignore(); + final progress = await reachedHalf.future.timeout(const Duration(seconds: 5)); + await stream.dispose(); + await subscription.cancel(); + return progress; +} + void main() { late CacheTestHarness h; @@ -84,6 +105,141 @@ void main() { expect(h.manager.getCacheUrl(source), h.manager.getCacheUrl(source)); }); + test('resumes a partial download after the stream is recreated', () async { + final source = h.origin.url('/resume.mp3'); + final half = h.origin.payload.length ~/ 2; + final bodyGate = Completer(); + h.origin + ..responseBodyGate = bodyGate + ..responseBodyGateAfterBytes = half; + + final interrupted = h.manager.createStream(source); + final interruptedProgress = await _abortAtHalf(interrupted); + expect(interruptedProgress, 0.5); + + // The first request is now abandoned; do not gate the resumed range. + h.origin + ..responseBodyGate = null + ..responseBodyGateAfterBytes = null; + bodyGate.complete(); + + final resumed = h.manager.createStream(source); + await resumed.validateCache(); // Wait for persisted metadata and state. + expect(resumed.progress, interruptedProgress); + expect(resumed.cachePosition, half); + + final file = await resumed.download(); + expect(Payload.hash(await file.readAsBytes()), h.payloadHash); + expect(h.origin.rangeHeaders, contains('bytes=$half-')); + + await resumed.dispose(); + }); + + test('a full request does not serve stale partial bytes after recreation', + () async { + final source = h.origin.url('/request-resume.mp3'); + final half = h.origin.payload.length ~/ 2; + final bodyGate = Completer(); + h.origin + ..responseBodyGate = bodyGate + ..responseBodyGateAfterBytes = half; + + final interrupted = h.manager.createStream(source); + await _abortAtHalf(interrupted); + + h.origin + ..responseBodyGate = null + ..responseBodyGateAfterBytes = null + ..payload = Payload.generate(h.origin.payload.length, seed: 0xBEEF) + ..etag = '"v2"'; + final changedPayloadHash = h.payloadHash; + final responseStartGate = Completer(); + h.origin.responseStartGate = responseStartGate; + bodyGate.complete(); + final resumed = h.manager.createStream(source); + + // Start the resumed download using a request, then wait until its HTTP + // request is blocked before headers arrive. The next request exercises the + // window where CacheDownloader has stale persisted headers and partial + // bytes available. + resumed.request(start: 0, end: h.origin.payload.length).ignore(); + for (var i = 0; i < 50 && h.origin.requestCount < 2; i++) { + await Future.delayed(const Duration(milliseconds: 10)); + } + expect(h.origin.requestCount, greaterThanOrEqualTo(2)); + + // A correct implementation queues the request below until after + // validation, so it simply waits here before receiving the new file. + Future.delayed(const Duration(milliseconds: 100), () { + if (!responseStartGate.isCompleted) responseStartGate.complete(); + }); + + // Do not validate or explicitly resume the download. This request must + // reject the stale partial cache and serve only bytes from the changed + // source. + final response = await resumed.request( + start: 0, + end: h.origin.payload.length, + ); + final bytes = BytesBuilder(copy: false); + await response.stream + .timeout(const Duration(seconds: 5)) + .forEach(bytes.add); + + expect(Payload.hash(bytes.takeBytes()), changedPayloadHash); + expect(h.origin.rangeHeaders, contains('bytes=$half-')); + + await resumed.dispose(); + }); + + test('resets a partial cache when the source changes before resuming', + () async { + final source = h.origin.url('/changed-resume.mp3'); + final half = h.origin.payload.length ~/ 2; + final bodyGate = Completer(); + h.origin + ..responseBodyGate = bodyGate + ..responseBodyGateAfterBytes = half; + + final interrupted = h.manager.createStream(source); + final cacheFilePath = interrupted.cacheFile.path; + final originalPayloadHash = h.payloadHash; + await _abortAtHalf(interrupted); + + h.origin + ..responseBodyGate = null + ..responseBodyGateAfterBytes = null + ..payload = Payload.generate(h.origin.payload.length, seed: 0xCAFE) + ..etag = '"v2"'; + final changedPayloadHash = h.payloadHash; + expect(changedPayloadHash, isNot(originalPayloadHash)); + bodyGate.complete(); + + final resumed = h.manager.createStream(source); + await resumed.validateCache(); // Wait for persisted metadata and state. + expect(resumed.cacheFile.path, cacheFilePath); + expect(resumed.cachePosition, half); + + final cacheErrors = []; + final subscription = resumed.cacheStateStream.listen( + (_) {}, + onError: cacheErrors.add, + ); + final rangesBeforeResume = h.origin.rangeHeaders.length; + final file = await resumed.download(); + await subscription.cancel(); + final resumedRanges = h.origin.rangeHeaders.sublist(rangesBeforeResume); + + expect(cacheErrors, contains(isA())); + expect(resumedRanges, contains('bytes=$half-'), + reason: 'the stale partial cache must first be detected on resume'); + expect(resumedRanges, contains(null), + reason: 'the invalid partial cache must be reset before a full retry'); + expect(Payload.hash(await file.readAsBytes()), changedPayloadHash); + + await resumed.dispose(); + }); + test('deleteCache removes cached files once no streams are active', () async { final source = h.origin.url('/f.mp3'); final files = h.manager.getCacheFiles(source); diff --git a/test/io/buffered_io_sink_test.dart b/test/io/buffered_io_sink_test.dart index 915dc8f..83e5a54 100644 --- a/test/io/buffered_io_sink_test.dart +++ b/test/io/buffered_io_sink_test.dart @@ -1,9 +1,10 @@ -import 'dart:async'; import 'dart:io'; import 'dart:typed_data'; -import 'package:http_cache_stream/src/cache_stream/cache_downloader/buffered_io_sink.dart'; import 'package:flutter_test/flutter_test.dart'; +import 'package:http_cache_stream/src/cache_stream/cache_downloader/buffered_io_sink.dart'; +import 'package:http_cache_stream/src/cache_stream/response_streams/partial_cache_feed.dart'; +import 'package:http_cache_stream/src/models/exceptions/partial_cache_feed_exceptions.dart'; import '../support/payload.dart'; @@ -49,7 +50,9 @@ void main() { await sink.flush(); expect(sink.bufferSize, 0); expect(sink.flushedBytes, data.length); + expect(sink.feed.position, data.length); await sink.close(); + expect(sink.feed.isClosed, isTrue); }); test('append mode resumes from an existing partial file', () async { @@ -74,30 +77,106 @@ void main() { sink.add(data); final f = sink.waitForPosition(5 * 1024); await sink.flush(); - await f; // should not throw + await f.future; // should not throw await sink.close(); }); - test('waitForPosition times out when the target is never reached', () async { - final sink = BufferedIOSink(tmp('timeout.bin'), 0); - sink.add(Payload.generate(1024)); + test('feed waitForPosition completes once the target is flushed', () async { + final data = Payload.generate(10 * 1024); + final sink = BufferedIOSink(tmp('feed-wait.bin'), 0); + final feed = sink.feed; + sink.add(data); + + final wait = feed.waitForPosition(5 * 1024); await sink.flush(); - await expectLater( - sink.waitForPosition(1 << 30, const Duration(milliseconds: 100)), - throwsA(isA()), - ); + await wait.future; // should not throw + + expect(feed.position, data.length); await sink.close(); + expect(feed.isClosed, isTrue); }); - test('waitForPosition fails if the sink closes before reaching it', () async { + test('cancel fails the waiter and releases it from the feed', () async { + final sink = BufferedIOSink(tmp('cancel.bin'), 0); + final waiter = sink.waitForPosition(10 * 1024); + expect(waiter.isCompleted, isFalse); + + // Attach the matcher before cancelling: an unobserved error future would + // otherwise crash the test. + final expectation = expectLater( + waiter.future, throwsA(isA())); + waiter.cancel(); + expect(waiter.isCompleted, isTrue); + await expectation; + + // The waiter is no longer tracked, so reaching its position does nothing. + sink.add(Payload.generate(10 * 1024)); + await sink.flush(); + expect(sink.feed.position, 10 * 1024); + expect(sink.feed.failure, isNull); + + await sink.close(isDone: true); + }); + + test('cancel is a no-op once the waiter has been satisfied', () async { + final sink = BufferedIOSink(tmp('cancel-late.bin'), 0); + sink.add(Payload.generate(4 * 1024)); + + final waiter = sink.waitForPosition(2 * 1024); + await sink.flush(); + await waiter.future; + + waiter.cancel(); // Must not turn a satisfied wait into a failure + expect(waiter.isCompleted, isTrue); + await waiter.future; // Still completes normally + + await sink.close(isDone: true); + }); + + test('cancel is a no-op for a position that was already reached', () async { + final sink = BufferedIOSink(tmp('cancel-reached.bin'), 0); + sink.add(Payload.generate(4 * 1024)); + await sink.flush(); + + final waiter = sink.waitForPosition(1024); + expect(waiter.isCompleted, isTrue); + waiter.cancel(); + await waiter.future; // Completed waiters ignore cancel + + await sink.close(isDone: true); + }); + + test('waitForPosition fails as aborted if the sink closes before reaching it', + () async { final sink = BufferedIOSink(tmp('closed.bin'), 0); sink.add(Payload.generate(1024)); // Attach the matcher before closing: close() fails the waiter synchronously, // and an unobserved error future would otherwise crash the test. final expectation = expectLater( - sink.waitForPosition(10 * 1024 * 1024), throwsA(isA())); + sink.waitForPosition(10 * 1024 * 1024).future, + throwsA(isA())); await sink.close(); await expectation; + + // The feed carries the failure, so a reader with no known end position can + // tell the truncated content apart from an end of stream. + expect(sink.feed.failure, isA()); + }); + + test( + 'waitForPosition fails as closed when the sink is done before reaching it', + () async { + final sink = BufferedIOSink(tmp('closed-done.bin'), 0); + sink.add(Payload.generate(1024)); + final expectation = expectLater( + sink.waitForPosition(10 * 1024 * 1024).future, + throwsA(isA())); + await sink.close(isDone: true); + await expectation; + + // Reaching the end of the content is not a failure: flushedBytes is the + // true end, so readers must treat it as an end of stream. + expect(sink.feed.failure, isNull); }); test('adding to a closed sink throws', () async { diff --git a/test/io/partial_cache_file_stream_test.dart b/test/io/partial_cache_file_stream_test.dart new file mode 100644 index 0000000..6240fa7 --- /dev/null +++ b/test/io/partial_cache_file_stream_test.dart @@ -0,0 +1,249 @@ +import 'dart:async'; +import 'dart:io'; +import 'dart:typed_data'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:http/http.dart' as http; +import 'package:http_cache_stream/http_cache_stream.dart'; +import 'package:http_cache_stream/src/cache_stream/cache_downloader/buffered_io_sink.dart'; +import 'package:http_cache_stream/src/cache_stream/response_streams/partial_cache_file_stream.dart'; +import 'package:http_cache_stream/src/cache_stream/response_streams/partial_cache_feed.dart'; +import 'package:http_cache_stream/src/models/stream_response/partial_file_stream_response.dart'; +import 'package:http_cache_stream/src/models/stream_response/stream_response_range.dart'; + +import '../support/payload.dart'; + +void main() { + late Directory directory; + late CacheFiles cacheFiles; + + setUp(() async { + directory = await Directory.systemTemp.createTemp('hcs_partial_stream_'); + cacheFiles = CacheFiles.fromFile(File('${directory.path}/cache.bin')); + }); + + tearDown(() async { + if (directory.existsSync()) await directory.delete(recursive: true); + }); + + test('waits for future positions and enforces the requested range', () async { + final payload = Payload.generate(200 * 1024); + final sink = BufferedIOSink(cacheFiles.partial, 0); + sink.add(Uint8List.sublistView(payload, 0, 50 * 1024)); + await sink.flush(); + + final stream = PartialCacheFileStream( + StreamRange.validate(10 * 1024, 150 * 1024, payload.length), + cacheFiles, + sink.feed, + ); + final resultFuture = stream.expand((bytes) => bytes).toList(); + + await Future.delayed(Duration.zero); + sink.add(Uint8List.sublistView(payload, 50 * 1024)); + await sink.flush(); + + final result = await resultFuture; + expect( + Payload.hash(result), + Payload.hash(payload.sublist(10 * 1024, 150 * 1024)), + ); + await sink.close(); + }); + + test('supports independent repeated listeners', () async { + final payload = Payload.generate(128 * 1024); + final sink = BufferedIOSink(cacheFiles.partial, 0); + sink.add(payload); + await sink.close(); + + final stream = PartialCacheFileStream( + StreamRange.validate(0, payload.length, payload.length), + cacheFiles, + sink.feed, + ); + + final results = await Future.wait([ + stream.expand((bytes) => bytes).toList(), + stream.expand((bytes) => bytes).toList(), + ]); + expect(Payload.hash(results[0]), Payload.hash(payload)); + expect(Payload.hash(results[1]), Payload.hash(payload)); + }); + + test('an open-ended stream completes at the final feed position', () async { + final payload = Payload.generate(80 * 1024); + final sink = BufferedIOSink(cacheFiles.partial, 0); + final stream = PartialCacheFileStream( + StreamRange.validate(4 * 1024, null, null), + cacheFiles, + sink.feed, + ); + final resultFuture = stream.expand((bytes) => bytes).toList(); + + sink.add(payload); + await sink.close(isDone: true); + final result = await resultFuture; + + expect(Payload.hash(result), Payload.hash(payload.sublist(4 * 1024))); + }); + + test('a caught-up open-ended stream completes on a clean close', () async { + final payload = Payload.generate(4 * 1024); + final sink = BufferedIOSink(cacheFiles.partial, 0); + sink.add(payload); + await sink.flush(); + + final stream = PartialCacheFileStream( + StreamRange.validate(0, null, null), + cacheFiles, + sink.feed, + ); + final firstData = Completer(); + final done = Completer(); + final result = []; + Object? streamError; + + stream.listen( + (data) { + result.addAll(data); + if (!firstData.isCompleted) firstData.complete(); + }, + onError: (Object error) => streamError = error, + onDone: done.complete, + ); + + // Wait until the reader consumes every committed byte and starts waiting + // for the feed to advance. A clean producer close must wake that pending + // read as EOF, rather than surface the waiter's closed-position error. + await firstData.future; + await sink.close(isDone: true); + await done.future; + + expect(Payload.hash(result), Payload.hash(payload)); + expect(streamError, isNull); + }); + + test('a bounded stream errors when a clean feed closes before its end', + () async { + final payload = Payload.generate(4 * 1024); + final sink = BufferedIOSink(cacheFiles.partial, 0); + sink.add(payload); + await sink.flush(); + + final stream = PartialCacheFileStream( + StreamRange.validate(0, 8 * 1024, 8 * 1024), + cacheFiles, + sink.feed, + ); + final expectation = expectLater( + stream.expand((bytes) => bytes).toList(), + throwsA(isA()), + ); + + await sink.close(isDone: true); + await expectation; + }); + + test('an open-ended stream errors when the download is aborted', () async { + final payload = Payload.generate(80 * 1024); + final sink = BufferedIOSink(cacheFiles.partial, 0); + final stream = PartialCacheFileStream( + StreamRange.validate(4 * 1024, null, null), + cacheFiles, + sink.feed, + ); + final resultFuture = stream.expand((bytes) => bytes).toList(); + + sink.add(payload); + await sink + .close(); //Aborted: the source never reached the end of its content + + // Without a known end position, a clean close is the only end-of-stream + // signal. An aborted feed must not be reported as one, or the listener + // accepts the truncated content as complete. + await expectLater( + resultFuture, throwsA(isA())); + }); + + test('opens the completed file after partial-file promotion', () async { + final payload = Payload.generate(64 * 1024); + final sink = BufferedIOSink(cacheFiles.partial, 0); + sink.add(payload); + await sink.close(); + await cacheFiles.partial.rename(cacheFiles.complete.path); + + final stream = PartialCacheFileStream( + StreamRange.validate(0, payload.length, payload.length), + cacheFiles, + sink.feed, + ); + final result = await stream.expand((bytes) => bytes).toList(); + + expect(Payload.hash(result), Payload.hash(payload)); + }); + + test('streams from a completed feed with a fixed final position', () async { + final payload = Payload.generate(64 * 1024); + await cacheFiles.complete.writeAsBytes(payload); + + const finalPosition = 48 * 1024; + const PartialCacheFeed feed = PartialCacheFeed.completed(finalPosition); + final stream = PartialCacheFileStream( + StreamRange.validate(4 * 1024, null, null), + cacheFiles, + feed, + ); + final result = await stream.expand((bytes) => bytes).toList(); + + expect(Payload.hash(result), + Payload.hash(payload.sublist(4 * 1024, finalPosition))); + expect(feed.position, finalPosition); + expect(feed.isClosed, isTrue); + expect(feed.failure, isNull); + }); + + test('completed feed rejects positions beyond its final position', () async { + const feed = CompletedPartialCacheFeed(1024); + + expect(feed.waitForPosition(1024).isCompleted, isTrue); + await expectLater(feed.waitForPosition(1024).future, completes); + await expectLater( + feed.waitForPosition(1025).future, + throwsA(isA()), + ); + }); + + test('fromPartialFile creates lazy streams on demand', () async { + final payload = Payload.generate(96 * 1024); + final sink = BufferedIOSink(cacheFiles.partial, 0); + final headers = CachedResponseHeaders.fromBaseResponse( + http.Response( + '', + HttpStatus.ok, + headers: {HttpHeaders.contentLengthHeader: '${payload.length}'}, + ), + ); + final response = PartialFileStreamResponse( + const IntRange(8 * 1024, 80 * 1024), + cacheFiles, + headers, + sink.feed, + ); + + final firstStream = response.stream; + final secondStream = response.stream; + expect(identical(firstStream, secondStream), isFalse); + expect(response.source, ResponseSource.partialCacheFile); + response.cancel(); + + sink.add(payload); + await sink.flush(); + final result = await firstStream.expand((bytes) => bytes).toList(); + expect( + Payload.hash(result), + Payload.hash(payload.sublist(8 * 1024, 80 * 1024)), + ); + await sink.close(); + }); +} diff --git a/test/support/test_origin.dart b/test/support/test_origin.dart index 00e1797..5203270 100644 --- a/test/support/test_origin.dart +++ b/test/support/test_origin.dart @@ -1,3 +1,4 @@ +import 'dart:async'; import 'dart:io'; import 'dart:typed_data'; @@ -13,7 +14,9 @@ class TestOrigin { final HttpServer _server; /// The bytes this origin serves for a `200`/`206` response. - final Uint8List payload; + /// + /// Mutable so lifecycle tests can simulate changed content at the same URL. + Uint8List payload; // ---- Behavior knobs (mutate per-test) ---- @@ -46,6 +49,29 @@ class TestOrigin { /// then resets to null so a retry can succeed. int? dropAfterBytes; + /// When true, omits Content-Length and sends the response with chunked + /// transfer encoding, leaving the source length unknown until completion. + bool chunkedTransferEncoding = false; + + /// When set, the response flushes its body and waits for this gate before + /// closing. This allows tests to dispose a download before a chunked response + /// sends its clean end-of-stream signal. + Completer? responseCloseGate; + + /// When set, waits before sending response headers. This lets tests control + /// whether a cache request can read persisted partial bytes before the latest + /// origin headers have been received. + Completer? responseStartGate; + + /// When set with [responseBodyGateAfterBytes], the response sends that many + /// body bytes, flushes them, and waits before sending the remainder. This + /// makes it possible to stop a download at a deterministic partial position. + Completer? responseBodyGate; + + /// Number of bytes to send before waiting on [responseBodyGate]. Ignored when + /// it is outside the requested body's bounds. + int? responseBodyGateAfterBytes; + // ---- Observability ---- int requestCount = 0; @@ -53,11 +79,7 @@ class TestOrigin { String? lastRangeHeader; final List rangeHeaders = []; - // Reported as `localhost` (not the bound `127.0.0.1`) so the source host - // differs from the cache server's loopback host; otherwise the package treats - // the source URL as an already-encoded cache URL. `localhost` still resolves - // to loopback where the origin is listening. - Uri get baseUri => Uri(scheme: 'http', host: 'localhost', port: _server.port); + Uri get baseUri => Uri(scheme: 'http', host: '127.0.0.1', port: _server.port); /// A source URL on this origin for the given [path] (e.g. `/media/file.mp3`). Uri url(String path) => baseUri.replace(path: path); @@ -82,6 +104,10 @@ class TestOrigin { lastRangeHeader = rangeHeader; rangeHeaders.add(rangeHeader); + if (responseStartGate case final gate?) { + await gate.future; + } + final response = request.response; if (forcedStatusCode != null) { @@ -93,7 +119,10 @@ class TestOrigin { _setCommonHeaders(response); // Resolve the requested byte range. - final total = payload.length; + // Keep one request internally consistent if a test swaps [payload] while a + // previous response is paused at [responseBodyGate]. + final responsePayload = payload; + final total = responsePayload.length; int start = 0; int endEx = total; // exclusive final isRange = supportRanges && rangeHeader != null; @@ -118,14 +147,18 @@ class TestOrigin { } final bodyLength = endEx - start; - response.headers.contentLength = lyingContentLength ?? bodyLength; + if (chunkedTransferEncoding) { + response.headers.chunkedTransferEncoding = true; + } else { + response.headers.contentLength = lyingContentLength ?? bodyLength; + } if (request.method == 'HEAD') { await response.close(); return; } - final body = Uint8List.sublistView(payload, start, endEx); + final body = Uint8List.sublistView(responsePayload, start, endEx); final drop = dropAfterBytes; if (drop != null && drop < body.length) { @@ -137,7 +170,23 @@ class TestOrigin { return; } - response.add(body); + final bodyGate = responseBodyGate; + final bodyGateAfterBytes = responseBodyGateAfterBytes; + if (bodyGate != null && + bodyGateAfterBytes != null && + bodyGateAfterBytes > 0 && + bodyGateAfterBytes < body.length) { + response.add(Uint8List.sublistView(body, 0, bodyGateAfterBytes)); + await response.flush(); + await bodyGate.future; + response.add(Uint8List.sublistView(body, bodyGateAfterBytes)); + } else { + response.add(body); + } + if (responseCloseGate case final gate?) { + await response.flush(); + await gate.future; + } await response.close(); } diff --git a/test/unit/http_range_test.dart b/test/unit/http_range_test.dart index 1e96ef5..6321608 100644 --- a/test/unit/http_range_test.dart +++ b/test/unit/http_range_test.dart @@ -94,5 +94,15 @@ void main() { isFalse, ); }); + + test('equal when the response clamps the requested end to the source', () { + expect( + HttpRange.isEqual( + HttpRangeRequest(100, 999), + HttpRangeResponse(100, 499, sourceLength: 500), + ), + isTrue, + ); + }); }); } diff --git a/test/unit/int_range_test.dart b/test/unit/int_range_test.dart index a412da7..1d05794 100644 --- a/test/unit/int_range_test.dart +++ b/test/unit/int_range_test.dart @@ -33,8 +33,10 @@ void main() { expect(() => IntRange.validate(10, null, 5), throwsRangeError); }); - test('end beyond max throws', () { - expect(() => IntRange.validate(0, 600, 500), throwsRangeError); + test('end beyond max is clamped when the start is satisfiable', () { + final r = IntRange.validate(100, 600, 500); + expect(r.start, 100); + expect(r.end, 500); }); }); diff --git a/test/unit/url_codec_test.dart b/test/unit/url_codec_test.dart index d178705..8971719 100644 --- a/test/unit/url_codec_test.dart +++ b/test/unit/url_codec_test.dart @@ -54,6 +54,42 @@ void main() { expect(server.encodeSourceUrl(encoded), encoded); }); + test('same-host source URL on another port is encoded normally', () { + final source = Uri( + scheme: 'http', + host: server.serverUri.host, + port: server.serverUri.port + 1, + path: '/file.mp3', + ); + + expectRoundTrip(source); + }); + + test('same-host https source URL is encoded normally', () { + final source = Uri( + scheme: 'https', + host: server.serverUri.host, + port: server.serverUri.port, + path: '/file.mp3', + ); + + expectRoundTrip(source); + }); + + test('an encoded URL with a stale cache-server port is re-encoded', () { + final source = Uri.parse('https://example.com/file.mp3'); + final staleEncoded = server + .encodeSourceUrl(source) + .replace(port: server.serverUri.port + 1); + + final reEncoded = server.encodeSourceUrl(staleEncoded); + + expect(reEncoded.scheme, server.serverUri.scheme); + expect(reEncoded.host, server.serverUri.host); + expect(reEncoded.port, server.serverUri.port); + expect(server.decodeSourceUrl(reEncoded), source); + }); + test('a foreign URL does not validate as a cache URL', () { expect( server