Merge branch 'master' into mkitti-frame-compressor

.github/workflows/CI.yml

@@ -34,7 +34,7 @@ jobs:
         with:
           version: ${{ matrix.version }}
           arch: ${{ matrix.arch }}
-      - uses: julia-actions/cache@v1
+      - uses: julia-actions/cache@v2
       - uses: julia-actions/julia-buildpkg@v1
       - uses: julia-actions/julia-runtest@v1
       - uses: julia-actions/julia-processcoverage@v1

src/decompression.jl

@@ -77,7 +77,14 @@ function TranscodingStreams.process(codec::ZstdDecompressor, input::Memory, outp
         error[] = ErrorException("zstd error")
         return Δin, Δout, :error
     else
-        return Δin, Δout, code == 0 ? :end : :ok
+        if code == 0
+            return Δin, Δout, :end
+        elseif input.size == 0 && code > 0
+            error[] = ErrorException("zstd frame truncated. Expected at least $(code) more bytes")
+            return Δin, Δout, :error
+        else
+            return Δin, Δout, :ok
+        end
     end
 end
 

src/libzstd.jl

@@ -84,8 +84,37 @@ function reset!(cstream::CStream, srcsize::Integer)
 
 end
 
-function compress!(cstream::CStream)
-    return LibZstd.ZSTD_compressStream(cstream, cstream.obuffer, cstream.ibuffer)
+"""
+    compress!(cstream::CStream; endOp=:continue)
+
+Compress a Zstandard `CStream`. Optionally, provide one one of the following end directives
+* `:continue` - (default) collect more data, encoder decides when to output compressed result, for optimal compression ratio
+* `:flush` - flush any data provided so far
+* `:end` - flush any remaining data _and_ close current frame
+
+If `:end` is provided on the first call to `compress!`, the size of the input data will be recorded in the frame header.
+This is equivalent to calling [`CodecZstd.LibZstd.ZSTD_compress2`](@ref).
+See also [`CodecZstd.LibZstd.ZSTD_CCtx_setPledgedSrcSize`](@ref).
+
+See the Zstd manual for additional details:
+https://facebook.github.io/zstd/zstd_manual.html
+"""
+function compress!(cstream::CStream; endOp::Union{Symbol,LibZstd.ZSTD_EndDirective}=LibZstd.ZSTD_e_continue)
+    return LibZstd.ZSTD_compressStream2(cstream, cstream.obuffer, cstream.ibuffer, endOp)
+end
+
+# Support endOp keyword of compress! above when providing a Symbol
+function Base.convert(::Type{LibZstd.ZSTD_EndDirective}, endOp::Symbol)
+    endOp = if endOp == :continue
+        LibZstd.ZSTD_e_continue
+    elseif endOp == :flush
+        LibZstd.ZSTD_e_flush
+    elseif endOp == :end
+        LibZstd.ZSTD_e_end
+    else
+        throw(ArgumentError("Received value `:$endOp` for `endOp`, but only :continue, :flush, or :end are allowed values."))
+    end
+    return endOp
 end
 
 function frameCompress!(cstream::CStream)

test/compress_endOp.jl

@@ -0,0 +1,61 @@
+using CodecZstd
+using Test
+
+@testset "compress! endOp = :continue" begin
+    data = rand(1:100, 1024*1024)
+    cstream = CodecZstd.CStream()
+    cstream.ibuffer.src = pointer(data)
+    cstream.ibuffer.size = sizeof(data)
+    cstream.ibuffer.pos = 0
+    cstream.obuffer.dst = Base.Libc.malloc(sizeof(data)*2)
+    cstream.obuffer.size = sizeof(data)*2
+    cstream.obuffer.pos = 0
+    try
+        GC.@preserve data begin
+            # default endOp
+            @test CodecZstd.compress!(cstream; endOp=:continue) == 0
+            @test CodecZstd.find_decompressed_size(cstream.obuffer.dst, cstream.obuffer.pos) == CodecZstd.ZSTD_CONTENTSIZE_UNKNOWN
+        end
+    finally
+        Base.Libc.free(cstream.obuffer.dst)
+    end
+end
+
+@testset "compress! endOp = :flush" begin
+    data = rand(1:100, 1024*1024)
+    cstream = CodecZstd.CStream()
+    cstream.ibuffer.src = pointer(data)
+    cstream.ibuffer.size = sizeof(data)
+    cstream.ibuffer.pos = 0
+    cstream.obuffer.dst = Base.Libc.malloc(sizeof(data)*2)
+    cstream.obuffer.size = sizeof(data)*2
+    cstream.obuffer.pos = 0
+    try
+        GC.@preserve data begin
+            @test CodecZstd.compress!(cstream; endOp=:flush) == 0
+            @test CodecZstd.find_decompressed_size(cstream.obuffer.dst, cstream.obuffer.pos) == CodecZstd.ZSTD_CONTENTSIZE_UNKNOWN
+        end
+    finally
+        Base.Libc.free(cstream.obuffer.dst)
+    end
+end
+
+@testset "compress! endOp = :end" begin
+    data = rand(1:100, 1024*1024)
+    cstream = CodecZstd.CStream()
+    cstream.ibuffer.src = pointer(data)
+    cstream.ibuffer.size = sizeof(data)
+    cstream.ibuffer.pos = 0
+    cstream.obuffer.dst = Base.Libc.malloc(sizeof(data)*2)
+    cstream.obuffer.size = sizeof(data)*2
+    cstream.obuffer.pos = 0
+    try
+        GC.@preserve data begin
+            # The frame should contain the decompressed size
+            @test CodecZstd.compress!(cstream; endOp=:end) == 0
+            @test CodecZstd.find_decompressed_size(cstream.obuffer.dst, cstream.obuffer.pos) == sizeof(data)
+        end
+    finally
+        Base.Libc.free(cstream.obuffer.dst)
+    end
+end

test/runtests.jl

@@ -3,6 +3,8 @@ using Random
 using TranscodingStreams
 using Test
 
+Random.seed!(1234)
+
 @testset "Zstd Codec" begin
     codec = ZstdCompressor()
     @test codec isa ZstdCompressor
@@ -32,6 +34,20 @@ using Test
     @test read(ZstdDecompressorStream(IOBuffer(data))) == b"foo"
     @test read(ZstdDecompressorStream(IOBuffer(vcat(data, data)))) == b"foofoo"
 
+    @testset "Truncated frames" begin
+        # issue #24
+        @test_throws ErrorException transcode(ZstdDecompressor, UInt8[])
+        for trial in 1:1000
+            local uncompressed_data = rand(UInt8, rand(0:100))
+            local compressed_data = transcode(ZstdCompressor, uncompressed_data)
+            local L = length(compressed_data)
+            for n in 0:L-1
+                @test_throws ErrorException transcode(ZstdDecompressor, compressed_data[1:n])
+            end
+            @test transcode(ZstdDecompressor, compressed_data) == uncompressed_data
+        end
+    end
+
     @test ZstdCompressorStream <: TranscodingStreams.TranscodingStream
     @test ZstdDecompressorStream <: TranscodingStreams.TranscodingStream
 
@@ -48,4 +64,6 @@ using Test
     else
         @test_throws ZstdError throw(ZstdError(0xffffffffffffffba))
     end
+
+    include("compress_endOp.jl")
 end