summaryrefslogtreecommitdiffstatshomepage
path: root/plugins/coro-channel/init.lua
blob: 0b715a0d470ce0c500c9dd4c3d4b3751fc98f7c4 (plain) (blame)
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