Upgrade `package:tar` to 0.5.6 (#3540)
diff --git a/lib/src/third_party/tar/README.md b/lib/src/third_party/tar/README.md
index a2fbf7b..5e12e5a 100644
--- a/lib/src/third_party/tar/README.md
+++ b/lib/src/third_party/tar/README.md
@@ -4,4 +4,4 @@
tar-archives.
* Repository: `https://github.com/simolus3/tar/`
- * Revision: `901ae404e0a225d9b08e5253415ca092f5c08706`
+ * Revision: `23ee71d667f003fba8c80ee126d5e1330d17c141`
diff --git a/lib/src/third_party/tar/src/reader.dart b/lib/src/third_party/tar/src/reader.dart
index b9bc3d3..8502a26 100644
--- a/lib/src/third_party/tar/src/reader.dart
+++ b/lib/src/third_party/tar/src/reader.dart
@@ -3,7 +3,6 @@
import 'dart:convert';
import 'dart:typed_data';
-import 'package:async/async.dart';
import 'package:meta/meta.dart';
import 'package:typed_data/typed_data.dart';
@@ -27,21 +26,7 @@
final int _maxSpecialFileSize;
TarEntry? _current;
-
- /// The underlying content stream for the [_current] entry. Draining this
- /// stream will move the tar reader to the beginning of the next file.
- ///
- /// This is not the same as `_current.stream` for sparse files, which are
- /// reported as expanded through [TarEntry.contents].
- /// For that reason, we prefer to drain this stream when skipping a tar entry.
- /// When we know we're skipping data, there's no point expanding sparse holes.
- ///
- /// This stream is always set to null after being drained, and there can only
- /// be one [_underlyingContentStream] at a time.
- Stream<List<int>>? _underlyingContentStream;
-
- /// Whether [_current] has ever been listened to.
- bool _listenedToContentsOnce = false;
+ _CurrentEntryStream? _currentStream;
/// Whether we're in the process of reading tar headers.
bool _isReadingHeaders = false;
@@ -220,7 +205,9 @@
nextHeader.format = format;
_current = TarEntry(nextHeader, content);
- _listenedToContentsOnce = false;
+ final currentStreams = _currentStream;
+ assert(currentStreams == null ||
+ currentStreams.state == _EntryStreamState.preListen);
_isReadingHeaders = false;
return true;
}
@@ -233,8 +220,7 @@
_isDone = true;
_current = null;
- _underlyingContentStream = null;
- _listenedToContentsOnce = false;
+ _currentStream = null;
_isReadingHeaders = false;
// Note: Calling cancel is safe when the stream has already been completed.
@@ -276,18 +262,51 @@
}
_isReadingHeaders = true;
- final underlyingStream = _underlyingContentStream;
+ final underlyingStream = _currentStream;
if (underlyingStream != null) {
- if (_listenedToContentsOnce) {
- throw StateError(
+ switch (underlyingStream.state) {
+ case _EntryStreamState.preListen:
+ await underlyingStream.drain<void>();
+ // The stream should reset when drained (we do this in _publishStream)
+ assert(_currentStream == null);
+
+ break;
+ case _EntryStreamState.subscriptionActive:
+ throw StateError(
'Illegal call to TarReader.moveNext() while a previous stream was '
'active.\n'
'When listening to tar contents, make sure the stream is '
- 'complete or cancelled before calling TarReader.moveNext() again.');
- } else {
- await underlyingStream.drain<void>();
- // The stream should reset when drained (we do this in _publishStream)
- assert(_underlyingContentStream == null);
+ 'complete or cancelled before calling TarReader.moveNext() again.',
+ );
+ case _EntryStreamState.cancelled:
+ // ignore: cancel_subscriptions
+ final subscription = underlyingStream._sourceSubscription!;
+
+ // Re-purpose the existing subscription to drain the stream
+ assert(subscription.isPaused);
+
+ subscription
+ ..onData(null)
+ ..resume();
+
+ try {
+ await subscription.asFuture<void>();
+ } on Object {
+ await cancel();
+ rethrow;
+ } finally {
+ // This also draines the stream
+ _currentStream = null;
+ }
+
+ break;
+ case _EntryStreamState.done:
+ assert(
+ false,
+ 'Unreachable: There should not be a currentStream in a done state, '
+ 'as the stream is no longer current at that point.',
+ );
+ break;
}
}
}
@@ -418,59 +437,14 @@
/// Publishes an library-internal stream for users.
///
/// This adds a check to ensure that the stream we're exposing has the
- /// expected length. It also sets the [_underlyingContentStream] field when
- /// the stream starts and resets it when it's done.
- Stream<List<int>> _publishStream(Stream<List<int>> stream, int length) {
+ /// expected length. It also sets the [_currentStream] field and resets it
+ /// when it's done.
+ Stream<List<int>> _publishStream(Stream<Uint8List> stream, int length) {
// There can only be one content stream at a time. This precondition is
// checked by _prepareToReadHeaders.
- assert(_underlyingContentStream == null);
- Stream<List<int>>? thisStream;
+ assert(_currentStream == null);
- return thisStream =
- _underlyingContentStream = Stream.eventTransformed(stream, (sink) {
- // This callback is called when we have a listener. Make sure that, at
- // this point, this stream is still the active content stream.
- // If users store the contents of a tar header, then read more tar
- // entries, and finally try to read the stream of the old contents, they'd
- // get an exception about the straem already being listened to.
- // This can be a bit confusing, so this check enables a better error UX.
- if (thisStream != _underlyingContentStream) {
- throw StateError(
- 'Tried listening to an outdated tar entry. \n'
- 'As all tar entries found by a reader are backed by a single source '
- 'stream, only the latest tar entry can be read. It looks like you '
- 'stored the results of `tarEntry.contents` somewhere, called '
- '`reader.moveNext()` and then read the contents of the previous '
- 'entry.\n'
- 'For more details, including a discussion of workarounds, see '
- 'https://github.com/simolus3/tar/issues/18',
- );
- } else if (_listenedToContentsOnce) {
- throw StateError(
- 'A tar entry has been listened to multiple times. \n'
- 'As all tar entries are read from what\'s likely a single-'
- 'subscription stream, this is unsupported. If you didn\'t read a tar '
- 'entry multiple times yourself, perhaps you\'ve called `moveNext()` '
- 'before reading contents?',
- );
- }
-
- _listenedToContentsOnce = true;
-
- late _OutgoingStreamGuard guard;
- return guard = _OutgoingStreamGuard(
- length,
- sink,
- // Reset state when the stream is done. This will only be called when
- // the stream is done, not when a listener cancels.
- () {
- _underlyingContentStream = null;
- if (guard.hadError) {
- cancel();
- }
- },
- );
- });
+ return _currentStream = _CurrentEntryStream(this, stream, length);
}
/// Checks the PAX headers for GNU sparse headers.
@@ -881,23 +855,96 @@
}
}
-/// Event-sink tracking the length of emitted tar entry streams.
+enum _EntryStreamState {
+ preListen,
+ subscriptionActive,
+ cancelled,
+ done,
+}
+
+/// The underlying content stream for the [TarReader._current] entry. Draining
+/// this stream will move the tar reader to the beginning of the next file.
///
-/// [ChunkedStreamReader.readStream] might return a stream shorter than
-/// expected. That indicates an invalid tar file though, since the correct size
-/// is stored in the header.
-class _OutgoingStreamGuard extends EventSink<Uint8List> {
- int remainingContentSize;
- int remainingPaddingSize;
+/// This is not the same as `_current.stream` for sparse files, which are
+/// reported as expanded through [TarEntry.contents].
+/// For that reason, we prefer to drain this stream when skipping a tar entry.
+/// When we know we're skipping data, there's no point expanding sparse holes.
+///
+/// Draining this stream will set the [TarReader._currentStream] field back to
+/// null. There can only be one content stream at the time.
+class _CurrentEntryStream extends Stream<List<int>> {
+ _EntryStreamState state = _EntryStreamState.preListen;
- final EventSink<List<int>> out;
- void Function() onDone;
+ final TarReader _reader;
+ final Stream<Uint8List> _source;
- bool hadError = false;
- bool isInContent = true;
+ final StreamController<List<int>> _listener = StreamController(sync: true);
+ // ignore: cancel_subscriptions
+ StreamSubscription<List<int>>? _sourceSubscription;
- _OutgoingStreamGuard(this.remainingContentSize, this.out, this.onDone)
- : remainingPaddingSize = _paddingFor(remainingContentSize);
+ int _remainingContentSize;
+ int _remainingPaddingSize;
+ bool _hadError = false;
+ bool _isInContent = true;
+
+ _CurrentEntryStream(this._reader, this._source, this._remainingContentSize)
+ : _remainingPaddingSize = _paddingFor(_remainingContentSize);
+
+ @override
+ StreamSubscription<List<int>> listen(void Function(List<int> event)? onData,
+ {Function? onError, void Function()? onDone, bool? cancelOnError}) {
+ // Make sure that this entry is still the current one: If users store the
+ // contents of a tar entry, then read more tar entries, and finally try to
+ // read the stream of the old contents, they'd get an exception about the
+ // stream already being listened to.
+ // This can be a bit confusing, so this check enables a better error UX.
+ if (_reader._currentStream != this) {
+ throw StateError(
+ 'Tried listening to an outdated tar entry. \n'
+ 'As all tar entries found by a reader are backed by a single source '
+ 'stream, only the latest tar entry can be read. It looks like you '
+ 'stored the results of `tarEntry.contents` somewhere, called '
+ '`reader.moveNext()` and then read the contents of the previous '
+ 'entry.\n'
+ 'For more details, including a discussion of workarounds, see '
+ 'https://github.com/simolus3/tar/issues/18',
+ );
+ } else if (state != _EntryStreamState.preListen) {
+ throw StateError(
+ 'A tar entry has been listened to multiple times. \n'
+ 'As all tar entries are read from what\'s likely a single-'
+ 'subscription stream, this is unsupported. If you didn\'t read a tar '
+ 'entry multiple times yourself, perhaps you\'ve called `moveNext()` '
+ 'before reading contents?',
+ );
+ }
+
+ // Now we have a listener, so
+ state = _EntryStreamState.subscriptionActive;
+ // ignore: cancel_subscriptions
+ final sub = _sourceSubscription = _source.listen(
+ _forwardData,
+ onError: _forwardError,
+ onDone: _forwardDone,
+ );
+
+ _listener
+ ..onPause = sub.pause
+ ..onResume = sub.resume
+ ..onCancel = () {
+ // Pause the source subscription. When reading the next entry, the tar
+ // reader will drain the remaining source stream.
+ sub.pause();
+ state = _EntryStreamState.cancelled;
+ };
+
+ return _listener.stream.listen(
+ onData,
+ onError: onError,
+ onDone: onDone,
+ cancelOnError: cancelOnError,
+ );
+ }
static int _paddingFor(int contentSize) {
final offsetInLastBlock = contentSize.toUnsigned(blockSizeLog2);
@@ -907,47 +954,59 @@
return 0;
}
- @override
- void add(Uint8List event) {
- if (isInContent) {
- if (event.length <= remainingContentSize) {
+ void _assertInStateForForwarding() {
+ assert(state == _EntryStreamState.subscriptionActive &&
+ _listener.hasListener &&
+ !_listener.isPaused);
+ }
+
+ void _forwardData(Uint8List event) {
+ _assertInStateForForwarding();
+
+ if (_isInContent) {
+ if (event.length <= _remainingContentSize) {
// We can fully add this chunk as it consists entirely of data
- out.add(event);
- remainingContentSize -= event.length;
+ _listener.add(event);
+ _remainingContentSize -= event.length;
} else {
// We can add the first bytes as content, the others are padding that we
// shouldn't emit
- out.add(event.sublistView(0, remainingContentSize));
- isInContent = false;
- remainingPaddingSize -= event.length - remainingContentSize;
- remainingContentSize = 0;
+ _listener.add(event.sublistView(0, _remainingContentSize));
+ _isInContent = false;
+ _remainingPaddingSize -= event.length - _remainingContentSize;
+ _remainingContentSize = 0;
}
} else {
// Ok, the entire event is padding
- remainingPaddingSize -= event.length;
+ _remainingPaddingSize -= event.length;
}
// The underlying stream comes from pkg:tar, so if we get too many bytes
// that's a bug in this package.
- assert(remainingPaddingSize >= 0, 'Stream emitted to many bytes');
+ assert(_remainingPaddingSize >= 0, 'Stream emitted to many bytes');
}
- @override
- void addError(Object error, [StackTrace? stackTrace]) {
- hadError = true;
- out.addError(error, stackTrace);
+ void _forwardError(Object error, StackTrace trace) {
+ _assertInStateForForwarding();
+
+ _hadError = true;
+ _listener.addError(error, trace);
}
- @override
- void close() {
+ void _forwardDone() {
+ _assertInStateForForwarding();
+
+ // Now that the source stream is done, reset the stream state on the reader.
+ state = _EntryStreamState.done;
+ _sourceSubscription = null;
+ _reader._currentStream = null;
+
// If the stream stopped after an error, the user is already aware that
// something is wrong.
- if (remainingContentSize > 0 && !hadError) {
- out.addError(
+ if (_remainingContentSize > 0 && !_hadError) {
+ _listener.addError(
TarException('Unexpected end of tar file'), StackTrace.current);
}
-
- onDone();
- out.close();
+ _listener.close();
}
}
