Class: LogStash::Codecs::RetriggerableTask

Inherits:
Object
  • Object
show all
Defined in:
lib/logstash/codecs/retriggerable_task.rb

Constant Summary collapse

SLEEP_FOR = 
0.25.freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(delay, listener) ⇒ RetriggerableTask

Returns a new instance of RetriggerableTask.



8
9
10
11
12
13
14
 
# File 'lib/logstash/codecs/retriggerable_task.rb', line 8

def initialize(delay, listener)
  @count = calculate_count(delay)
  @listener = listener
  @counter = Concurrent::AtomicFixnum.new(0 + @count)
  @stopped = Concurrent::AtomicBoolean.new(false)
  @semaphore = Concurrent::Semaphore.new(1)
end

Instance Attribute Details

#thread ⇒ Object (readonly)

Returns the value of attribute thread.



6
7
8
 
# File 'lib/logstash/codecs/retriggerable_task.rb', line 6

def thread
  @thread
end

Instance Method Details

#close ⇒ Object 



29
30
31
 
# File 'lib/logstash/codecs/retriggerable_task.rb', line 29

def close
  @stopped.make_true
end

#counter ⇒ Object 



33
34
35
 
# File 'lib/logstash/codecs/retriggerable_task.rb', line 33

def counter
  @counter.value
end

#executing? ⇒ Boolean

Returns:

  • (Boolean) 


37
38
39
 
# File 'lib/logstash/codecs/retriggerable_task.rb', line 37

def executing?
  running? && counter < 1
end

#pending? ⇒ Boolean

Returns:

  • (Boolean) 


41
42
43
 
# File 'lib/logstash/codecs/retriggerable_task.rb', line 41

def pending?
  running? && counter > 0
end

#retrigger ⇒ Object 



16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/logstash/codecs/retriggerable_task.rb', line 16

def retrigger
  return if stopped?
  if executing?
    @semaphore.acquire
  end

  if pending?
    reset_counter
  else
    start
  end
end