Use optimized stream chunker
diff --git a/lib/src/reader.dart b/lib/src/reader.dart
index f1877aa..df89c02 100644
--- a/lib/src/reader.dart
+++ b/lib/src/reader.dart
@@ -22,7 +22,7 @@
@sealed
class TarReader implements StreamIterator<TarEntry> {
/// A chunked stream iterator to enable us to get our data.
- final StreamIterator<Uint8List> _chunkedStream;
+ final StreamBlockReader _reader;
final PaxHeaders _paxHeaders = PaxHeaders();
final int _maxSpecialFileSize;
@@ -84,7 +84,7 @@
TarReader(Stream<List<int>> tarStream,
{int maxSpecialFileSize = defaultSpecialLength,
bool disallowTrailingData = false})
- : _chunkedStream = StreamIterator(tarStream.inChunks),
+ : _reader = StreamBlockReader(tarStream),
_checkNoTrailingData = disallowTrailingData,
_maxSpecialFileSize = maxSpecialFileSize;
@@ -236,7 +236,7 @@
// Note: Calling cancel is safe when the stream has already been completed.
// It's a noop in that case, which is what we want.
- return _chunkedStream.cancel();
+ return _reader.close();
}
/// Utility function for quickly iterating through all entries in [tarStream].
@@ -307,12 +307,12 @@
Uint8List? block;
do {
- block = await _chunkedStream.nextOrNull;
- if (block != null && !block.isAllZeroes) {
+ block = await _reader.nextBlock();
+ if (!block.isAllZeroes) {
throw TarException(
'Illegal content after the end of the tar archive.');
}
- } while (block != null && block.length == blockSize);
+ } while (block.length == blockSize);
// The stream is done when we couldn't read the full block.
}
@@ -325,7 +325,7 @@
/// Reads a 512-byte block or throws if [allowEmpty] is false.
Future<Uint8List> _readFullBlock({bool allowEmpty = false}) async {
- final block = await _chunkedStream.nextOrNull ?? emptyUint8List;
+ final block = await _reader.nextBlock();
if (block.length != blockSize && !(allowEmpty && block.isEmpty)) {
_unexpectedEof();
}
@@ -337,8 +337,8 @@
final result = Uint8List(blockSize * amount);
for (var i = 0; i < amount; i++) {
- final chunk = await _chunkedStream.nextOrNull;
- if (chunk == null || chunk.length != blockSize) {
+ final chunk = await _reader.nextBlock();
+ if (chunk.length != blockSize) {
_unexpectedEof();
}
@@ -361,10 +361,10 @@
if (rawHeader.isEmpty) return null;
if (rawHeader.isAllZeroes) {
- final next = await _chunkedStream.nextOrNull;
+ final next = await _reader.nextBlock();
// Exactly 1 block of zeroes is read and EOF is hit.
- if (next == null) return null;
+ if (next.isEmpty) return null;
if (next.isAllZeroes) {
// Two blocks of zeros are read - Normal EOF.
@@ -399,7 +399,7 @@
final streamBlockCount = numBlocks(sparseDataLength);
final safeStream = _publishStream(
- _chunkedStream.nextElements(streamBlockCount), sparseDataLength);
+ _reader.nextBlocks(streamBlockCount), sparseDataLength);
return sparseStream(safeStream, sparseHoles, header.size);
} else {
var size = header.size;
@@ -413,8 +413,7 @@
return _publishStream(const Stream<Never>.empty(), 0);
} else {
final blockCount = numBlocks(header.size);
- return _publishStream(
- _chunkedStream.nextElements(blockCount), header.size);
+ return _publishStream(_reader.nextBlocks(blockCount), header.size);
}
}
}
@@ -514,8 +513,8 @@
/// Ensures that [block] h as at least [n] tokens.
Future<void> feedTokens(int n) async {
while (newLineCount < n) {
- final newBlock = await _chunkedStream.nextOrNull;
- if (newBlock == null || newBlock.length < blockSize) {
+ final newBlock = await _reader.nextBlock();
+ if (newBlock.length < blockSize) {
throw TarException.header(
'GNU Sparse Map does not have enough lines!');
}
@@ -659,8 +658,8 @@
while (isExtended) {
// Ok, we have a new block of sparse headers to process
- final block = await _chunkedStream.nextOrNull;
- if (block == null || block.length < blockSize) {
+ final block = await _reader.nextBlock();
+ if (block.length < blockSize) {
throw TarException.header('Unexpected EoF while reading sparse maps');
}
@@ -852,40 +851,48 @@
/// expected. That indicates an invalid tar file though, since the correct size
/// is stored in the header.
class _OutgoingStreamGuard extends EventSink<Uint8List> {
- final int size;
- final int expectedBlockCount;
final EventSink<List<int>> out;
void Function() onDone;
- int emittedBlocks = 0;
+ int remainingContentSize;
+ int remainingPaddingSize;
+ bool isInContent = true;
bool hadError = false;
- _OutgoingStreamGuard(this.size, this.out, this.onDone)
- : expectedBlockCount = numBlocks(size);
+ _OutgoingStreamGuard(this.remainingContentSize, this.out, this.onDone)
+ : remainingPaddingSize = _paddingFor(remainingContentSize);
+
+ static int _paddingFor(int contentSize) {
+ final offsetInLastBlock = contentSize.toUnsigned(blockSizeLog2);
+ if (offsetInLastBlock != 0) {
+ return blockSize - offsetInLastBlock;
+ }
+ return 0;
+ }
@override
void add(Uint8List event) {
- if (event.length != blockSize) {
- addError(TarException('Unexpected end of tar file'), StackTrace.current);
- return;
- }
-
- emittedBlocks++;
- // We have checks limiting the length of outgoing streams. If the stream is
- // larger than expected, that's a bug in pkg:tar.
- assert(
- emittedBlocks <= expectedBlockCount,
- 'Stream now emitted $emittedBlocks blocks, but only expected '
- '$expectedBlockCount');
-
- if (emittedBlocks == expectedBlockCount) {
- // This is the last block, which we may have to truncate because tar
- // blocks are padded with zeroes at the end
- out.add(event.sublistView(0, size.toUnsigned(blockSizeLog2)));
+ if (isInContent) {
+ if (event.length < remainingContentSize) {
+ // We can fully add this chunk as it consists entirely of data
+ out.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;
+ }
} else {
- // This blocks just goes out fully
- out.add(event);
+ // Ok, the entire event is padding
+ 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');
}
@override
@@ -896,15 +903,15 @@
@override
void close() {
- onDone();
-
// If the stream stopped after an error, the user is already aware that
// something is wrong.
- if (emittedBlocks < expectedBlockCount && !hadError) {
+ if (remainingContentSize > 0 && !hadError) {
+ hadError = true;
out.addError(
TarException('Unexpected end of tar file'), StackTrace.current);
}
+ onDone();
out.close();
}
}
diff --git a/lib/src/utils.dart b/lib/src/utils.dart
index 1a0b018..c55917b 100644
--- a/lib/src/utils.dart
+++ b/lib/src/utils.dart
@@ -231,226 +231,323 @@
}
}
-Stream<Uint8List> _inChunks(Stream<List<int>> input) {
- Uint8List? startedChunk;
- int missingForNextChunk = 0;
+class StreamBlockReader {
+ final Stream<List<int>> _source;
+ StreamSubscription<List<int>>? _inputSubscription;
- Uint8List? pendingEvent;
- int offsetInPendingEvent = 0;
+ Completer<Uint8List>? _nextBlockCompleter;
- late StreamSubscription<List<int>> inputSubscription;
- final controller = StreamController<Uint8List>(sync: true);
+ StreamController<Uint8List>? _nextBlocksStream;
+ int _remainingBlocksInStream = 0;
- var isResuming = false;
+ Uint8List? _pendingBlock;
+ int _missingInPendingBlock = 0;
- void startChunk(Uint8List source, int startOffset) {
- assert(startedChunk == null);
+ Uint8List? _waitingForDispatch;
+ int _offsetInWaiting = 0;
+
+ bool _closed = false;
+
+ StreamBlockReader(this._source);
+
+ int get _blocksToServe {
+ if (_nextBlockCompleter != null) {
+ return 1;
+ } else {
+ return _remainingBlocksInStream;
+ }
+ }
+
+ void _listenOrResume() {
+ final subsription = _inputSubscription;
+ if (subsription == null) {
+ _inputSubscription =
+ _source.listen(_onData, onError: _onError, onDone: _onDone);
+ } else {
+ subsription.resume();
+ }
+ }
+
+ /// Emits ready [blocks].
+ ///
+ /// Returns whether we should pause emitting further data afterwards.
+ bool _emitBlocks(Uint8List blocks, {int amount = 1}) {
+ final stream = _nextBlocksStream;
+ final future = _nextBlockCompleter;
+ final subscription = _inputSubscription;
+ assert((future == null) ^ (stream == null));
+ assert(subscription != null);
+
+ final bool shouldPause;
+
+ if (stream != null) {
+ stream.add(blocks);
+ final remaining = _remainingBlocksInStream -= amount;
+ assert(remaining >= 0);
+
+ final done = remaining == 0;
+ if (done) {
+ stream.close();
+ _nextBlocksStream = null;
+ }
+
+ shouldPause = done || stream.isPausedOrClosed;
+ } else if (future != null) {
+ assert(amount == 1);
+ future.complete(blocks);
+ _nextBlockCompleter = null;
+ shouldPause = true;
+ } else {
+ // Exactly one of them must be non-null
+ throw AssertionError();
+ }
+
+ if (shouldPause && !subscription!.isPaused) {
+ subscription.pause();
+ }
+ return shouldPause;
+ }
+
+ void _startPendingBlock(Uint8List source, int startOffset) {
+ assert(_pendingBlock == null);
final availableData = source.length - startOffset;
assert(availableData < blockSize);
- startedChunk = Uint8List(blockSize)
+ _pendingBlock = Uint8List(blockSize)
..setAll(0, source.sublistView(startOffset));
- missingForNextChunk = blockSize - availableData;
+ _missingInPendingBlock = blockSize - availableData;
}
- void handleData(List<int> data) {
- assert(pendingEvent == null,
- 'Had pending events while receiving new data from source.');
- final typedData = data.asUint8List();
+ void _saveNewUndispatchedState(Uint8List chunk, int offsetInChunk) {
+ assert(_waitingForDispatch == null);
+ _saveUndispatchedState(chunk, offsetInChunk);
+ }
+
+ void _saveUndispatchedState(Uint8List chunk, int offsetInChunk) {
+ assert(_pendingBlock == null);
+ final remaining = chunk.length - offsetInChunk;
+ if (remaining < blockSize) {
+ _waitingForDispatch = null;
+ _offsetInWaiting = 0;
+ _startPendingBlock(chunk, offsetInChunk);
+ } else {
+ _waitingForDispatch = chunk;
+ _offsetInWaiting = offsetInChunk;
+ }
+ }
+
+ void _onData(List<int> chunk) {
+ assert(_waitingForDispatch == null,
+ 'Had pending data while receiving new event');
+
+ final typedChunk = chunk.asUint8List();
// The start offset of the new data that we didn't process yet.
var offsetInData = 0;
- bool saveStateIfPaused() {
- if (controller.isPausedOrClosed) {
- pendingEvent = typedData;
- offsetInPendingEvent = offsetInData;
- return true;
- }
- return false;
+
+ void savePendingChunk() {
+ _saveNewUndispatchedState(typedChunk, offsetInData);
}
- // Try completing the pending chunk first, if it exists
- if (startedChunk != null) {
- final started = startedChunk!;
- final startOffsetInStarted = blockSize - missingForNextChunk;
+ // Try finishing the pending block first, if it exists.
+ if (_pendingBlock != null) {
+ final started = _pendingBlock!;
+ final missing = _missingInPendingBlock;
+ final startOffsetInStarted = blockSize - missing;
- if (data.length >= missingForNextChunk) {
- // Fill up the chunk, then go on with the remaining data
- started.setAll(startOffsetInStarted,
- typedData.sublistView(0, missingForNextChunk));
- controller.add(started);
- offsetInData += missingForNextChunk;
+ if (typedChunk.length >= missing) {
+ // We can complete the block
+ started.setAll(
+ startOffsetInStarted, typedChunk.sublistView(0, missing));
+ offsetInData = missing;
+ final done = _emitBlocks(started);
- // We just finished serving the started chunk, so reset that
- startedChunk = null;
- missingForNextChunk = 0;
+ // We just finished serving the started block, so reset that
+ _pendingBlock = null;
+ _missingInPendingBlock = 0;
- // If the controller was paused in a response to the add, stop serving
- // events and store the rest of the input for later
- if (saveStateIfPaused()) return;
+ if (done) {
+ savePendingChunk();
+ return;
+ }
} else {
- // Ok, we can't finish the pending chunk with the new data but at least
+ // We can't complete the pending block with the new data, but at least
// we can continue filling it up
- started.setAll(startOffsetInStarted, typedData);
- missingForNextChunk -= typedData.length;
+ started.setAll(startOffsetInStarted, typedChunk);
+ _missingInPendingBlock -= typedChunk.length;
+ assert(_missingInPendingBlock > 0);
return;
}
}
- // The started chunk has been completed, continue by adding chunks as they
- // come.
- assert(startedChunk == null);
+ // When we get to this point, the started chunk has been completed and we
+ // can continue adding chunks as they come.
+ assert(_pendingBlock == null);
- while (offsetInData < typedData.length) {
- // Can we serve a full block from the new event
- if (offsetInData <= typedData.length - blockSize) {
- // Yup, then let's do that
- final end = offsetInData + blockSize;
- controller.add(typedData.sublistView(offsetInData, end));
- offsetInData = end;
+ // Serve as many blocks as we can from the current chunk
+ final blocksToServe = _blocksToServe;
+ assert(blocksToServe >= 1);
+ final availableBlocksInChunk =
+ (typedChunk.length - offsetInData) ~/ blockSize;
+ final blocksToDeliverNow = min(blocksToServe, availableBlocksInChunk);
- // Once again, stop and save state if the controller was paused.
- if (saveStateIfPaused()) return;
+ if (blocksToDeliverNow > 0) {
+ final length = blockSize * blocksToDeliverNow;
+ final end = offsetInData + length;
+ final done = _emitBlocks(typedChunk.sublistView(offsetInData, end),
+ amount: blocksToDeliverNow);
+ offsetInData = end;
+
+ // Once again, stop and save state if this has completed the output
+ if (done) {
+ savePendingChunk();
+ return;
+ }
+ }
+
+ // If we're here, the remaining data in the chunk was not large enough to
+ // serve a full block. Store the remaining data in a pending block.
+ _startPendingBlock(typedChunk, offsetInData);
+ }
+
+ void _onError(Object error, [StackTrace? trace]) {
+ final stream = _nextBlocksStream;
+ final future = _nextBlockCompleter;
+
+ if (stream != null) {
+ stream.addError(error, trace);
+ } else if (future != null) {
+ future.completeError(error, trace);
+ _nextBlockCompleter = null;
+ }
+
+ final subscription = _inputSubscription;
+ if (!subscription!.isPaused) {
+ subscription.pause();
+ }
+ }
+
+ void _onDone() {
+ if (_blocksToServe > 0) {
+ if (_pendingBlock != null) {
+ _emitBlocks(
+ _pendingBlock!.sublistView(0, blockSize - _missingInPendingBlock));
+ _pendingBlock = null;
+ _missingInPendingBlock = 0;
} else {
- // Ok, no full block but we can start a new pending chunk
- startChunk(typedData, offsetInData);
- break;
+ // Complete with an empty block
+ _emitBlocks(emptyUint8List);
}
+
+ _inputSubscription = null;
+ }
+ close();
+ }
+
+ void _checkIdle(String method) {
+ if (_nextBlockCompleter != null) {
+ throw StateError("Can't call $method before a previous call to "
+ 'nextBlock() has completed.');
+ }
+ if (_nextBlocksStream != null) {
+ throw StateError("Can't call $method before a previous call to "
+ 'nextBlocks() was completed.');
}
}
- void startResume() {
- isResuming = false;
- if (controller.isPausedOrClosed) return;
-
- // Start dispatching pending events before we resume the subscription on the
- // main stream.
- if (pendingEvent != null) {
- final pending = pendingEvent!;
- assert(startedChunk == null, 'Had pending events and a started chunk');
-
- while (offsetInPendingEvent < pending.length) {
- // Can we serve a full block from pending data?
- if (offsetInPendingEvent <= pending.length - blockSize) {
- final end = offsetInPendingEvent + blockSize;
- controller.add(pending.sublistView(offsetInPendingEvent, end));
- offsetInPendingEvent = end;
-
- // Pause if we got a pause request in response to the add
- if (controller.isPausedOrClosed) return;
- } else {
- // Store pending block that we can't finish with the pending chunk.
- startChunk(pending, offsetInPendingEvent);
- break;
- }
- }
-
- // Pending events have been dispatched
- offsetInPendingEvent = 0;
- pendingEvent = null;
- }
-
- // When we're here, all pending events have been dispatched and the
- // controller is not paused. Let's continue then!
- inputSubscription.resume();
- }
-
- void onDone() {
- // Input stream is done. If we have a block left over, emit that now
- if (startedChunk != null) {
- controller
- .add(startedChunk!.sublistView(0, blockSize - missingForNextChunk));
- }
- controller.close();
- }
-
- controller
- ..onListen = () {
- inputSubscription = input.listen(handleData,
- onError: controller.addError, onDone: onDone);
- }
- ..onPause = () {
- if (!inputSubscription.isPaused) {
- inputSubscription.pause();
- }
- }
- ..onResume = () {
- // This is a bit hacky. Our subscription is most likely going to be a
- // _BufferingStreamSubscription which will buffer events that were added
- // in this callback. However, we really want to know when the subscription
- // has been paused in response to a new event because we can handle that
- // very efficiently.
- if (!isResuming) {
- scheduleMicrotask(startResume);
- }
- isResuming = true;
- }
- ..onCancel = () {
- inputSubscription.cancel();
- };
-
- return controller.stream;
-}
-
-extension on StreamController<dynamic> {
- bool get isPausedOrClosed => isPaused || isClosed;
-}
-
-extension ReadInChunks on Stream<List<int>> {
- /// A stream emitting values in chunks of `512` byte blocks.
+ /// Reads the next block from the input stream.
///
- /// The last emitted chunk may be shorter than the regular block length.
- Stream<Uint8List> get inChunks => _inChunks(this);
-}
+ /// The returned list may be empty or smaller than [blockSize] if the input
+ /// stream ends.
+ Future<Uint8List> nextBlock() {
+ _checkIdle('nextBlock');
+ if (_closed) return Future.value(emptyUint8List);
-extension NextOrNull<T extends Object> on StreamIterator<T> {
- Future<T?> get nextOrNull async {
- if (await moveNext()) {
- return current;
- } else {
- return null;
+ // If we have pending data to serve, do that before we start resuming the
+ // stream subscription
+ final waiting = _waitingForDispatch;
+ if (waiting != null) {
+ final offset = _offsetInWaiting;
+ assert(waiting.length - offset >= blockSize);
+
+ final end = offset + blockSize;
+ final result = Future<Uint8List>.value(waiting.sublistView(offset, end));
+
+ if (end == waiting.length) {
+ // We just completed all data waiting for dispatch
+ _waitingForDispatch = null;
+ _offsetInWaiting = 0;
+ } else {
+ // Store new pending data
+ _saveUndispatchedState(waiting, offset + blockSize);
+ }
+ return result;
}
+
+ // Note: Our logic in _emitBlocks depends on this completer NOT being sync
+ final next = _nextBlockCompleter = Completer();
+ _listenOrResume();
+ return next.future;
}
- Stream<T> nextElements(int count) {
- if (count == 0) return const Stream<Never>.empty();
+ Stream<Uint8List> nextBlocks(int amount) {
+ _checkIdle('nextBlocks');
- var remaining = count;
- Future<Object?>? pendingOperation;
- final controller = StreamController<T>(sync: true);
+ if (amount == 0 || _closed) return const Stream.empty();
- void addNewEvent() {
- if (pendingOperation == null && remaining > 0) {
- // Start fetching data
- final fetchNext = nextOrNull.then<Object?>((value) {
- remaining--;
- if (value != null) {
- // Got data, let's add it
- controller.add(value);
- }
+ // This controller must not be sync: We add events in response to onListen
+ // and expect a delay after closing it in _emitBlocks
+ final controller = _nextBlocksStream = StreamController();
+ _remainingBlocksInStream = amount;
- if (value == null || remaining == 0) {
- // Finished, so close the controller
- return controller.close();
- } else {
- return Future.value();
- }
- }, onError: controller.addError);
+ void onListenOrResume() {
+ // Once again, check if we have pending data to serve
+ final waiting = _waitingForDispatch;
+ if (waiting != null) {
+ final offset = _offsetInWaiting;
+ final blocksWaiting = (waiting.length - offset) ~/ blockSize;
+ assert(blocksWaiting > 0);
- pendingOperation = fetchNext.whenComplete(() {
- pendingOperation = null;
- if (!controller.isPaused && remaining > 0) {
- // Controller isn't paused, so add another event
- addNewEvent();
- }
- });
+ final blocksToEmit = min(blocksWaiting, amount);
+ final end = offset + blocksToEmit * blockSize;
+ final done =
+ _emitBlocks(waiting.sublistView(offset, end), amount: blocksToEmit);
+ _saveUndispatchedState(waiting, end);
+ if (!done) {
+ _listenOrResume();
+ }
+ } else {
+ // No data waiting to be served, so we definitely have to start
+ // listening
+ _listenOrResume();
}
}
controller
- ..onListen = addNewEvent
- ..onResume = addNewEvent;
-
+ ..onListen = onListenOrResume
+ ..onPause = () {
+ final sub = _inputSubscription;
+ if (!sub!.isPaused) sub.pause();
+ }
+ ..onResume = onListenOrResume;
return controller.stream;
}
+
+ Future<void> close() async {
+ if (_closed) return;
+
+ _closed = true;
+ await _inputSubscription?.cancel();
+
+ _nextBlockCompleter?.complete(emptyUint8List);
+
+ final stream = _nextBlocksStream;
+ if (stream != null) {
+ await stream.close();
+ }
+ }
+}
+
+extension on StreamController<dynamic> {
+ bool get isPausedOrClosed => isPaused || isClosed;
}
diff --git a/pubspec.yaml b/pubspec.yaml
index 8886a5d..1e60694 100644
--- a/pubspec.yaml
+++ b/pubspec.yaml
@@ -10,7 +10,6 @@
async: ^2.6.0
meta: ^1.3.0
typed_data: ^1.3.0
- stream_transform: ^2.0.0
dev_dependencies:
charcode: ^1.2.0
