Class: GetaroundUtils::Utils::AsyncQueue
- Inherits:
-
Object
- Object
- GetaroundUtils::Utils::AsyncQueue
show all
- Extended by:
- Mixins::Loggable
- Defined in:
- lib/getaround_utils/utils/async_queue.rb
Constant Summary
collapse
- MAX_QUEUE_SIZE =
100
- MUTEX =
Mutex.new
Class Method Summary
collapse
base_append_infos_to_loggable, base_loggable_logger, class_name, loggable
Class Method Details
11
12
13
|
# File 'lib/getaround_utils/utils/async_queue.rb', line 11
def perform
raise NotImplementedError
end
|
15
16
17
18
19
20
21
22
23
24
|
# File 'lib/getaround_utils/utils/async_queue.rb', line 15
def perform_async(*args)
start_once!
if @queue.size > MAX_QUEUE_SIZE
loggable('warn', 'Queue is overflowing')
return
end
@queue.push(args)
end
|
.reset ⇒ Object
52
53
54
55
|
# File 'lib/getaround_utils/utils/async_queue.rb', line 52
def reset
terminate
@parent = nil
end
|
.start_once! ⇒ Object
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
|
# File 'lib/getaround_utils/utils/async_queue.rb', line 26
def start_once!
MUTEX.synchronize do
return unless @parent.nil?
@parent = Process.pid
@queue = Queue.new
@worker = Thread.new do
while args = @queue.pop
perform(*args)
end
rescue ClosedQueueError
nil
rescue StandardError => e
loggable('error', e.message, class: e.class.to_s, backtrace: e.backtrace)
end
at_exit { terminate }
end
end
|
.terminate ⇒ Object
47
48
49
50
|
# File 'lib/getaround_utils/utils/async_queue.rb', line 47
def terminate
@queue&.close
@worker&.join
end
|