Sha256: 46914b582a012374e49f4e50c8133c7a2426d5deacdc1c5d0af28014deb7e345
Contents?: true
Size: 1.65 KB
Versions: 13
Compression:
Stored size: 1.65 KB
Contents
require 'set'
module Norikra
class OutputPool
attr_accessor :pool
def initialize
@pool = {} # { query_name => [events] }
@groups = {} # { group_name => Set(query_names) }
@mutex = Mutex.new
end
def queries
@pool.keys
end
def push(query_name, query_group, events) # events must be [time(int), event_record]
# called with blank events for window leavings (and/or other situations)
return if events.size < 1
@mutex.synchronize do
if @groups[query_group]
@groups[query_group].add(query_name) # Set is unique set of elements
else
@groups[query_group] ||= Set.new([query_name])
end
@pool[query_name] ||= []
@pool[query_name].push(events)
end
end
# returns [[time, event], ...], but not remove from pool
def fetch(query_name)
events = @mutex.synchronize do
@pool.fetch(query_name, [])
end
events.reduce(&:+) || []
end
# returns [[time(int from epoch), event], ...], event: hash
def pop(query_name)
events = @mutex.synchronize do
@pool.delete(query_name) || []
end
events.reduce(&:+) || []
end
# returns {query_name => [[time, event], ...]}
def sweep(group=nil)
return {} if @groups[group].nil?
ret = {}
sweep_pool = @mutex.synchronize do
sweeped = {}
@groups[group].each do |qname|
sweeped[qname] = @pool.delete(qname) if @pool[qname] && @pool[qname].size > 0
end
sweeped
end
sweep_pool.keys.each do |k|
ret[k] = sweep_pool[k].reduce(&:+) || []
end
ret
end
end
end
Version data entries
13 entries across 13 versions & 1 rubygems