Class: Fluent::EventLastValueOutput

Inherits:
BufferedOutput
  • Object
show all
Defined in:
lib/fluent/plugin/out_eventlastvalue.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ EventLastValueOutput 



3
4
5
 
# File 'lib/fluent/plugin/out_eventlastvalue.rb', line 3

def initialize
  super
end

Instance Attribute Details

#last_values ⇒ Object

Returns the value of attribute last_values.



12
13
14
 
# File 'lib/fluent/plugin/out_eventlastvalue.rb', line 12

def last_values
  @last_values
end

Instance Method Details

#configure(conf) ⇒ Object 



14
15
16
 
# File 'lib/fluent/plugin/out_eventlastvalue.rb', line 14

def configure(conf)
  super
end

#format(tag, time, record) ⇒ Object 



22
23
24
25
 
# File 'lib/fluent/plugin/out_eventlastvalue.rb', line 22

def format(tag, time, record)
  return '' unless @last_value_key.nil? || record[@last_value_key]
  [record[@id_key], record, (record[@comparator_key] || 0).to_f].to_json + "\n"
end

#start ⇒ Object 



18
19
20
 
# File 'lib/fluent/plugin/out_eventlastvalue.rb', line 18

def start
  super
end

#write(chunk) ⇒ Object 



27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/fluent/plugin/out_eventlastvalue.rb', line 27

def write(chunk)
  last_values = Hash.new {|hash, key| hash[key] = Hash.new {|h,k| h[k] = nil } }
  last_times = Hash.new {|hash, key| hash[key] = 0}
  chunk.open do |io|
    items = io.read.split("\n")
    items.each do |item|
      key, event, t = JSON.parse(item)
      if @comparator_key.nil? || last_times[key] < t
        last_times[key] = t
        last_values[key] = event
      end
    end
  end

  last_values.each do |key, last_record|
    Fluent::Engine.emit(@emit_to, Time.now.to_i, last_record)
  end
end