diff --git a/CHANGELOG.md b/CHANGELOG.md
index 561163d..6c54d3b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,7 +1,7 @@
 ## 0.4.0
 
 - Support generating tar files with GNU-style long link names
- - Add `format` parameter to `tarWritingSink` and `tarTransformerWith`
+ - Add `format` parameter to `tarWritingSink` and `tarWriterWith`
 
 ## 0.3.3
 
diff --git a/example/main.dart b/example/main.dart
index 65cab46..6dfd101 100644
--- a/example/main.dart
+++ b/example/main.dart
@@ -1,3 +1,4 @@
+import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
 
@@ -5,7 +6,8 @@
 
 Future<void> main() async {
   // Start reading a tar file
-  final reader = TarReader(File('reference/gnu.tar').openRead());
+  final reader =
+      TarReader(File('reference/neats_test/gnu-incremental.tar').openRead());
 
   while (await reader.moveNext()) {
     final header = reader.current.header;
@@ -13,7 +15,10 @@
 
     // Print the output if it's a regular file
     if (header.typeFlag == TypeFlag.reg) {
-      await reader.current.contents.transform(utf8.decoder).forEach(print);
+      await reader.current.contents.forEach((chunk) {
+        print(chunk.length);
+        print(utf8.decode(chunk));
+      });
     }
   }
 
diff --git a/lib/src/constants.dart b/lib/src/constants.dart
index aac7669..312a245 100644
--- a/lib/src/constants.dart
+++ b/lib/src/constants.dart
@@ -258,3 +258,4 @@
 
 /// A full TAR block of zeros.
 final zeroBlock = Uint8List(blockSize);
+final emptyUint8List = Uint8List(0);
diff --git a/lib/src/reader.dart b/lib/src/reader.dart
index 713235a..f1877aa 100644
--- a/lib/src/reader.dart
+++ b/lib/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';
 
@@ -23,13 +22,10 @@
 @sealed
 class TarReader implements StreamIterator<TarEntry> {
   /// A chunked stream iterator to enable us to get our data.
-  final ChunkedStreamReader<int> _chunkedStream;
+  final StreamIterator<Uint8List> _chunkedStream;
   final PaxHeaders _paxHeaders = PaxHeaders();
   final int _maxSpecialFileSize;
 
-  /// Skip the next [_skipNext] elements when reading in the stream.
-  int _skipNext = 0;
-
   TarEntry? _current;
 
   /// The underlying content stream for the [_current] entry. Draining this
@@ -88,7 +84,7 @@
   TarReader(Stream<List<int>> tarStream,
       {int maxSpecialFileSize = defaultSpecialLength,
       bool disallowTrailingData = false})
-      : _chunkedStream = ChunkedStreamReader(tarStream),
+      : _chunkedStream = StreamIterator(tarStream.inChunks),
         _checkNoTrailingData = disallowTrailingData,
         _maxSpecialFileSize = maxSpecialFileSize;
 
@@ -152,13 +148,7 @@
     // iterates through one or more "header files" until it finds a
     // "normal file".
     while (true) {
-      if (_skipNext > 0) {
-        await _readFullBlock(_skipNext);
-        _skipNext = 0;
-      }
-
-      final rawHeader =
-          await _readFullBlock(blockSize, allowEmpty: eofAcceptable);
+      final rawHeader = await _readFullBlock(allowEmpty: eofAcceptable);
 
       nextHeader = await _readHeader(rawHeader);
       if (nextHeader == null) {
@@ -179,20 +169,19 @@
       if (nextHeader.typeFlag == TypeFlag.xHeader ||
           nextHeader.typeFlag == TypeFlag.xGlobalHeader) {
         format = format.mayOnlyBe(TarFormat.pax);
-        final paxHeaderSize = _checkSpecialSize(nextHeader.size);
-        final rawPaxHeaders = await _readFullBlock(paxHeaderSize);
+        final paxHeaderBlocks = await _readFullBlocks(
+            _checkSpecialBlocks(numBlocks(nextHeader.size)));
+        final rawPaxHeaders = paxHeaderBlocks.sublistView(0, nextHeader.size);
 
         _paxHeaders.readPaxHeaders(
             rawPaxHeaders, nextHeader.typeFlag == TypeFlag.xGlobalHeader);
-        _markPaddingToSkip(paxHeaderSize);
-
-        // This is a meta header affecting the next header.
+        // This is a meta header affecting the next header
         continue;
       } else if (nextHeader.typeFlag == TypeFlag.gnuLongLink ||
           nextHeader.typeFlag == TypeFlag.gnuLongName) {
         format = format.mayOnlyBe(TarFormat.gnu);
-        final realName = await _readFullBlock(
-            _checkSpecialSize(nextBlockSize(nextHeader.size)));
+        final realName = await _readFullBlocks(
+            _checkSpecialBlocks(numBlocks(nextHeader.size)));
 
         final readName = realName.readString(0, realName.length);
         if (nextHeader.typeFlag == TypeFlag.gnuLongName) {
@@ -300,13 +289,14 @@
     }
   }
 
-  int _checkSpecialSize(int size) {
+  int _checkSpecialBlocks(int count) {
+    final size = blockSize * count;
     if (size > _maxSpecialFileSize) {
       throw TarException(
           'TAR file contains hidden entry with an invalid size of $size.');
     }
 
-    return size;
+    return count;
   }
 
   /// Ater we detected the end of a tar file, optionally check for trailing
@@ -314,15 +304,15 @@
   Future<void> _handleExpectedEof() async {
     if (_checkNoTrailingData) {
       // Trailing zeroes are okay, but don't allow any more data here.
-      Uint8List block;
+      Uint8List? block;
 
       do {
-        block = await _chunkedStream.readBytes(blockSize);
-        if (!block.isAllZeroes) {
+        block = await _chunkedStream.nextOrNull;
+        if (block != null && !block.isAllZeroes) {
           throw TarException(
               'Illegal content after the end of the tar archive.');
         }
-      } while (block.length == blockSize);
+      } while (block != null && block.length == blockSize);
       // The stream is done when we couldn't read the full block.
     }
 
@@ -333,17 +323,31 @@
     throw TarException.header('Unexpected end of file');
   }
 
-  /// Reads a block with the requested [size], or throws an unexpected EoF
-  /// exception.
-  Future<Uint8List> _readFullBlock(int size, {bool allowEmpty = false}) async {
-    final block = await _chunkedStream.readBytes(size);
-    if (block.length != size && !(allowEmpty && block.isEmpty)) {
+  /// Reads a 512-byte block or throws if [allowEmpty] is false.
+  Future<Uint8List> _readFullBlock({bool allowEmpty = false}) async {
+    final block = await _chunkedStream.nextOrNull ?? emptyUint8List;
+    if (block.length != blockSize && !(allowEmpty && block.isEmpty)) {
       _unexpectedEof();
     }
 
     return block;
   }
 
+  Future<Uint8List> _readFullBlocks(int amount) async {
+    final result = Uint8List(blockSize * amount);
+
+    for (var i = 0; i < amount; i++) {
+      final chunk = await _chunkedStream.nextOrNull;
+      if (chunk == null || chunk.length != blockSize) {
+        _unexpectedEof();
+      }
+
+      result.setAll(i * blockSize, chunk);
+    }
+
+    return result;
+  }
+
   /// Reads the next block header and assumes that the underlying reader
   /// is already aligned to a block boundary. It returns the raw block of the
   /// header in case further processing is required.
@@ -357,12 +361,12 @@
     if (rawHeader.isEmpty) return null;
 
     if (rawHeader.isAllZeroes) {
-      rawHeader = await _chunkedStream.readBytes(blockSize);
+      final next = await _chunkedStream.nextOrNull;
 
       // Exactly 1 block of zeroes is read and EOF is hit.
-      if (rawHeader.isEmpty) return null;
+      if (next == null) return null;
 
-      if (rawHeader.isAllZeroes) {
+      if (next.isAllZeroes) {
         // Two blocks of zeros are read - Normal EOF.
         return null;
       }
@@ -393,9 +397,9 @@
       final sparseDataLength =
           sparseData.fold<int>(0, (value, element) => value + element.length);
 
-      final streamLength = nextBlockSize(sparseDataLength);
-      final safeStream =
-          _publishStream(_chunkedStream.readStream(streamLength), streamLength);
+      final streamBlockCount = numBlocks(sparseDataLength);
+      final safeStream = _publishStream(
+          _chunkedStream.nextElements(streamBlockCount), sparseDataLength);
       return sparseStream(safeStream, sparseHoles, header.size);
     } else {
       var size = header.size;
@@ -408,9 +412,9 @@
       if (size == 0) {
         return _publishStream(const Stream<Never>.empty(), 0);
       } else {
-        _markPaddingToSkip(size);
+        final blockCount = numBlocks(header.size);
         return _publishStream(
-            _chunkedStream.readStream(header.size), header.size);
+            _chunkedStream.nextElements(blockCount), header.size);
       }
     }
   }
@@ -443,15 +447,6 @@
     });
   }
 
-  /// Skips to the next block after reading [readSize] bytes from the beginning
-  /// of a previous block.
-  void _markPaddingToSkip(int readSize) {
-    final offsetInLastBlock = readSize.toUnsigned(blockSizeLog2);
-    if (offsetInLastBlock != 0) {
-      _skipNext = blockSize - offsetInLastBlock;
-    }
-  }
-
   /// Checks the PAX headers for GNU sparse headers.
   /// If they are found, then this function reads the sparse map and returns it.
   /// This assumes that 0.0 headers have already been converted to 0.1 headers
@@ -519,8 +514,8 @@
     /// Ensures that [block] h as at least [n] tokens.
     Future<void> feedTokens(int n) async {
       while (newLineCount < n) {
-        final newBlock = await _chunkedStream.readBytes(blockSize);
-        if (newBlock.length < blockSize) {
+        final newBlock = await _chunkedStream.nextOrNull;
+        if (newBlock == null || newBlock.length < blockSize) {
           throw TarException.header(
               'GNU Sparse Map does not have enough lines!');
         }
@@ -629,7 +624,7 @@
   /// Thus, this function will read from the chunked stream iterator to fetch
   /// extra headers.
   ///
-  /// See also: https://www.gnu.org/software/tar/manual/html_section/tar_94.html#SEC191
+  /// See also: https://www.gnu.org/software/tar/manual/tar.html#Old-GNU-Format
   Future<List<SparseEntry>> _readOldGNUSparseMap(
       HeaderImpl header, Uint8List rawHeader) async {
     // Make sure that the input format is GNU.
@@ -639,54 +634,46 @@
       throw TarException.header('Tried to read sparse map of non-GNU header');
     }
 
+    // Read the real size of the file when sparse holes are expanded.
     header.size = rawHeader.readNumeric(483, 12);
-    final sparseMaps = <Uint8List>[];
+    final sparseEntries = <SparseEntry>[];
 
-    var sparse = rawHeader.sublistView(386, 483);
-    sparseMaps.add(sparse);
+    bool readEntry(Uint8List source, int offset) {
+      // If a sparse header starts with a null byte, it marks the end of the
+      // sparse structures.
+      if (rawHeader[offset] == 0) return false;
 
-    while (true) {
-      final maxEntries = sparse.length ~/ 24;
-      if (sparse[24 * maxEntries] > 0) {
-        // If there are more entries, read an extension header and parse its
-        // entries.
-        sparse = await _chunkedStream.readBytes(blockSize);
-        sparseMaps.add(sparse);
-        continue;
+      final fileOffset = source.readNumeric(offset, 12);
+      final length = source.readNumeric(offset + 12, 12);
+
+      sparseEntries.add(SparseEntry(fileOffset, length));
+      return true;
+    }
+
+    // The first four sparse headers are stored in the tar header itself
+    for (var i = 0; i < 4; i++) {
+      final offset = 386 + 24 * i;
+      if (!readEntry(rawHeader, offset)) break;
+    }
+    var isExtended = rawHeader[482] != 0;
+
+    while (isExtended) {
+      // Ok, we have a new block of sparse headers to process
+      final block = await _chunkedStream.nextOrNull;
+      if (block == null || block.length < blockSize) {
+        throw TarException.header('Unexpected EoF while reading sparse maps');
       }
 
-      break;
-    }
-
-    try {
-      return _processOldGNUSparseMap(sparseMaps);
-    } on FormatException {
-      throw TarException('Invalid old GNU Sparse Map');
-    }
-  }
-
-  /// Process [sparseMaps], which is known to be an OLD GNU v0.1 sparse map.
-  ///
-  /// For details, see https://www.gnu.org/software/tar/manual/html_section/tar_94.html#SEC191
-  List<SparseEntry> _processOldGNUSparseMap(List<Uint8List> sparseMaps) {
-    final sparseData = <SparseEntry>[];
-
-    for (final sparseMap in sparseMaps) {
-      final maxEntries = sparseMap.length ~/ 24;
-      for (var i = 0; i < maxEntries; i++) {
-        // This termination condition is identical to GNU and BSD tar.
-        if (sparseMap[i * 24] == 0) {
-          // Don't return, need to process extended headers (even if empty)
-          break;
-        }
-
-        final offset = sparseMap.readNumeric(i * 24, 12);
-        final length = sparseMap.readNumeric(i * 24 + 12, 12);
-
-        sparseData.add(SparseEntry(offset, length));
+      // A full block of sparse data contains up to 21 entries
+      for (var i = 0; i < 21; i++) {
+        if (!readEntry(block, i * 24)) break;
       }
+
+      // The last bytes indicates whether another sparse header block follows.
+      isExtended = block[504] != 0;
     }
-    return sparseData;
+
+    return sparseEntries;
   }
 }
 
@@ -864,27 +851,41 @@
 /// [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<List<int>> {
-  final int expectedSize;
+class _OutgoingStreamGuard extends EventSink<Uint8List> {
+  final int size;
+  final int expectedBlockCount;
   final EventSink<List<int>> out;
   void Function() onDone;
 
-  int emittedSize = 0;
+  int emittedBlocks = 0;
   bool hadError = false;
 
-  _OutgoingStreamGuard(this.expectedSize, this.out, this.onDone);
+  _OutgoingStreamGuard(this.size, this.out, this.onDone)
+      : expectedBlockCount = numBlocks(size);
 
   @override
-  void add(List<int> event) {
-    emittedSize += event.length;
+  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(
-        emittedSize <= expectedSize,
-        'Stream now emitted $emittedSize bytes, but only expected '
-        '$expectedSize');
+        emittedBlocks <= expectedBlockCount,
+        'Stream now emitted $emittedBlocks blocks, but only expected '
+        '$expectedBlockCount');
 
-    out.add(event);
+    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)));
+    } else {
+      // This blocks just goes out fully
+      out.add(event);
+    }
   }
 
   @override
@@ -899,7 +900,7 @@
 
     // If the stream stopped after an error, the user is already aware that
     // something is wrong.
-    if (emittedSize < expectedSize && !hadError) {
+    if (emittedBlocks < expectedBlockCount && !hadError) {
       out.addError(
           TarException('Unexpected end of tar file'), StackTrace.current);
     }
diff --git a/lib/src/utils.dart b/lib/src/utils.dart
index a6b7e13..1a0b018 100644
--- a/lib/src/utils.dart
+++ b/lib/src/utils.dart
@@ -1,3 +1,4 @@
+import 'dart:async';
 import 'dart:convert';
 import 'dart:math';
 import 'dart:typed_data';
@@ -229,3 +230,227 @@
     yield Uint8List(remainingBytes);
   }
 }
+
+Stream<Uint8List> _inChunks(Stream<List<int>> input) {
+  Uint8List? startedChunk;
+  int missingForNextChunk = 0;
+
+  Uint8List? pendingEvent;
+  int offsetInPendingEvent = 0;
+
+  late StreamSubscription<List<int>> inputSubscription;
+  final controller = StreamController<Uint8List>(sync: true);
+
+  var isResuming = false;
+
+  void startChunk(Uint8List source, int startOffset) {
+    assert(startedChunk == null);
+    final availableData = source.length - startOffset;
+    assert(availableData < blockSize);
+
+    startedChunk = Uint8List(blockSize)
+      ..setAll(0, source.sublistView(startOffset));
+    missingForNextChunk = blockSize - availableData;
+  }
+
+  void handleData(List<int> data) {
+    assert(pendingEvent == null,
+        'Had pending events while receiving new data from source.');
+    final typedData = data.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;
+    }
+
+    // Try completing the pending chunk first, if it exists
+    if (startedChunk != null) {
+      final started = startedChunk!;
+      final startOffsetInStarted = blockSize - missingForNextChunk;
+
+      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;
+
+        // We just finished serving the started chunk, so reset that
+        startedChunk = null;
+        missingForNextChunk = 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;
+      } else {
+        // Ok, we can't finish the pending chunk with the new data but at least
+        // we can continue filling it up
+        started.setAll(startOffsetInStarted, typedData);
+        missingForNextChunk -= typedData.length;
+        return;
+      }
+    }
+
+    // The started chunk has been completed, continue by adding chunks as they
+    // come.
+    assert(startedChunk == 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;
+
+        // Once again, stop and save state if the controller was paused.
+        if (saveStateIfPaused()) return;
+      } else {
+        // Ok, no full block but we can start a new pending chunk
+        startChunk(typedData, offsetInData);
+        break;
+      }
+    }
+  }
+
+  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.
+  ///
+  /// The last emitted chunk may be shorter than the regular block length.
+  Stream<Uint8List> get inChunks => _inChunks(this);
+}
+
+extension NextOrNull<T extends Object> on StreamIterator<T> {
+  Future<T?> get nextOrNull async {
+    if (await moveNext()) {
+      return current;
+    } else {
+      return null;
+    }
+  }
+
+  Stream<T> nextElements(int count) {
+    if (count == 0) return const Stream<Never>.empty();
+
+    var remaining = count;
+    Future<Object?>? pendingOperation;
+    final controller = StreamController<T>(sync: true);
+
+    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);
+          }
+
+          if (value == null || remaining == 0) {
+            // Finished, so close the controller
+            return controller.close();
+          } else {
+            return Future.value();
+          }
+        }, onError: controller.addError);
+
+        pendingOperation = fetchNext.whenComplete(() {
+          pendingOperation = null;
+          if (!controller.isPaused && remaining > 0) {
+            // Controller isn't paused, so add another event
+            addNewEvent();
+          }
+        });
+      }
+    }
+
+    controller
+      ..onListen = addNewEvent
+      ..onResume = addNewEvent;
+
+    return controller.stream;
+  }
+}
diff --git a/pubspec.yaml b/pubspec.yaml
index 1e60694..8886a5d 100644
--- a/pubspec.yaml
+++ b/pubspec.yaml
@@ -10,6 +10,7 @@
   async: ^2.6.0
   meta: ^1.3.0
   typed_data: ^1.3.0
+  stream_transform: ^2.0.0
 
 dev_dependencies:
   charcode: ^1.2.0
diff --git a/test/utils_test.dart b/test/utils_test.dart
index 32de010..e8d5ea2 100644
--- a/test/utils_test.dart
+++ b/test/utils_test.dart
@@ -183,6 +183,10 @@
       });
     }
   });
+
+  group('inChunks', () {
+    test('keeps data in tact', () {});
+  });
 }
 
 Uint8List _bytes(String str) {