Class: Sqskiq::Manager

Inherits:
Object
  • Object
show all
Includes:
Celluloid, SignalHandler
Defined in:
lib/sqskiq/manager.rb

Instance Method Summary collapse

Methods included from SignalHandler

#shutting_down, #subscribe_for_shutdown

Constructor Details

#initialize ⇒ Manager

Returns a new instance of Manager.



10
11
12
 
# File 'lib/sqskiq/manager.rb', line 10

def initialize
  subscribe_for_shutdown
end

Instance Method Details

#batch_process_done(messages) ⇒ Object 



26
27
28
29
 
# File 'lib/sqskiq/manager.rb', line 26

def batch_process_done(messages)
  @deleter.async.delete(messages)
  new_fetch(1)
end

#bootstrap ⇒ Object 



14
15
16
17
18
19
20
 
# File 'lib/sqskiq/manager.rb', line 14

def bootstrap
  @fetcher = Celluloid::Actor[:fetcher]
  @batch_processor = Celluloid::Actor[:batch_processor]
  @deleter = Celluloid::Actor[:deleter]

  new_fetch(@fetcher.size)
end

#fetch_done(messages) ⇒ Object 



22
23
24
 
# File 'lib/sqskiq/manager.rb', line 22

def fetch_done(messages)
  @batch_processor.async.batch_process(messages) unless @shutting_down
end

#new_fetch(num) ⇒ Object 



31
32
33
 
# File 'lib/sqskiq/manager.rb', line 31

def new_fetch(num)
  num.times { @fetcher.async.fetch unless @shutting_down }
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


35
36
37
 
# File 'lib/sqskiq/manager.rb', line 35

def running?
  not (@shutting_down and @deleter.busy_size == 0 and @batch_processor.busy_size == 0)
end