Ruby coroutines proof of concept

fiber.rb

require 'fiber'
require 'singleton'
require 'em-synchrony'
require 'eventmachine'

class Scheduler
  include Singleton

  def initialize
    @fibers = {}
  end

  def create_task(req_id)
    Fiber.new do
      @fibers[req_id] = Fiber.current
      yield Fiber.yield # call the callback block when resumed
    end.resume
  end

  def create_suspended_task(req_id)
    @fibers[req_id] = Fiber.current
    return Fiber.yield # return with synchronous style
  end

  def resume(req_id)
    fiber = @fibers[req_id]
    if fiber != nil and fiber.alive?
      fiber.resume({ 'ReqID': req_id })
    end
  end
end

class WebSocket
  def send(req_id)
    # Simulate server latency
    EM.add_timer(2) { Scheduler.instance.resume(req_id) }
  end
end

class BlinkTrade
  def initialize
    @transport = WebSocket.new
  end

  def send(msg, &block)

    @transport.send(msg[:ReqID])

    if block != nil
      # Create task with callback style
      Scheduler.instance.create_task(msg[:ReqID]) do |data|
        instance_exec(data, &block)
      end
    else
      # Create task with synchronous style
      return Scheduler.instance.create_suspended_task(msg[:ReqID])
    end
  end

  def connect
    # Create root fiber
    Fiber.new { yield }.resume
  end

  def heartbeat(&block)
    req_id = Random.new.rand(1e8).to_i
    msg = {
      'ReqID': req_id,
    }
    return self.send(msg, &block)
  end
end

EM.run {
  blinktrade = BlinkTrade.new
  puts 'Calling outside of root fiber'
  blinktrade.heartbeat { |latency| puts latency }

  blinktrade.connect do
    puts 'Calling with callback'
    blinktrade.heartbeat { |latency| puts latency }

    puts 'Calling first request'
    latency = blinktrade.heartbeat
    puts latency

    puts 'Calling second callback'
    blinktrade.heartbeat { |latency| puts latency }

    puts 'Calling second resquest'
    latency = blinktrade.heartbeat
    puts latency

    puts 'Calling third resquest'
    latency = blinktrade.heartbeat
    puts latency

    # Calling outside of root fiber
    # Calling with callback
    # Calling first request
    # (PAUSE)
    # {:ReqID=>42431788}
    # {:ReqID=>89334334}
    # {:ReqID=>99005836}
    # Calling second callback
    # Calling second resquest
    # (PAUSE)
    # {:ReqID=>15384646}
    # {:ReqID=>17572}
    # Calling third resquest
    # (PAUSE)
    # {:ReqID=>39415239}

    EM.stop
  end
}