Add reset API

Sources/CompressionCore/StreamingCompressor.swift

@@ -25,6 +25,12 @@ public protocol StreamingCompressor: ~Copyable, Sendable {
     mutating func finish(
         into output: inout OutputSpan<UInt8>
     ) throws(Failure) -> Bool
+
+    /// Discard all stream state and start over, keeping `configuration`.
+    ///
+    /// Any data compressed after this is independent of what came before, so it
+    /// decodes on its own. A stream that was never finished is dropped.
+    mutating func reset()
 }
 
 extension StreamingCompressor where Self: ~Copyable {

Sources/CompressionCore/StreamingDecompressor.swift

@@ -29,6 +29,12 @@ public protocol StreamingDecompressor: ~Copyable, Sendable {
         _ chunk: Span<UInt8>,
         into output: inout OutputSpan<UInt8>
     ) throws(Failure) -> Int
+
+    /// Discard all stream state and start over, keeping `configuration`.
+    ///
+    /// Any data compressed after this is independent of what came before, so it
+    /// decodes on its own. A stream that was never finished is dropped.
+    mutating func reset()
 }
 
 extension StreamingDecompressor where Self: ~Copyable {

Sources/Zlib/DeflateConfiguration.swift

@@ -8,7 +8,14 @@ extension Deflate {
         public var memory: MemoryUsage
         public var strategy: Strategy
 
-        public init(
+        public init() {
+            self.level = .default
+            self.format = .zlib
+            self.memory = .default
+            self.strategy = .default
+        }
+
+        package init(
             level: Level = .default,
             format: Format = .zlib,
             memory: MemoryUsage = .default,

Sources/Zlib/DeflateStreamingCompressor.swift

@@ -8,9 +8,13 @@ extension Deflate {
         @usableFromInline
         let stream: ZStreamBox
 
+        @usableFromInline
+        var isFinished: Bool
+
         public init(configuration: Configuration = .default) {
             self.configuration = configuration
             self.stream = .init()
+            self.isFinished = false
             unsafe stream.value.zalloc = nil
             unsafe stream.value.zfree = nil
             unsafe stream.value.opaque = nil
@@ -35,6 +39,8 @@ extension Deflate {
             _ chunk: Span<UInt8>,
             into output: inout OutputSpan<UInt8>
         ) throws(Deflate.Error) -> Int {
+            precondition(!self.isFinished, "Stream is finished, call reset if you want to start over")
+
             guard !output.isFull else { return 0 }
 
             unsafe stream.value.avail_in = UInt32(chunk.count)
@@ -68,31 +74,33 @@ extension Deflate {
         ) throws(Deflate.Error) -> Bool {
             guard output.freeCapacity > 0 else { return false }
 
-            let streamRef = stream
-            unsafe streamRef.value.avail_in = 0
-            unsafe streamRef.value.next_in = nil
+            unsafe stream.value.avail_in = 0
+            unsafe stream.value.next_in = nil
             try deflateStep(flag: CZLIB_Z_SYNC_FLUSH, into: &output)
 
             // zlib signals a complete flush by leaving output space unused.
-            return unsafe streamRef.value.avail_out != 0
+            return unsafe stream.value.avail_out != 0
         }
 
         @inlinable
         public func finish(
             into output: inout OutputSpan<UInt8>
         ) throws(Deflate.Error) -> Bool {
-            let streamRef = stream
-            unsafe streamRef.value.avail_in = 0
-
-            guard !output.isFull else {
-                return false
-            }
+            guard !self.isFinished else { return true }
+            unsafe stream.value.avail_in = 0
+            guard !output.isFull else { return false }
 
             let status = try deflateStep(flag: CZLIB_Z_FINISH, into: &output)
 
             return status == CZLIB_Z_STREAM_END
         }
 
+        @inlinable
+        public mutating func reset() {
+            unsafe czlib_z_deflateReset(&self.stream.value)
+            self.isFinished = false
+        }
+
         /// Run a single `deflate` call into the free space of `output`.
         ///
         /// `output` must have free capacity.

Sources/Zlib/DeflateStreamingDecompressor.swift

@@ -130,5 +130,12 @@ extension Deflate {
 
             return chunk.count - Int(unsafe stream.value.avail_in)
         }
+
+        @inlinable
+        public mutating func reset() {
+            unsafe czlib_z_inflateReset(&self.stream.value)
+            self.decompressedBytesCount = 0
+            self._isFinished = false
+        }
     }
 }

Sources/Zstd/ZstdStreamingCompressor.swift

@@ -29,6 +29,8 @@ extension Zstd: StreamingCompressionAlgorithm {
 
         @inlinable
         public mutating func compress(_ chunk: Span<UInt8>, into output: inout OutputSpan<UInt8>) throws(Zstd.Error) -> Int {
+            precondition(!self.isFinished, "Stream is finished, call reset if you want to start over")
+
             var consumed = 0
 
             while consumed < chunk.count {
@@ -87,13 +89,19 @@ extension Zstd: StreamingCompressionAlgorithm {
             return isFinished
         }
 
+        @inlinable
+        public mutating func reset() {
+            unsafe ZSTD_CCtx_reset(self.stream.value, ZSTD_reset_session_only)
+            self.isFinished = false
+        }
+
         @inlinable
         public mutating func flush(into output: inout OutputSpan<UInt8>) throws(Zstd.Error) -> Bool {
+            precondition(!isFinished, "Stream is finished, call reset if you want to start over")
             guard !output.isFull else { return false }
 
             let free = output.freeCapacity
             let result = unsafe output.withUnsafeMutableBufferPointer { tail, initialisedCount -> CZstd_StreamResult? in
-
                 guard free > 0, let base = tail.baseAddress else { return nil }
 
                 let result = unsafe CZstd_compressStream2NoInput(

Sources/Zstd/ZstdStreamingDecompressor.swift

@@ -116,5 +116,12 @@ extension Zstd: StreamingDecompressionAlgorithm {
 
             return consumed
         }
+
+        @inlinable
+        public mutating func reset() {
+            unsafe ZSTD_DCtx_reset(self.stream.value, ZSTD_reset_session_only)
+            self.isFinished = false
+            self.decompressedBytesCount = 0
+        }
     }
 }

Tests/DeflateTests/DeflateResetTests.swift

@@ -0,0 +1,136 @@
+import CompressionDeflate
+import Testing
+
+private let payload: [UInt8] = {
+    var bytes = [UInt8]()
+    while bytes.count < 4_096 { bytes += Array("the quick brown fox jumps over the lazy dog, ".utf8) }
+    return Array(bytes.prefix(4_096))
+}()
+
+private let other: [UInt8] = Array(repeating: 0x5A, count: 2_048)
+
+private func bytes(_ span: Span<UInt8>) -> [UInt8] {
+    var out = [UInt8]()
+    out.reserveCapacity(span.count)
+    for i in span.indices { out.append(span[i]) }
+    return out
+}
+
+@Suite("Deflate reset")
+struct DeflateResetTests {
+    @Test("A stream compressed after reset decodes on its own")
+    func streamAfterResetIsSelfContained() throws {
+        var compressor = Deflate.StreamingCompressor()
+        var firstStream = [UInt8]()
+        var secondStream = [UInt8]()
+
+        try compressor.compress(payload.span) { firstStream += bytes($0) }
+        try compressor.finish { firstStream += bytes($0) }
+        compressor.reset()
+        try compressor.compress(payload.span) { secondStream += bytes($0) }
+        try compressor.finish { secondStream += bytes($0) }
+
+        let decompressor = Deflate.Decompressor()
+        #expect(try decompressor.decompress(firstStream) == payload)
+        #expect(try decompressor.decompress(secondStream) == payload)
+    }
+
+    @Test("reset discards a stream that was never finished")
+    func resetDiscardsUnfinishedStream() throws {
+        var compressor = Deflate.StreamingCompressor()
+        var discarded = [UInt8]()
+        try compressor.compress(payload.span) { discarded += bytes($0) }
+
+        compressor.reset()
+
+        var stream = [UInt8]()
+        try compressor.compress(other.span) { stream += bytes($0) }
+        try compressor.finish { stream += bytes($0) }
+
+        #expect(try Deflate.Decompressor().decompress(stream) == other)
+    }
+
+    @Test("After reset the compressor behaves like a fresh one with the same configuration")
+    func resetKeepsConfiguration() throws {
+        let configuration = Deflate.CompressionConfiguration(level: .best, format: .gzip)
+
+        var reused = Deflate.StreamingCompressor(configuration: configuration)
+        var throwaway = [UInt8]()
+        try reused.compress(payload.span) { throwaway += bytes($0) }
+        try reused.finish { throwaway += bytes($0) }
+        reused.reset()
+        var reusedStream = [UInt8]()
+        try reused.compress(payload.span) { reusedStream += bytes($0) }
+        try reused.finish { reusedStream += bytes($0) }
+
+        var fresh = Deflate.StreamingCompressor(configuration: configuration)
+        var freshStream = [UInt8]()
+        try fresh.compress(payload.span) { freshStream += bytes($0) }
+        try fresh.finish { freshStream += bytes($0) }
+
+        #expect(reusedStream == freshStream, "reset must not drop level, format or memory")
+        let decompressor = Deflate.Decompressor(configuration: .init(format: .gzip))
+        #expect(try decompressor.decompress(reusedStream) == payload)
+    }
+
+    @Test("After reset a second independent stream decodes")
+    func resetAllowsASecondStream() throws {
+        let stream = try Deflate.Compressor().compress(payload)
+        var decompressor = Deflate.StreamingDecompressor()
+
+        var first = [UInt8]()
+        try decompressor.decompress(stream.span) { first += bytes($0) }
+        // #expect can't read a property off a ~Copyable value, so bind it first.
+        let finishedAfterFirst = decompressor.isFinished
+        #expect(first == payload)
+        #expect(finishedAfterFirst)
+
+        decompressor.reset()
+        let finishedAfterReset = decompressor.isFinished
+        #expect(!finishedAfterReset)
+
+        var second = [UInt8]()
+        try decompressor.decompress(stream.span) { second += bytes($0) }
+        let finishedAfterSecond = decompressor.isFinished
+        #expect(second == payload)
+        #expect(finishedAfterSecond)
+    }
+
+    /// Without this, a reused decompressor keeps counting bytes from the stream
+    /// before it and trips `maxDecompressedSize` on a stream that is within the limit.
+    @Test("reset clears the byte count so maxDecompressedSize applies per stream")
+    func resetClearsTheByteCount() throws {
+        let stream = try Deflate.Compressor().compress(payload)
+        let configuration = Deflate.DecompressionConfiguration(maxDecompressedSize: payload.count)
+        var decompressor = Deflate.StreamingDecompressor(configuration: configuration)
+
+        var first = [UInt8]()
+        try decompressor.decompress(stream.span) { first += bytes($0) }
+        #expect(first == payload)
+
+        decompressor.reset()
+
+        var second = [UInt8]()
+        try decompressor.decompress(stream.span) { second += bytes($0) }
+        #expect(second == payload)
+    }
+
+    @Test("reset discards a partially fed stream")
+    func resetDiscardsPartialInput() throws {
+        let stream = try Deflate.Compressor().compress(payload)
+        var decompressor = Deflate.StreamingDecompressor()
+
+        var partial = [UInt8]()
+        try decompressor.decompress(Array(stream.prefix(stream.count / 2)).span) { partial += bytes($0) }
+        let finishedAfterPartial = decompressor.isFinished
+        #expect(!finishedAfterPartial)
+
+        decompressor.reset()
+
+        var whole = [UInt8]()
+        try decompressor.decompress(stream.span) { whole += bytes($0) }
+        let finishedAtEnd = decompressor.isFinished
+        #expect(whole == payload)
+        #expect(finishedAtEnd)
+    }
+}