Control flow inversion (async push -> sync pull) via coroutines

control-inversion.lua

-- Coroutine wrappers
function expect (token)
  local this = coroutine.running()
  return coroutine.yield(token, this)
end

function start (stream)
  return coroutine.create(stream)
end

function resume (waiting, stream, ...)
  local status, tok, f = coroutine.resume(stream, ...)
  if tok and f then
    waiting[tok] = f
  end
end


-- Lifecycle management
function init (...)
  local args = {...}
  
  local waiting = {}
  for i,s in pairs(args) do
    resume(waiting, start(s))
  end

  return waiting
end

function run (waiting)
  local line = io.read()

  if waiting[line] then
    local f = waiting[line]
    waiting[line] = nil

    resume(waiting, f, line)
  end

  if next(waiting) then
    run(waiting)
  end
end


-- Example streams
function stream1 ()
  print("1: want foo")
  expect("foo")
  print("1: want bar")
  expect("bar")
  print("1: wheeeea!")
end

function stream2 ()
  print("2: want baza")
  expect("baza")
  print("2: want goo")
  expect("goo")
  print("2: woohoo!")
end


--
run(init(stream1, stream2))

Example run:

1: want foo
2: want baza
foo
1: want bar
baza
2: want goo
goo
2: woohoo!
bar
1: wheeeea!