redesign APIs

Author: bicycle1885

src/TranscodingStreams.jl

@@ -2,10 +2,13 @@ module TranscodingStreams
 
 export TranscodingStream
 
+include("memory.jl")
+include("buffer.jl")
 include("codec.jl")
 include("state.jl")
 include("stream.jl")
 include("io.jl")
 include("identity.jl")
+include("testtools.jl")
 
 end # module

src/buffer.jl

@@ -0,0 +1,176 @@
+# Buffer
+# ======
+
+# Data Layout
+# -----------
+#
+# Buffered data are stored in `data` and two position fields are used to keep
+# track of buffered data and margin.
+#
+#             buffered data     margin
+#            |<----------->||<----------->|
+#     |....xxxxxxxxxXXXXXXXXXXXXXXX..............|
+#     ^    ^        ^              ^             ^
+#     1    markpos  bufferpos      marginpos     endof(data)
+
+mutable struct Buffer
+    data::Vector{UInt8}
+    markpos::Int
+    bufferpos::Int
+    marginpos::Int
+
+    function Buffer(size::Integer)
+        @assert size > 0
+        return new(Vector{UInt8}(size), 0, 1, 1)
+    end
+end
+
+function Base.length(buf::Buffer)
+    return length(buf.data)
+end
+
+function bufferptr(buf)
+    return pointer(buf.data, buf.bufferpos)
+end
+
+function buffersize(buf)
+    return buf.marginpos - buf.bufferpos
+end
+
+function buffermem(buf)
+    return Memory(bufferptr(buf), buffersize(buf))
+end
+
+function readbyte!(buf)
+    b = buf.data[buf.bufferpos]
+    buf.bufferpos += 1
+    return b
+end
+
+function writebyte!(buf, b::UInt8)
+    buf.data[buf.marginpos] = b
+    buf.marginpos += 1
+    return 1
+end
+
+function marginptr(buf)
+    return pointer(buf.data, buf.marginpos)
+end
+
+function marginsize(buf)
+    return endof(buf.data) - buf.marginpos + 1
+end
+
+function marginmem(buf)
+    return Memory(marginptr(buf), marginsize(buf))
+end
+
+function mark!(buf::Buffer)
+    return buf.markpos = buf.bufferpos
+end
+
+function unmark!(buf::Buffer)
+    if buf.markpos == 0
+        return false
+    else
+        buf.markpos = 0
+        return true
+    end
+end
+
+function reset!(buf::Buffer)
+    if buf.markpos == 0
+        throw(ArgumentError("not marked"))
+    end
+    buf.bufferpos = buf.markpos
+    buf.markpos = 0
+    return buf.bufferpos
+end
+
+# Make margin with ≥`minsize`.
+function makemargin!(buf, minsize::Int)::Int
+    if buffersize(buf) == 0
+        if buf.markpos == 0
+            buf.bufferpos = buf.marginpos = 1
+        else
+            buf.bufferpos = buf.marginpos = buf.markpos
+        end
+    end
+    if marginsize(buf) < minsize
+        # shift data to left
+        if buf.markpos == 0
+            datapos = buf.bufferpos
+            datasize = buffersize(buf)
+        else
+            datapos = buf.markpos
+            datasize = buffersize(buf) + buf.bufferpos - buf.markpos
+        end
+        copy!(buf.data, 1, buf.data, datapos, datasize)
+        buf.bufferpos -= datapos - 1
+        buf.marginpos = buf.bufferpos + datasize
+    end
+    if marginsize(buf) < minsize
+        # expand data buffer
+        resize!(buf.data, buf.marginpos + minsize - 1)
+    end
+    @assert marginsize(buf) ≥ minsize
+    return marginsize(buf)
+end
+
+function emptybuffer!(buf::Buffer)
+    buf.marginpos = buf.bufferpos
+    return buf
+end
+
+function skipbuffer!(buf::Buffer, n::Integer)
+    if n > buffersize(buf)
+        throw(ArgumentError("too large skip size"))
+    end
+    buf.bufferpos += n
+    return buf
+end
+
+# Discard buffered data and initialize positions.
+function initbuffer!(buf::Buffer)
+    buf.markpos = 0
+    buf.bufferpos = buf.marginpos = 1
+    return buf
+end
+
+# Read as much data as possbile from `input` to the margin of `output`.
+# This function will not block if `input` has buffered data.
+function readdata!(input::IO, output::Buffer)
+    nread::Int = 0
+    navail = nb_available(input)
+    if navail == 0 && marginsize(output) > 0 && !eof(input)
+        nread += writebyte!(output, read(input, UInt8))
+        navail = nb_available(input)
+    end
+    n = min(navail, marginsize(output))
+    Base.unsafe_read(input, marginptr(output), n)
+    output.marginpos += n
+    nread += n
+    return nread
+end
+
+function readdata!(buf::Buffer, dst::Vector{UInt8}, dpos::Integer, sz::Integer)
+    copy!(dst, dpos, buf.data, buf.bufferpos, sz)
+    buf.bufferpos += sz
+    return dst
+end
+
+# Write as much data as possible to `output` from the buffer of `input`.
+function writebuffer!(output::IO, input::Buffer)
+    while buffersize(input) > 0
+        input.bufferpos += Base.unsafe_write(output, bufferptr(input), buffersize(input))
+    end
+end
+
+function findbyte(buf, byte::UInt8)
+    ptr = ccall(:memchr, Ptr{Void}, (Ptr{Void}, Cint, Csize_t), pointer(buf.data, buf.bufferpos), byte, buffersize(buf))
+    if ptr == C_NULL
+        return 0
+    else
+        return Int(ptr - pointer(buf.data, buf.bufferpos)) + buf.bufferpos
+    end
+end

src/codec.jl

@@ -1,124 +1,50 @@
 # Codec Interfaces
 # ================
 
-abstract type Codec end
-
-abstract type Mode end
-struct Read  <: Mode end
-struct Write <: Mode end
-
-primitive type ProcCode 8 end
-const PROC_INIT   = reinterpret(ProcCode, 0x00)
-const PROC_OK     = reinterpret(ProcCode, 0x01)
-const PROC_FINISH = reinterpret(ProcCode, 0x02)
-
-
-# Start
-# -----
-
 """
-    start(::Type{Read}, codec::Codec, source::IO)::Void
-
-Start transcoding using `codec` with read mode.
+An abstract codec type.
 
-This method will be called only once before calling `process` first time.
-
-Implementing this method is optional; there is a default method that does nothing.
+Any codec supporting transcoding interfaces must be a subtype of this type.
 """
-function start(::Type{Read}, ::Codec, ::IO)
-    return nothing
-end
-
-"""
-    start(::Type{Write}, codec::Codec, sink::IO)::Void
-
-Start transcoding using `codec` with write mode.
-
-This method will be called only once before calling `process` first time.
-
-Implementing this method is optional; there is a default method that does nothing.
-"""
-function start(::Type{Write}, ::Codec, ::IO)
-    return nothing
-end
+abstract type Codec end
 
 
-# Process
+# Methods
 # -------
 
 """
-    process(::Type{Read}, codec::Codec, source::IO, output::Ptr{UInt8}, nbytes::Int)::Tuple{Int,ProcCode}
-
-Transcode data using `codec` with read mode.
-
-      <>    >-----(process)----->  [........]
-    source          codec            output
-
-This method reads some data from `source` and write the transcoded data to
-`output` at most `nbytes`, and then returns the number of written bytes and an
-appropriate return code.  It will be called repeatedly to incrementally
-transcode input data from `source`. It can assume `output` points to a valid
-memory position and `nbytes` are positive.  The intermediate result of
-transcoding is indicated by the return value of it.  If processing works
-properly and there are still data to write, it must return the number of written
-bytes and `PROC_OK`.  If processing finishes and there are no remaining data to
-write, it must return the number of written bytes and `PROC_FINISH`. Therefore,
-after finishing reading data from `source` and all (possibly buffered) data are
-written to `output`, it is expected to return `(0, PROC_FINISH)` forever.
+    initialize(codec::Codec)::Void
 
-This method may throw an exception when transcoding fails. For example, when
-`codec` is a decompression algorithm, it may throw an exception if it reads
-malfomed data from `source`.  However, this method must not throw `EOFError`
-when reading data from `source`.  Also note that it is responsible for this
-method to release resources allocated by `codec` when an exception happens.
+Initialize `codec`.
 """
-function process(::Type{Read}, codec::Codec, source::IO, input::Ptr{UInt8}, nbytes::Int)
-    error("codec $(typeof(codec)) does not implement read mode")
+function initialize(codec::Codec)
+    return nothing
 end
 
 """
-    process(::Type{Write}, codec::Codec, sink::IO, input::Ptr{UInt8}, nbytes::Int)::Tuple{Int,ProcCode}
-
-Transcode data using `codec` with write mode.
+    finalize(codec::Codec)::Void
 
-      <>    <-----(transcode)-----<  [.......]
-     sink            codec             input
-
-This method reads some data from `input` and write the transcoded bytes to
-`sink` at most `nbytes`, and then returns the number of written bytes and an
-appropriate return code. It can assume `input` points to a valid memory position
-and `nbytes` is positive.
+Finalize `codec`.
 """
-function process(::Type{Write}, codec::Codec, sink::IO, input::Ptr{UInt8}, nbytes::Int)
-    error("codec $(typeof(codec)) does not implement write mode")
+function finalize(codec::Codec)::Void
+    return nothing
 end
 
-
-# Finish
-# ------
-
 """
-    finish(::Type{Read}, codec::Codec, source::IO)::Void
-
-Finish transcoding using `codec` with read mode.
-
-This method will be called only once after calling `process` last time.
+    startproc(codec::Codec, state::Symbol)::Symbol
 
-Implementing this method is optional; there is a default method that does nothing.
+Start data processing with `codec` of `state`.
 """
-function finish(::Type{Read}, ::Codec, ::IO)
-    return nothing
+function startproc(codec::Codec, state::Symbol)::Symbol
+    return :ok
 end
 
 """
-    finish(::Type{Write}, codec::Codec, sink::IO)::Void
-
-Finish transcoding using `codec` with write mode.
+    process(codec::Codec, input::Memory, output::Memory)::Tuple{Int,Int,Symbol}
 
-This method will be called only once after calling `process` last time.
-
-Implementing this method is optional; there is a default method that does nothing.
+Do data processing with `codec`.
 """
-function finish(::Type{Write}, ::Codec, ::IO)
-    return nothing
+function process(codec::Codec, input::Memory, output::Memory)::Tuple{Int,Int,Symbol}
+    # no default method
+    throw(MethodError(process, (codec, input, output)))
 end

src/identity.jl

@@ -3,16 +3,8 @@
 
 struct Identity <: Codec end
 
-function process(::Type{Read}, ::Identity, stream::IO, data::Ptr{UInt8}, nbytes::Int)::Tuple{Int,ProcCode}
-    n = unsafe_read(stream, data, nbytes)
-    if eof(stream)
-        return n, PROC_FINISH
-    else
-        return n, PROC_OK
-    end
-end
-
-function process(::Type{Write}, ::Identity, stream::IO, data::Ptr{UInt8}, nbytes::Int)::Tuple{Int,ProcCode}
-    n = unsafe_write(stream, data, nbytes)
-    return Int(n), PROC_OK
+function process(::Identity, input::Memory, output::Memory)
+    n = Int(min(input.size, output.size))
+    unsafe_copy!(output.ptr, input.ptr, n)
+    return n, n, ifelse(input.size == 0, :end, :ok)
 end

src/io.jl

@@ -27,3 +27,15 @@ function unsafe_read(input::IO, output::Ptr{UInt8}, nbytes::Int)::Int
     nread += n
     return nread
 end
+
+function unsafe_read(input::IO, mem::Memory)
+    return unsafe_read(input, mem.ptr, Int(mem.size))
+end
+
+function unsafe_write(output::IO, ptr::Ptr{UInt8}, nbytes::Int)::Int
+    return Base.unsafe_write(output, ptr, nbytes)
+end
+
+function unsafe_write(output::IO, mem::Memory)
+    return Base.unsafe_write(output, mem.ptr, mem.size)
+end

src/memory.jl

@@ -0,0 +1,11 @@
+# Memory
+# ======
+
+immutable Memory
+    ptr::Ptr{UInt8}
+    size::UInt
+end
+
+function Base.length(mem::Memory)
+    return mem.size
+end