Unread extra data when stopping (#209)

Author: nhz2

docs/src/examples.md

@@ -157,7 +157,8 @@ eof(stream)   #> true
 In the case where you need to reuse the wrapped stream, the code above must be
 slightly modified because the transcoding stream may read more bytes than
 necessary from the wrapped stream. Wrapping the stream with `NoopStream` solves
-the problem because adjacent transcoding streams share the same buffer.
+the problem because any extra data read after the end of the chunk will be 
+stored back in the internal buffer of the wrapped transcoding stream.
 ```julia
 using CodecZlib
 using TranscodingStreams

ext/TestExt.jl

@@ -81,24 +81,26 @@ function TranscodingStreams.test_chunked_read(Encoder, Decoder)
     alpha = b"色即是空"
     encoder = Encoder()
     initialize(encoder)
-    for _ in 1:500
-        chunks = [rand(alpha, rand(0:100)) for _ in 1:rand(1:100)]
-        data = mapfoldl(x->transcode(encoder, x), vcat, chunks, init=UInt8[])
-        buffer = NoopStream(IOBuffer(data))
-        ok = true
-        for chunk in chunks
-            stream = TranscodingStream(Decoder(), buffer, stop_on_end=true)
-            ok &= read(stream) == chunk
-            ok &= position(stream) == length(chunk)
-            ok &= eof(stream)
-            ok &= isreadable(stream)
+    for sharedbuf in false:true
+        for _ in 1:500
+            chunks = [rand(alpha, rand(0:100)) for _ in 1:rand(1:100)]
+            data = mapfoldl(x->transcode(encoder, x), vcat, chunks, init=UInt8[])
+            buffer = NoopStream(IOBuffer(data))
+            ok = true
+            for chunk in chunks
+                stream = TranscodingStream(Decoder(), buffer; stop_on_end=true, sharedbuf)
+                ok &= read(stream) == chunk
+                ok &= position(stream) == length(chunk)
+                ok &= eof(stream)
+                ok &= isreadable(stream)
+                close(stream)
+            end
+            # read without stop_on_end should read the full data.
+            stream = TranscodingStream(Decoder(), IOBuffer(data))
+            ok &= read(stream) == reduce(vcat, chunks)
             close(stream)
+            Test.@test ok
         end
-        # read without stop_on_end should read the full data.
-        stream = TranscodingStream(Decoder(), IOBuffer(data))
-        ok &= read(stream) == reduce(vcat, chunks)
-        close(stream)
-        Test.@test ok
     end
     finalize(encoder)
 end

src/stream.jl

@@ -700,6 +700,10 @@ function callprocess(stream::TranscodingStream, inbuf::Buffer, outbuf::Buffer)
         makemargin!(outbuf, max(16, marginsize(outbuf) * 2))
     elseif state.code == :end && state.stop_on_end
         if stream.state.mode == :read
+            if stream.stream isa TranscodingStream && !has_sharedbuf(stream) && !iszero(buffersize(inbuf))
+                # unread data to match behavior if inbuf was shared.
+                GC.@preserve inbuf unsafe_unread(stream.stream, bufferptr(inbuf), buffersize(inbuf))
+            end
             changemode!(stream, :stop)
         end
     end