pipeutils.lua

-- 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,
}