Class: Fluent::RouteOutput
- Inherits:
-
MultiOutput
- Object
- MultiOutput
- Fluent::RouteOutput
- Defined in:
- lib/fluent/plugin/out_route.rb
Defined Under Namespace
Classes: Route
Instance Attribute Summary collapse
-
#routes ⇒ Object
readonly
TODO tag_transform regexp.
Instance Method Summary collapse
- #configure(conf) ⇒ Object
- #emit(tag, es, chain) ⇒ Object
-
#initialize ⇒ RouteOutput
constructor
A new instance of RouteOutput.
Constructor Details
#initialize ⇒ RouteOutput
Returns a new instance of RouteOutput.
85 86 87 88 89 90 |
# File 'lib/fluent/plugin/out_route.rb', line 85 def initialize super @routes = [] @tag_cache = {} @match_cache = {} end |
Instance Attribute Details
#routes ⇒ Object (readonly)
TODO tag_transform regexp
96 97 98 |
# File 'lib/fluent/plugin/out_route.rb', line 96 def routes @routes end |
Instance Method Details
#configure(conf) ⇒ Object
108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 |
# File 'lib/fluent/plugin/out_route.rb', line 108 def configure(conf) super if @remove_tag_prefix @prefix_match = /^#{Regexp.escape(@remove_tag_prefix)}\.?/ else @prefix_match = // end if @add_tag_prefix @tag_prefix = "#{@add_tag_prefix}." else @tag_prefix = "" end conf.elements.select {|e| e.name == 'route' }.each {|e| route = Route.new(e.arg, router) route.configure(e) @routes << route } end |
#emit(tag, es, chain) ⇒ Object
131 132 133 134 135 136 137 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/fluent/plugin/out_route.rb', line 131 def emit(tag, es, chain) ntag, targets = @match_cache[tag] unless targets ntag = tag.sub(@prefix_match, @tag_prefix) targets = [] @routes.each {|r| if r.match?(ntag) targets << r break unless r.copy? end } if @match_cache.size < 1024 # TODO size limit @match_cache[tag] = [ntag, targets] end end case targets.size when 0 return when 1 targets.first.emit(ntag, es) chain.next else unless es.repeatable? m = MultiEventStream.new es.each {|time,record| m.add(time, record) } es = m end targets.each {|t| t.emit(ntag, es) } chain.next end end |