Add ZlibError type and move initialize into startproc

src/compression.jl

@@ -155,14 +155,6 @@ end
 # Methods
 # -------
 
-function TranscodingStreams.initialize(codec::CompressorCodec)
-    code = deflate_init!(codec.zstream, codec.level, codec.windowbits)
-    if code != Z_OK
-        zerror(codec.zstream, code)
-    end
-    return
-end
-
 function TranscodingStreams.finalize(codec::CompressorCodec)
     zstream = codec.zstream
     if zstream.state != C_NULL
@@ -174,33 +166,55 @@ function TranscodingStreams.finalize(codec::CompressorCodec)
     return
 end
 
-function TranscodingStreams.startproc(codec::CompressorCodec, state::Symbol, error::Error)
-    code = deflate_reset!(codec.zstream)
-    if code == Z_OK
-        return :ok
+function TranscodingStreams.startproc(codec::CompressorCodec, state::Symbol, error_ref::Error)
+    if codec.zstream.state == C_NULL
+        code = deflate_init!(codec.zstream, codec.level, codec.windowbits)
+        # errors in deflate_init! do not require clean up, so just throw
+        if code == Z_OK
+            return :ok
+        elseif code == Z_MEM_ERROR
+            throw(OutOfMemoryError())
+        elseif code == Z_STREAM_ERROR
+            error("Z_STREAM_ERROR: invalid parameter, this should be caught in the codec constructor")
+        elseif code == Z_VERSION_ERROR
+            error("Z_VERSION_ERROR: zlib library version is incompatible")
+        else
+            error("unexpected libz error code: $(code)")
+        end
     else
-        error[] = ErrorException(zlib_error_message(codec.zstream, code))
-        return :error
+        code = deflate_reset!(codec.zstream)
+        # errors in deflate_reset! do not require clean up, so just throw
+        if code == Z_OK
+            return :ok
+        elseif code == Z_STREAM_ERROR
+            error("Z_STREAM_ERROR: the source stream state was inconsistent")
+        else
+            error("unexpected libz error code: $(code)")
+        end
     end
 end
 
-function TranscodingStreams.process(codec::CompressorCodec, input::Memory, output::Memory, error::Error)
+function TranscodingStreams.process(codec::CompressorCodec, input::Memory, output::Memory, error_ref::Error)
     zstream = codec.zstream
+    if zstream.state == C_NULL
+        error("startproc must be called before process")
+    end
     zstream.next_in = input.ptr
     avail_in = min(input.size, typemax(UInt32))
     zstream.avail_in = avail_in
     zstream.next_out = output.ptr
     avail_out = min(output.size, typemax(UInt32))
     zstream.avail_out = avail_out
     code = deflate!(zstream, zstream.avail_in > 0 ? Z_NO_FLUSH : Z_FINISH)
+    @assert code != Z_STREAM_ERROR # state not clobbered
     Δin = Int(avail_in - zstream.avail_in)
     Δout = Int(avail_out - zstream.avail_out)
     if code == Z_OK
         return Δin, Δout, :ok
     elseif code == Z_STREAM_END
         return Δin, Δout, :end
     else
-        error[] = ErrorException(zlib_error_message(zstream, code))
+        error_ref[] = ErrorException(zlib_error_message(zstream, code))
         return Δin, Δout, :error
     end
 end

src/decompression.jl

@@ -143,14 +143,6 @@ end
 # Methods
 # -------
 
-function TranscodingStreams.initialize(codec::DecompressorCodec)
-    code = inflate_init!(codec.zstream, codec.windowbits)
-    if code != Z_OK
-        zerror(codec.zstream, code)
-    end
-    return
-end
-
 function TranscodingStreams.finalize(codec::DecompressorCodec)
     zstream = codec.zstream
     if zstream.state != C_NULL
@@ -162,18 +154,42 @@ function TranscodingStreams.finalize(codec::DecompressorCodec)
     return
 end
 
-function TranscodingStreams.startproc(codec::DecompressorCodec, ::Symbol, error::Error)
-    code = inflate_reset!(codec.zstream)
-    if code == Z_OK
-        return :ok
+function TranscodingStreams.startproc(codec::DecompressorCodec, ::Symbol, error_ref::Error)
+    # indicate that no input data is being provided for future zlib compat
+    codec.zstream.next_in = C_NULL
+    codec.zstream.avail_in = 0
+    if codec.zstream.state == C_NULL
+        code = inflate_init!(codec.zstream, codec.windowbits)
+        # errors in inflate_init! do not require clean up, so just throw
+        if code == Z_OK
+            return :ok
+        elseif code == Z_MEM_ERROR
+            throw(OutOfMemoryError())
+        elseif code == Z_STREAM_ERROR
+            error("Z_STREAM_ERROR: invalid parameter, this should be caught in the codec constructor")
+        elseif code == Z_VERSION_ERROR
+            error("Z_VERSION_ERROR: zlib library version is incompatible")
+        else
+            error("unexpected libz error code: $(code)")
+        end
     else
-        error[] = ErrorException(zlib_error_message(codec.zstream, code))
-        return :error
+        code = inflate_reset!(codec.zstream)
+        # errors in inflate_reset! do not require clean up, so just throw
+        if code == Z_OK
+            return :ok
+        elseif code == Z_STREAM_ERROR
+            error("Z_STREAM_ERROR: the source stream state was inconsistent")
+        else
+            error("unexpected libz error code: $(code)")
+        end
     end
 end
 
-function TranscodingStreams.process(codec::DecompressorCodec, input::Memory, output::Memory, error::Error)
+function TranscodingStreams.process(codec::DecompressorCodec, input::Memory, output::Memory, error_ref::Error)
     zstream = codec.zstream
+    if zstream.state == C_NULL
+        error("startproc must be called before process")
+    end
     zstream.next_in = input.ptr
 
     avail_in = min(input.size, typemax(UInt32))
@@ -182,14 +198,20 @@ function TranscodingStreams.process(codec::DecompressorCodec, input::Memory, out
     avail_out = min(output.size, typemax(UInt32))
     zstream.avail_out = avail_out
     code = inflate!(zstream, Z_NO_FLUSH)
+    @assert code != Z_STREAM_ERROR # state not clobbered
     Δin = Int(avail_in - zstream.avail_in)
     Δout = Int(avail_out - zstream.avail_out)
     if code == Z_OK
         return Δin, Δout, :ok
     elseif code == Z_STREAM_END
         return Δin, Δout, :end
+    elseif code == Z_MEM_ERROR
+        throw(OutOfMemoryError())
+    elseif code == Z_BUF_ERROR && iszero(input.size)
+        error_ref[] = ZlibError("the compressed stream may be truncated")
+        return Δin, Δout, :error
     else
-        error[] = ErrorException(zlib_error_message(zstream, code))
+        error_ref[] = ZlibError(zlib_error_message(zstream, code))
         return Δin, Δout, :error
     end
 end

src/libz.jl

@@ -39,9 +39,17 @@ end
 
 const Z_DEFAULT_COMPRESSION = Cint(-1)
 
-const Z_OK         = Cint(0)
-const Z_STREAM_END = Cint(1)
-const Z_BUF_ERROR  = Cint(-5)
+const Z_OK            = Cint(0)
+const Z_STREAM_END    = Cint(1)
+const Z_NEED_DICT     = Cint(2)
+const Z_ERRNO         = Cint(-1)
+const Z_STREAM_ERROR  = Cint(-2)
+const Z_DATA_ERROR    = Cint(-3)
+const Z_MEM_ERROR     = Cint(-4)
+const Z_BUF_ERROR     = Cint(-5)
+const Z_VERSION_ERROR = Cint(-6)
+# Return codes for the compression/decompression functions. Negative values
+# are errors, positive values are used for special but normal events.
 
 const Z_NO_FLUSH      = Cint(0)
 const Z_SYNC_FLUSH    = Cint(2)
@@ -59,7 +67,9 @@ function version()
     return unsafe_string(ccall((:zlibVersion, libz), Ptr{UInt8}, ()))
 end
 
-const zlib_version = version()
+# This is the version of zlib used to make this wrapper.
+# The `_init!` functions will return an error if the library is not compatible.
+const zlib_version = "1.3.1"
 
 function deflate_init!(zstream::ZStream, level::Integer, windowbits::Integer)
     return ccall((:deflateInit2_, libz), Cint, (Ref{ZStream}, Cint, Cint, Cint, Cint, Cint, Cstring, Cint), zstream, level, Z_DEFLATED, windowbits, #=default memlevel=#8, #=default strategy=#0, zlib_version, sizeof(ZStream))
@@ -93,14 +103,28 @@ function inflate!(zstream::ZStream, flush::Integer)
     return ccall((:inflate, libz), Cint, (Ref{ZStream}, Cint), zstream, flush)
 end
 
+# Error
+# -----
+
+struct ZlibError <: Exception
+    msg::String
+end
+
+function Base.showerror(io::IO, err::ZlibError)
+    print(io, "ZlibError: ")
+    print(io, err.msg)
+    nothing
+end
+
+
 function zerror(zstream::ZStream, code::Integer)
-    return throw(ErrorException(zlib_error_message(zstream, code)))
+    throw(ZlibError(zlib_error_message(zstream, code)))
 end
 
 function zlib_error_message(zstream::ZStream, code::Integer)
     if zstream.msg == C_NULL
-        return "zlib error: <no message> (code: $(code))"
+        return "<no message> (code: $(code))"
     else
-        return "zlib error: $(unsafe_string(zstream.msg)) (code: $(code))"
+        return "$(unsafe_string(zstream.msg)) (code: $(code))"
     end
 end

test/runtests.jl

@@ -1,4 +1,5 @@
 using CodecZlib
+using CodecZlib: ZlibError
 using Test
 using Aqua: Aqua
 using TranscodingStreams:
@@ -56,7 +57,7 @@ const testdir = @__DIR__
     gzip_data_corrupted[1] = 0x00  # corrupt header
     file = IOBuffer(gzip_data_corrupted)
     stream = GzipDecompressorStream(file)
-    @test_throws ErrorException read(stream)
+    @test_throws ZlibError read(stream)
     @test_throws ArgumentError read(stream)
     @test !isopen(stream)
     @test isopen(file)
@@ -160,7 +161,7 @@ end
     close(stream)
 
     stream = TranscodingStream(GzipDecompressor(gziponly=true), IOBuffer(zlib_data))
-    @test_throws Exception read(stream)
+    @test_throws ZlibError read(stream)
     close(stream)
 
     file = IOBuffer(b"foo")
@@ -251,7 +252,7 @@ end
 
     @testset "panic" begin
         stream = TranscodingStream(GzipDecompressor(), IOBuffer("some invalid data"))
-        @test_throws ErrorException read(stream)
+        @test_throws ZlibError read(stream)
         @test_throws ArgumentError eof(stream)
     end
 
@@ -304,3 +305,32 @@ end
         @test_throws ArgumentError TranscodingStreams.stats(stream)
     end
 end
+
+@testset "unexpected end of stream errors" begin
+    tests = [
+        (ZlibCompressor, ZlibDecompressor),
+        (DeflateCompressor, DeflateDecompressor),
+        (GzipCompressor, GzipDecompressor),
+    ]
+    @testset "$(encoder)" for (encoder, decoder) in tests
+        local uncompressed = rand(UInt8, 1000)
+        local compressed = transcode(encoder, uncompressed)
+        for i in 0:length(compressed)-1
+            @test_throws ZlibError("the compressed stream may be truncated") transcode(decoder, compressed[1:i])
+        end
+        @test transcode(decoder, compressed) == uncompressed
+        # compressing empty vector should still work
+        @test transcode(decoder, transcode(encoder, UInt8[])) == UInt8[]
+    end
+end
+@testset "data errors" begin
+    @test_throws ZlibError transcode(ZlibDecompressor, zeros(UInt8, 10))
+    local uncompressed = rand(UInt8, 1000)
+    local compressed = transcode(ZlibCompressor, uncompressed)
+    compressed[70] ⊻= 0x01
+    @test_throws ZlibError transcode(ZlibDecompressor, compressed)
+end
+@testset "error printing" begin
+    @test sprint(Base.showerror, ZlibError("test error message")) ==
+        "ZlibError: test error message"
+end