123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354 |
- -- A collection of pipe-based utility functions
- -- A convenience wrapper for chunking data arriving in bursts into more sizable
- -- blocks; `o` will be called once per chunk. The `flush` method can be used to
- -- drain the internal buffer. `flush` MUST be called at the end of the stream,
- -- **even if the stream is a multiple of the chunk size** due to internal
- -- buffering. Flushing results in smaller chunk(s) being output, of course.
- local function chunker(o, csize, prio)
- assert (type(o) == "function" and type(csize) == "number" and 1 <= csize)
- local p = pipe.create(function(p)
- -- wait until it looks very likely that read is going to succeed
- -- and we won't have to unread. This may hold slightly more than
- -- a chunk in the underlying pipe object.
- if 256 * (p:nrec() - 1) <= csize then return nil end
- local d = p:read(csize)
- if #d < csize
- then p:unread(d) return false
- else o(d) return true
- end
- end, prio or node.task.LOW_PRIORITY)
- return {
- flush = function() for d in p:reader(csize) do o(d) end end,
- write = function(d) p:write(d) end
- }
- end
- -- Stream and decode lines of complete base64 blocks, calling `o(data)` with
- -- decoded chunks or calling `e(badinput, errorstr)` on error; the error
- -- callback must ensure that this conduit is never written to again.
- local function debase64(o, e, prio)
- assert (type(o) == "function" and type(e) == "function")
- local p = pipe.create(function(p)
- local s = p:read("\n+")
- if s:sub(-1) == "\n" then -- guard against incomplete line
- s = s:match("^%s*(%S*)%s*$")
- if #s ~= 0 then -- guard against empty line
- local ok, d = pcall(encoder.fromBase64, s)
- if ok then o(d) else e(s, d); return false end
- end
- return true
- else
- p:unread(s)
- return false
- end
- end, prio or node.task.LOW_PRIORITY)
- return { write = function(d) p:write(d) end }
- end
- return {
- chunker = chunker,
- debase64 = debase64,
- }
|