Start cooperative scheduler;

examples/delay.lua

@@ -0,0 +1,31 @@
+local rx = require 'rx'
+
+local timerResolution = .25
+local function log(message)
+  print('[' .. string.format('%.2f', rx.scheduler.currentTime) .. '] ' .. message)
+end
+
+rx.scheduler:schedule(function()
+  log('this is like a setTimeout')
+end, 2)
+
+rx.scheduler:schedule(function()
+  local i = 1
+  while true do
+    log('this prints i twice per second: ' .. i)
+    i = i + 1
+    coroutine.yield(.5)
+  end
+end)
+
+rx.scheduler:schedule(function()
+  for i = 1, 3 do
+    log('this will print for 3 updates after 1 second')
+    coroutine.yield()
+  end
+end, 1)
+
+repeat
+  rx.scheduler:update(timerResolution)
+  os.execute('sleep ' .. timerResolution)
+until rx.scheduler.currentTime >= 3

rx.lua

@@ -250,7 +250,55 @@ function Observable:combineLatest(...)
   end)
 end
 
+----
+-- Scheduler
+-- Schedulers manage groups of Observables.
+local Scheduler = {}
+
+----
+-- Cooperative Scheduler
+-- Manages Observables using coroutines and a virtual clock that must be updated manually.
+local Cooperative = {}
+Cooperative.__index = Cooperative
+
+--
+Cooperative.create = function(currentTime)
+  local self = {
+    tasks = {},
+    currentTime = currentTime or 0
+  }
+
+  return setmetatable(self, Cooperative)
+end
+
+--
+function Cooperative:schedule(action, delay)
+  table.insert(self.tasks, {
+    thread = coroutine.create(action),
+    due = self.currentTime + (delay or 0)
+  })
+end
+
+--
+function Cooperative:update(dt)
+  self.currentTime = self.currentTime + dt
+  for i = #self.tasks, 1, -1 do
+    local task = self.tasks[i]
+    if self.currentTime >= task.due then
+      local _, delay = coroutine.resume(task.thread)
+      task.due = math.max(task.due + (delay or 0), self.currentTime)
+      if coroutine.status(task.thread) == 'dead' then
+        table.remove(self.tasks, i)
+      end
+    end
+  end
+end
+
+Scheduler.Cooperative = Cooperative
+
 return {
   Observer = Observer,
-  Observable = Observable
+  Observable = Observable,
+  Scheduler = Scheduler,
+  scheduler = Scheduler.Cooperative.create()
 }