diff options
Diffstat (limited to 'plugins/coro-channel/init.lua')
-rw-r--r-- | plugins/coro-channel/init.lua | 128 |
1 files changed, 128 insertions, 0 deletions
diff --git a/plugins/coro-channel/init.lua b/plugins/coro-channel/init.lua new file mode 100644 index 00000000000..0b715a0d470 --- /dev/null +++ b/plugins/coro-channel/init.lua @@ -0,0 +1,128 @@ +local exports = {} +exports.name = "creationix/coro-channel" +exports.version = "1.2.0" +exports.homepage = "https://github.com/luvit/lit/blob/master/deps/coro-channel.lua" +exports.description = "An adapter for wrapping uv streams as coro-streams and chaining filters." +exports.tags = {"coro", "adapter"} +exports.license = "MIT" +exports.author = { name = "Tim Caswell" } + +local function wrapRead(socket) + local paused = true + local queue = {} + local waiting + local onRead + + function onRead(err, chunk) + local data = err and {nil, err} or {chunk} + if waiting then + local thread = waiting + waiting = nil + assert(coroutine.resume(thread, unpack(data))) + else + queue[#queue + 1] = data + if not paused then + paused = true + assert(socket:read_stop()) + end + end + end + + return function () + if #queue > 0 then + return unpack(table.remove(queue, 1)) + end + if paused then + paused = false + assert(socket:read_start(onRead)) + end + waiting = coroutine.running() + return coroutine.yield() + end + +end + +local function wrapWrite(socket) + + local function wait() + local thread = coroutine.running() + return function (err) + assert(coroutine.resume(thread, err)) + end + end + + local function shutdown() + socket:shutdown(wait()) + coroutine.yield() + if not socket:is_closing() then + socket:close() + end + end + + return function (chunk) + if chunk == nil then + return shutdown() + end + assert(socket:write(chunk, wait())) + local err = coroutine.yield() + return not err, err + end + +end + +exports.wrapRead = wrapRead +exports.wrapWrite = wrapWrite + +-- Given a raw uv_stream_t userdata, return coro-friendly read/write functions. +function exports.wrapStream(socket) + return wrapRead(socket), wrapWrite(socket) +end + + +function exports.chain(...) + local args = {...} + local nargs = select("#", ...) + return function (read, write) + local threads = {} -- coroutine thread for each item + local waiting = {} -- flag when waiting to pull from upstream + local boxes = {} -- storage when waiting to write to downstream + for i = 1, nargs do + threads[i] = coroutine.create(args[i]) + waiting[i] = false + local r, w + if i == 1 then + r = read + else + function r() + local j = i - 1 + if boxes[j] then + local data = boxes[j] + boxes[j] = nil + assert(coroutine.resume(threads[j])) + return unpack(data) + else + waiting[i] = true + return coroutine.yield() + end + end + end + if i == nargs then + w = write + else + function w(...) + local j = i + 1 + if waiting[j] then + waiting[j] = false + assert(coroutine.resume(threads[j], ...)) + else + boxes[i] = {...} + coroutine.yield() + end + end + end + assert(coroutine.resume(threads[i], r, w)) + end + end +end + +return exports |