Class: Fusuma::Runner
- Inherits:
-
Object
- Object
- Fusuma::Runner
- Defined in:
- lib/fusuma.rb
Overview
main class
Class Method Summary collapse
Instance Method Summary collapse
- #buffer(event) ⇒ Array<Plugin::Buffers::Buffer>, NilClass
- #clear_expired_events ⇒ Object
- #detect(buffers) ⇒ Array<Event>
- #execute(condition, context, event) ⇒ Object
- #filter(event) ⇒ Plugin::Events::Event
-
#initialize ⇒ Runner
constructor
A new instance of Runner.
- #input ⇒ Plugin::Events::Event
- #merge(events) ⇒ Plugin::Events::Event, NilClass
- #parse(event) ⇒ Plugin::Events::Event
- #pipeline ⇒ Object
- #run ⇒ Object
-
#run_with_lineprof(count: 1000) ⇒ Object
For performance monitoring.
Constructor Details
#initialize ⇒ Runner
Returns a new instance of Runner.
58 59 60 61 62 63 64 65 |
# File 'lib/fusuma.rb', line 58 def initialize @inputs = Plugin::Inputs::Input.plugins.map(&:new) @filters = Plugin::Filters::Filter.plugins.map(&:new) @parsers = Plugin::Parsers::Parser.plugins.map(&:new) @buffers = Plugin::Buffers::Buffer.plugins.map(&:new) @detectors = Plugin::Detectors::Detector.plugins.map(&:new) @executors = Plugin::Executors::Executor.plugins.map(&:new) end |
Class Method Details
.run(option = {}) ⇒ Object
15 16 17 18 19 20 21 22 |
# File 'lib/fusuma.rb', line 15 def run(option = {}) set_trap (option) instance = new ## NOTE: Uncomment following line to measure performance # instance.run_with_lineprof instance.run end |
Instance Method Details
#buffer(event) ⇒ Array<Plugin::Buffers::Buffer>, NilClass
114 115 116 |
# File 'lib/fusuma.rb', line 114 def buffer(event) @buffers.select { |b| b.buffer(event) } end |
#clear_expired_events ⇒ Object
191 192 193 |
# File 'lib/fusuma.rb', line 191 def clear_expired_events @buffers.each(&:clear_expired) end |
#detect(buffers) ⇒ Array<Event>
120 121 122 123 124 125 126 127 128 129 130 131 132 133 |
# File 'lib/fusuma.rb', line 120 def detect(buffers) matched_detectors = @detectors.select do |detector| detector.watch? || buffers.any? { |b| detector.sources.include?(b.type) } end events = matched_detectors.each_with_object([]) do |detector, detected| Array(detector.detect(@buffers)).each { |e| detected << e } end return if events.empty? events end |
#execute(condition, context, event) ⇒ Object
169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 |
# File 'lib/fusuma.rb', line 169 def execute(condition, context, event) return unless event # Find executable condition and executor executor = Config::Searcher.with_context(context) do Config::Searcher.with_condition(condition) do @executors.find { |e| e.executable?(event) } end end return if executor.nil? # Check interval and execute Config::Searcher.with_context(context) do Config::Searcher.with_condition(condition) do executor.enough_interval?(event) && executor.update_interval(event) && executor.execute(event) end end end |
#filter(event) ⇒ Plugin::Events::Event
101 102 103 |
# File 'lib/fusuma.rb', line 101 def filter(event) event if @filters.any? { |f| f.filter(event) } end |
#input ⇒ Plugin::Events::Event
95 96 97 |
# File 'lib/fusuma.rb', line 95 def input Plugin::Inputs::Input.select(@inputs) end |
#merge(events) ⇒ Plugin::Events::Event, NilClass
138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 |
# File 'lib/fusuma.rb', line 138 def merge(events) index_events, context_events = events.partition { |event| event.record.type == :index } main_events, modifiers = index_events.partition { |event| event.record.mergable? } request_context = context_events.each_with_object({}) do |e, results| results[e.record.name] = e.record.value end main_events.sort_by! { |e| e.record.trigger_priority } matched_condition = nil matched_context = nil event = main_events.find do |main_event| matched_context = Config::Searcher.find_context(request_context) do matched_condition, modified_record = Config::Searcher.find_condition do main_event.record.merge(records: modifiers.map(&:record)) end if matched_condition && modified_record main_event.record = modified_record else matched_condition, = Config::Searcher.find_condition do Config.search(main_event.record.index) && Config.find_execute_key(main_event.record.index) end end end end return if event.nil? [matched_condition, matched_context, event] end |
#parse(event) ⇒ Plugin::Events::Event
107 108 109 |
# File 'lib/fusuma.rb', line 107 def parse(event) @parsers.reduce(event) { |e, p| p.parse(e) if e } end |
#pipeline ⇒ Object
71 72 73 74 75 76 77 78 79 80 |
# File 'lib/fusuma.rb', line 71 def pipeline event = input || return clear_expired_events filtered = filter(event) || return parsed = parse(filtered) || return buffered = buffer(parsed) || return detected = detect(buffered) || return condition, context, event = merge(detected) || return execute(condition, context, event) end |
#run ⇒ Object
67 68 69 |
# File 'lib/fusuma.rb', line 67 def run loop { pipeline } end |
#run_with_lineprof(count: 1000) ⇒ Object
For performance monitoring
83 84 85 86 87 88 89 90 91 92 |
# File 'lib/fusuma.rb', line 83 def run_with_lineprof(count: 1000) require 'rblineprof' require 'rblineprof-report' profile = lineprof(%r{#{Pathname.new(__FILE__).parent}/.}) do count.times { pipeline } end LineProf.report(profile) exit 0 end |