1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
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
|