lib/rainbows/fiber/queue.rb

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