lib/rainbows/fiber/queue.rb

   1 # -*- encoding: binary -*-
   2 # :enddoc:
   3 #
   4 # a self-sufficient Queue implementation for Fiber-based concurrency
   5 # models.  This requires no external scheduler, so it may be used with
   6 # Revactor as well as FiberSpawn and FiberPool.
   7 class Rainbows::Fiber::Queue < Struct.new(:queue, :waiters)
   8   def initialize(queue = [], waiters = [])
   9     # move elements of the Queue into an Array
  10     if queue.class.name == "Queue"
  11       queue = queue.length.times.map { queue.pop }
  12     end
  13     super queue, waiters
  14   end
  15
  16   def shift
  17     # ah the joys of not having to deal with race conditions
  18     if queue.empty?
  19       waiters << Fiber.current
  20       Fiber.yield
  21     end
  22     queue.shift
  23   end
  24
  25   def <<(obj)
  26     queue << obj
  27     blocked = waiters.shift and blocked.resume
  28     queue # not quite 100% compatible but no-one's looking :>
  29   end
  30 end