Class: Mode::Connector::Scheduler

Inherits:
Object
  • Object
show all
Defined in:
lib/mode/connector/scheduler.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(data_sources, options = {}) ⇒ Scheduler

Returns a new instance of Scheduler.



10
11
12
13
14
 
# File 'lib/mode/connector/scheduler.rb', line 10

def initialize(data_sources, options = {})
  @data_sources = data_sources
  @scheduler    = Rufus::Scheduler.new
  @max_jobs     = (options[:max_jobs] || 4).to_i
end

Instance Attribute Details

#data_sources ⇒ Object (readonly)

Returns the value of attribute data_sources.



8
9
10
 
# File 'lib/mode/connector/scheduler.rb', line 8

def data_sources
  @data_sources
end

#max_jobs ⇒ Object (readonly)

Returns the value of attribute max_jobs.



6
7
8
 
# File 'lib/mode/connector/scheduler.rb', line 6

def max_jobs
  @max_jobs
end

#scheduler ⇒ Object (readonly)

Returns the value of attribute scheduler.



7
8
9
 
# File 'lib/mode/connector/scheduler.rb', line 7

def scheduler
  @scheduler
end

Instance Method Details

#processors ⇒ Object 



39
40
41
 
# File 'lib/mode/connector/scheduler.rb', line 39

def processors
  scheduler.jobs(:tag => 'processor')
end

#start! ⇒ Object 



16
17
18
19
20
21
22
23
24
25
26
27
28
29
 
# File 'lib/mode/connector/scheduler.rb', line 16

def start!
  begin
    register
    data_sources.each(&:connection)
    # Make sure to leave this outside the scheduler intervals
  rescue => err
    Mode::Logger.instance.error(
      self.class.name, err.message, err.backtrace)
  end

  scheduler.interval('15m') { register }
  scheduler.interval('5s')  { process_messages }
  scheduler.join
end

#stop! ⇒ Object 



31
32
33
34
35
36
37
 
# File 'lib/mode/connector/scheduler.rb', line 31

def stop!
  stopper = Thread.new {
    scheduler.stop # Stop polling
  }

  stopper.join # wait for jobs to finish
end