Class: LogStash::Filters::Empow::PluginLogic

Inherits:
Object
  • Object
show all
Includes:
Util::Loggable
Defined in:
lib/logstash/filters/plugin-logic.rb

Instance Method Summary collapse

Constructor Details

#initialize(classifer, field_handler, max_parking_time, max_parked_events, tag_on_timeout, tag_on_error) ⇒ PluginLogic

Returns a new instance of PluginLogic.



12
13
14
15
16
17
18
19
20
21
22
23
24
25
 
# File 'lib/logstash/filters/plugin-logic.rb', line 12

def initialize(classifer, field_handler, max_parking_time, max_parked_events, tag_on_timeout, tag_on_error)
	@logger ||= self.logger
	#@logger.info("initializing classifier")

	@field_handler = field_handler

	@max_parking_time = max_parking_time
	@max_parked_events = max_parked_events
	@tag_on_timeout = tag_on_timeout
	@tag_on_error = tag_on_error

	@classifer = classifer
	@parked_events = Concurrent::Array.new
end

Instance Method Details

#classify(event) ⇒ Object 



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/logstash/filters/plugin-logic.rb', line 31

def classify(event)
	request = @field_handler.event_to_classification_request(event)

	if request.nil?
		@tag_on_error.each{|tag| event.tag(tag)}
		return event
	end

	if classify_event(request, event)
		return event
	else
		park(event)

		if @parked_events.length > @max_parked_events
			tuple = @parked_events.shift
			
			if !tuple.nil?
				unparked_event = tuple[:event]
				unparked_event.uncancel
				return unparked_event
			end
		end

		return nil
	end
end

#close ⇒ Object 



27
28
29
 
# File 'lib/logstash/filters/plugin-logic.rb', line 27

def close
	@classifer.close
end

#flush(options = {}) ⇒ Object 



58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/logstash/filters/plugin-logic.rb', line 58

def flush(options = {})
	# tag flushed events, 
	events_to_flush = []

    		if options[:final] # indicating "final flush" special event, flush everything
    			while tuple = @parked_events.shift do
    				events_to_flush << tuple[:event]
    			end
    		else
    			@parked_events.delete_if do |tuple|
    				process_parked_event(tuple, events_to_flush)
    			end
    		end

    		return events_to_flush
end