Reading JSON objects from streaming buffer

[tk3369]

Hi, I'm just looking for suggestions. I want to read JSON objects from an IO in a streaming fashion e.g. taking 1024 bytes at a time. That means I have 3 possible situations when parsing JSON objects from the data buffer:

1. The buffer contains exactly one JSON object
2. The buffer contains more than one JSON object
3. The buffer contains a partial JSON object

When I use JSON3.read, it returns the first JSON object from the string. For case 2, I want to read the first object and shift the pointer so that I can call JSON3.read again to get the next object. l will repeat until I hit case 3 and by then I will trigger a read from the IO.

julia> s = """{"a":1,"b":2}{"a":3,"b":4}{"""
"{\"a\":1,\"b\":2}{\"a\":3,\"b\":4}{"

julia> JSON3.read(s)
JSON3.Object{Base.CodeUnits{UInt8, String}, Vector{UInt64}} with 2 entries:
  :a => 1
  :b => 2

julia> length("""{"a":1,"b":2}""")
13

So, I think I just need a way to get the offset after reading an object (in the above example, the value is 14). Any suggestion?

[quinnj]

I have this pattern that I use when reading from a non-exhausting IO stream:

# Threads.@spawn producer(io, n, outgoing_msgs, incoming_msgs, buflen)
function producer(io::IO, n::Int, outgoing_msgs::Channel{Message}=Channel{Message}(n), incoming_msgs::Channel{Message}=Channel{Message}(n), buflen::Int=1024)
    # Create a buffer to hold the incoming messages
    max_msg_len = 256
    buf = Vector{UInt8}(undef, buflen)
    safelen = buflen - max_msg_len
    pos = writepos = 1
    while isopen(outgoing_msgs) && isopen(incoming_msgs)
        # reads data from IO until buf is full
        read!(io, view(buf, writepos:len))
        while pos < safelen
            msg = take!(outgoing_msgs)
            StructUtils.reset!(msg) # reset the message for reuse
            pos += JSON.parse!(view(buf, pos:buflen), msg)
            put!(incoming_msgs, msg)
        end
        nbytes_remaining = len - pos + 1
        unsafe_copyto!(buf, 1, buf, pos, nbytes_remaining)
        writepos = nbytes_remaining + 1
        pos = 1
    end
end

# Threads.@spawn consumer(f, outgoing_msgs, incoming_msgs)
function consumer(f, outgoing_msgs::Channel{Message}, incoming_msgs::Channel{Message})
    while isopen(outgoing_msgs) && isopen(incoming_msgs)
        msg = take!(incoming_msgs)
        f(msg)
        put!(outgoing_msgs, msg)
    end
end

The highlights include:

• Using 2 channels to pass parsed Message instances between producer and consumer
• Producer fills a buffer, parses messages and sends them to consumer via incoming_msgs channel
• Once the producer has read messages up to the max length of a message, it shifts the remaining bytes down and refills the buffer
• Takes advantage of pre-allocated mutable Message objects and JSON.parse! which mutates the msg object while parsing and returns the # of bytes parsed from buf
• Consumer is pretty simple: takes from incoming, calls f(msg) and then sends the message back to producer via outgoing_msgs channel