Class: Fluent::Test::Driver::Filter

Inherits:
BaseOwner show all
Includes:
EventFeeder
Defined in:
lib/fluent/test/driver/filter.rb

Constant Summary

Constants inherited from Base

Base::DEFAULT_TIMEOUT

Instance Attribute Summary collapse

Attributes inherited from Base

#instance, #logs

Instance Method Summary collapse

Methods included from EventFeeder

#feed, #feed_to_plugin, #run

Methods inherited from BaseOwner

#configure, #emit_count, #emit_error_event, #emit_event_stream, #error_events, #event_streams, #events, #record_count, #run

Methods inherited from Base

#break_if, #broken?, #configure, #end_if, #instance_hook_before_stopped, #instance_shutdown, #instance_start, #run, #run_actual, #stop?

Constructor Details

#initialize(klass, opts: {}, &block) ⇒ Filter

Returns a new instance of Filter.

Raises:

  • (ArgumentError) 


30
31
32
33
34
 
# File 'lib/fluent/test/driver/filter.rb', line 30

def initialize(klass, opts: {}, &block)
  super
  raise ArgumentError, "plugin is not an instance of Fluent::Plugin::Filter" unless @instance.is_a? Fluent::Plugin::Filter
  @filtered = []
end

Instance Attribute Details

#filtered ⇒ Object (readonly)

Returns the value of attribute filtered.



28
29
30
 
# File 'lib/fluent/test/driver/filter.rb', line 28

def filtered
  @filtered
end

Instance Method Details

#filtered_records ⇒ Object 



36
37
38
 
# File 'lib/fluent/test/driver/filter.rb', line 36

def filtered_records
  @filtered.map {|_time, record| record }
end

#filtered_time ⇒ Object 



40
41
42
 
# File 'lib/fluent/test/driver/filter.rb', line 40

def filtered_time
  @filtered.map {|time, _record| time }
end

#instance_hook_after_started ⇒ Object 



44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/fluent/test/driver/filter.rb', line 44

def instance_hook_after_started
  super
  filter_hook = ->(time, record) { @filtered << [time, record] }
  m = Module.new do
    define_method(:filter_stream) do |tag, es|
      new_es = super(tag, es)
      new_es.each do |time, record|
        filter_hook.call(time, record)
      end
      new_es
    end
  end
  @instance.singleton_class.prepend(m)
end