pipeutils.lua 2.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354
  1. -- A collection of pipe-based utility functions
  2. -- A convenience wrapper for chunking data arriving in bursts into more sizable
  3. -- blocks; `o` will be called once per chunk. The `flush` method can be used to
  4. -- drain the internal buffer. `flush` MUST be called at the end of the stream,
  5. -- **even if the stream is a multiple of the chunk size** due to internal
  6. -- buffering. Flushing results in smaller chunk(s) being output, of course.
  7. local function chunker(o, csize, prio)
  8. assert (type(o) == "function" and type(csize) == "number" and 1 <= csize)
  9. local p = pipe.create(function(p)
  10. -- wait until it looks very likely that read is going to succeed
  11. -- and we won't have to unread. This may hold slightly more than
  12. -- a chunk in the underlying pipe object.
  13. if 256 * (p:nrec() - 1) <= csize then return nil end
  14. local d = p:read(csize)
  15. if #d < csize
  16. then p:unread(d) return false
  17. else o(d) return true
  18. end
  19. end, prio or node.task.LOW_PRIORITY)
  20. return {
  21. flush = function() for d in p:reader(csize) do o(d) end end,
  22. write = function(d) p:write(d) end
  23. }
  24. end
  25. -- Stream and decode lines of complete base64 blocks, calling `o(data)` with
  26. -- decoded chunks or calling `e(badinput, errorstr)` on error; the error
  27. -- callback must ensure that this conduit is never written to again.
  28. local function debase64(o, e, prio)
  29. assert (type(o) == "function" and type(e) == "function")
  30. local p = pipe.create(function(p)
  31. local s = p:read("\n+")
  32. if s:sub(-1) == "\n" then -- guard against incomplete line
  33. s = s:match("^%s*(%S*)%s*$")
  34. if #s ~= 0 then -- guard against empty line
  35. local ok, d = pcall(encoder.fromBase64, s)
  36. if ok then o(d) else e(s, d); return false end
  37. end
  38. return true
  39. else
  40. p:unread(s)
  41. return false
  42. end
  43. end, prio or node.task.LOW_PRIORITY)
  44. return { write = function(d) p:write(d) end }
  45. end
  46. return {
  47. chunker = chunker,
  48. debase64 = debase64,
  49. }