Sha256: 548ffd76e40efe6345e163850419bfbb57fe4e03afbd1ac0ff8474ea622264a2
Contents?: true
Size: 1.26 KB
Versions: 11
Compression:
Stored size: 1.26 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
def asynchronic?
true
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