Class: Takagi::EventBus::AsyncExecutor::ThreadExecutor

Inherits:
Object
  • Object
show all
Defined in:
lib/takagi/event_bus/async_executor.rb,
sig/takagi/event_bus/async_executor.rbs

Overview

Thread-based executor (default fallback)

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(size:) ⇒ ThreadExecutor

Returns a new instance of ThreadExecutor.

Parameters:

  • size: (Object) 


12
13
14
15
16
17
18
 
# File 'lib/takagi/event_bus/async_executor.rb', line 12

def initialize(size:)
  @size = size.positive? ? size : 1
  @queue = Queue.new
  @threads = []
  @shutdown = false
  start_workers
end

Instance Attribute Details

#size ⇒ Object (readonly)

Returns the value of attribute size.

Returns:

  • (Object) 


10
11
12
 
# File 'lib/takagi/event_bus/async_executor.rb', line 10

def size
  @size
end

Instance Method Details

#post(handler, message) ⇒ Object

Parameters:

  • handler (Object)
  • message (Object)

Returns:

  • (Object) 


20
21
22
23
24
 
# File 'lib/takagi/event_bus/async_executor.rb', line 20

def post(handler, message)
  raise 'Executor is shutdown' if @shutdown

  @queue << [handler, message]
end

#register_handler(_handler) ⇒ nil

Parameters:

  • _handler (Object)

Returns:

  • (nil) 


26
 
# File 'lib/takagi/event_bus/async_executor.rb', line 26

def register_handler(_handler); end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


39
40
41
 
# File 'lib/takagi/event_bus/async_executor.rb', line 39

def running?
  !@shutdown
end

#shutdown ⇒ nil, untyped

Returns:

  • (nil, untyped) 


30
31
32
33
34
35
36
37
 
# File 'lib/takagi/event_bus/async_executor.rb', line 30

def shutdown
  return if @shutdown

  @shutdown = true
  @size.times { @queue << nil }
  @threads.each(&:join)
  @threads.clear
end

#start_workers ⇒ Object

Returns:

  • (Object) 


49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/takagi/event_bus/async_executor.rb', line 49

def start_workers
  @size.times do |index|
    @threads << Thread.new do
      Thread.current.name = "EventBus-ThreadExecutor-#{index}"
      loop do
        job = @queue.pop
        break if job.nil?

        handler, message = job
        begin
          handler.call(message)
        rescue StandardError => e
          warn "EventBus ThreadExecutor error: #{e.class} - #{e.message}"
        end
      end
    end
  end
end

#stats ⇒ { mode: :threads, size: untyped }

Returns:

  • ({ mode: :threads, size: untyped }) 


43
44
45
 
# File 'lib/takagi/event_bus/async_executor.rb', line 43

def stats
  { mode: :threads, size: @size }
end

#unregister_handler(_handler) ⇒ nil

Parameters:

  • _handler (Object)

Returns:

  • (nil) 


28
 
# File 'lib/takagi/event_bus/async_executor.rb', line 28

def unregister_handler(_handler); end