Class: JFlow::Activity::Worker

Inherits:
Object
  • Object
show all
Defined in:
lib/jflow/activity/worker.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(domain, tasklist) ⇒ Worker

Returns a new instance of Worker.



7
8
9
10
 
# File 'lib/jflow/activity/worker.rb', line 7

def initialize(domain, tasklist)
  @domain = domain
  @tasklist = tasklist
end

Instance Attribute Details

#domain ⇒ Object (readonly)

Returns the value of attribute domain.



5
6
7
 
# File 'lib/jflow/activity/worker.rb', line 5

def domain
  @domain
end

#tasklist ⇒ Object (readonly)

Returns the value of attribute tasklist.



5
6
7
 
# File 'lib/jflow/activity/worker.rb', line 5

def tasklist
  @tasklist
end

Instance Method Details

#poll ⇒ Object 



21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/jflow/activity/worker.rb', line 21

def poll
  Thread.current.set_state(:polling)
  response = JFlow.configuration.swf_client.poll_for_activity_task(poll_params)
  if response.task_token
    task = JFlow::Activity::Task.new(response)
    log "Got task #{task.workflow_id}-#{task.run_id}"
    if should_be_working?
      process(task)
    else
      #The worker is shuting down, we don't want to start working on anything
      #so we fail the task and let the decider queue it up for retry later
      task.failed!(Exception.new("Worker is going down!"))
    end
  else
    log "Got no task"
  end
end

#process(task) ⇒ Object 



39
40
41
42
43
44
45
46
 
# File 'lib/jflow/activity/worker.rb', line 39

def process(task)
  begin
    Thread.current.set_state(:working)
    task.run!
  rescue => exception
    task.failed!(exception)
  end
end

#start! ⇒ Object 



12
13
14
15
16
17
18
 
# File 'lib/jflow/activity/worker.rb', line 12

def start!
  while should_be_working?
    log "Polling for task on #{domain} - #{tasklist}"
    poll
  end
  log "Thread is marked as exiting, stopping the poll"
end