Class: SoarAuditingProvider::AuditingWorker

Inherits:
SoarThreadWorker::ThreadWorker
  • Object
show all
Defined in:
lib/soar_auditing_provider/auditing_worker.rb

Instance Method Summary collapse

Constructor Details

#initialize ⇒ AuditingWorker

Returns a new instance of AuditingWorker.



5
6
7
8
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 5

def initialize
  @queue = Queue.new
  @start_mutex = Mutex.new
end

Instance Method Details

#configure(queue_worker_configuration:, auditor_audit_method:) ⇒ Object 



10
11
12
13
14
15
16
17
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 10

def configure(queue_worker_configuration: ,auditor_audit_method: )
  validate_configuration(queue_worker_configuration)
  @maximum_queue_size = queue_worker_configuration['queue_size'].to_i
  @initial_back_off_in_seconds = queue_worker_configuration['initial_back_off_in_seconds'].to_i
  @back_off_multiplier = queue_worker_configuration['back_off_multiplier'].to_i
  @maximum_back_off_attempts = queue_worker_configuration['back_off_attempts'].to_i
  @auditor_audit_method = auditor_audit_method
end

#enqueue(level, data) ⇒ Object 



19
20
21
22
23
24
25
26
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 19

def enqueue(level, data)
  if @queue.size < @maximum_queue_size then
    @queue.push({:level => level, :data => data})
  else
    raise AuditingOverflowError
  end
  ensure_worker_is_running
end

#execute ⇒ Object 



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 37

def execute
  audit_event = @queue.pop
  failed_before = false
  begin
    if @stopping
      @queue.push(audit_event) if audit_event #push the event back into the queue so that fallback flush mechanism can deal with this audit event
      return true #indicates to thread worder that we are done executing since we are in the process of stopping
    end
    exponential_back_off(start_at_last_attempt: failed_before) {
      @auditor_audit_method.call(audit_event[:level],audit_event[:data])
    }
  rescue Exception => e
    print_exception_with_message_to_stderr(nil,e)
    failed_before = true
    retry
  end
  return false #indicates to thread worder that we are not done executing
end

#flush(timeout = 1) ⇒ Object 



56
57
58
59
60
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 56

def flush(timeout = 1)
  ensure_worker_is_running
  wait_for_worker_to_clear_queue(timeout)
  fallback_flush_to_stderr if @queue.size > 0
end

#start(verbose: false) ⇒ Object 



28
29
30
31
32
33
34
35
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 28

def start(verbose: false)
  @start_mutex.synchronize {
    if not running? then
      super()
      $stderr.puts("Auditing worker was not running and respawned") if verbose
    end
  }
end