Sha256: 07a645261544db917dea55fc13814bc5427e8e6d0ed669adb150d058fb17f9ec
Contents?: true
Size: 1.22 KB
Versions: 11
Compression:
Stored size: 1.22 KB
Contents
module Asynchronic
module QueueEngine
class InMemory
attr_reader :default_queue
def initialize(options={})
@default_queue = options[:default_queue]
@queues ||= Hash.new { |h,k| h[k] = Queue.new }
end
def default_queue
@default_queue ||= Asynchronic.default_queue
end
def [](name)
@queues[name]
end
def queues
@queues.keys.map(&:to_sym)
end
def clear
@queues.clear
end
def listener
Listener.new
end
class Queue
extend Forwardable
def_delegators :@queue, :size, :empty?, :to_a
def initialize
@queue = []
@mutex = Mutex.new
end
def pop
@mutex.synchronize { @queue.shift }
end
def push(message)
@mutex.synchronize { @queue.push message }
end
end
class Listener
def listen(queue, &block)
@stopping = false
loop do
break if @stopping
item = queue.pop
next unless item
block.call item
end
end
def stop
@stopping = true
end
end
end
end
end
Version data entries
11 entries across 11 versions & 1 rubygems