Class: Taskinator::Task

Inherits:
Object
  • Object
show all
Includes:
Comparable, Instrumentation, Persistence, Workflow
Defined in:
lib/taskinator/task.rb

Direct Known Subclasses

Job, Step, SubProcess

Defined Under Namespace

Classes: Job, Step, SubProcess

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Instrumentation

#cancelled_payload, #completed_payload, #enqueued_payload, #failed_payload, #instrument, #paused_payload, #processing_payload, #resumed_payload

Methods included from Persistence

add_process_to_list, deserialize, included, processes_list_key, serialize

Methods included from Workflow

#current_state, #current_state=, #transition

Constructor Details

#initialize(process, options = {}) ⇒ Task

Returns a new instance of Task.

Raises:

  • (ArgumentError) 


33
34
35
36
37
38
39
40
41
42
 
# File 'lib/taskinator/task.rb', line 33

def initialize(process, options={})
  raise ArgumentError, 'process' if process.nil? || !process.is_a?(Process)

  @uuid = "#{process.uuid}:task:#{Taskinator.generate_uuid}"
  @process = process
  @options = options
  @queue = options.delete(:queue)
  @created_at = Time.now.utc
  @updated_at = created_at
end

Instance Attribute Details

#created_at ⇒ Object (readonly)

Returns the value of attribute created_at.



27
28
29
 
# File 'lib/taskinator/task.rb', line 27

def created_at
  @created_at
end

#next ⇒ Object

the next task in the sequence



31
32
33
 
# File 'lib/taskinator/task.rb', line 31

def next
  @next
end

#options ⇒ Object (readonly)

Returns the value of attribute options.



25
26
27
 
# File 'lib/taskinator/task.rb', line 25

def options
  @options
end

#process ⇒ Object (readonly)

Returns the value of attribute process.



23
24
25
 
# File 'lib/taskinator/task.rb', line 23

def process
  @process
end

#queue ⇒ Object (readonly)

Returns the value of attribute queue.



26
27
28
 
# File 'lib/taskinator/task.rb', line 26

def queue
  @queue
end

#updated_at ⇒ Object (readonly)

Returns the value of attribute updated_at.



28
29
30
 
# File 'lib/taskinator/task.rb', line 28

def updated_at
  @updated_at
end

#uuid ⇒ Object (readonly)

Returns the value of attribute uuid.



24
25
26
 
# File 'lib/taskinator/task.rb', line 24

def uuid
  @uuid
end

Class Method Details

.define_job_task(process, job, args, options = {}) ⇒ Object 



14
15
16
 
# File 'lib/taskinator/task.rb', line 14

def define_job_task(process, job, args, options={})
  Job.new(process, job, args, options)
end

.define_step_task(process, method, args, options = {}) ⇒ Object 



10
11
12
 
# File 'lib/taskinator/task.rb', line 10

def define_step_task(process, method, args, options={})
  Step.new(process, method, args, options)
end

.define_sub_process_task(process, sub_process, options = {}) ⇒ Object 



18
19
20
 
# File 'lib/taskinator/task.rb', line 18

def define_sub_process_task(process, sub_process, options={})
  SubProcess.new(process, sub_process, options)
end

Instance Method Details

#<=>(other) ⇒ Object 



54
55
56
 
# File 'lib/taskinator/task.rb', line 54

def <=>(other)
  uuid <=> other.uuid
end

#accept(visitor) ⇒ Object 



44
45
46
47
48
49
50
51
52
 
# File 'lib/taskinator/task.rb', line 44

def accept(visitor)
  visitor.visit_attribute(:uuid)
  visitor.visit_process_reference(:process)
  visitor.visit_task_reference(:next)
  visitor.visit_args(:options)
  visitor.visit_attribute(:queue)
  visitor.visit_attribute_time(:created_at)
  visitor.visit_attribute_time(:updated_at)
end

#cancel! ⇒ Object 



104
105
106
107
108
109
110
111
 
# File 'lib/taskinator/task.rb', line 104

def cancel!
  transition(:cancelled) do
    self.incr_cancelled if incr_count?
    instrument('taskinator.task.cancelled', cancelled_payload) do
      cancel if respond_to?(:cancel)
    end
  end
end

#cancelled? ⇒ Boolean

Returns:

  • (Boolean) 


113
114
115
 
# File 'lib/taskinator/task.rb', line 113

def cancelled?
  super || process.cancelled?
end

#complete! ⇒ Object 



93
94
95
96
97
98
99
100
101
102
 
# File 'lib/taskinator/task.rb', line 93

def complete!
  transition(:completed) do
    self.incr_completed if incr_count?
    instrument('taskinator.task.completed', completed_payload) do
      complete if respond_to?(:complete)
      # notify the process that this task has completed
      process.task_completed(self)
    end
  end
end

#enqueue ⇒ Object

subclasses must implement the following methods

Raises:

  • (NotImplementedError) 


136
137
138
 
# File 'lib/taskinator/task.rb', line 136

def enqueue
  raise NotImplementedError
end

#enqueue! ⇒ Object 



62
63
64
65
66
67
68
69
70
 
# File 'lib/taskinator/task.rb', line 62

def enqueue!
  return if paused? || cancelled?

  transition(:enqueued) do
    instrument('taskinator.task.enqueued', enqueued_payload) do
      enqueue
    end
  end
end

#fail!(error) ⇒ Object 



117
118
119
120
121
122
123
124
125
126
 
# File 'lib/taskinator/task.rb', line 117

def fail!(error)
  transition(:failed) do
    self.incr_failed if incr_count?
    instrument('taskinator.task.failed', failed_payload(error)) do
      fail(error) if respond_to?(:fail)
      # notify the process that this task has failed
      process.task_failed(self, error)
    end
  end
end

#incr_count? ⇒ Boolean

Returns:

  • (Boolean) 


128
129
130
 
# File 'lib/taskinator/task.rb', line 128

def incr_count?
  true
end

#paused? ⇒ Boolean

helper method

Returns:

  • (Boolean) 


89
90
91
 
# File 'lib/taskinator/task.rb', line 89

def paused?
  super || process.paused?
end

#start ⇒ Object

Raises:

  • (NotImplementedError) 


140
141
142
 
# File 'lib/taskinator/task.rb', line 140

def start
  raise NotImplementedError
end

#start! ⇒ Object 



72
73
74
75
76
77
78
79
80
81
 
# File 'lib/taskinator/task.rb', line 72

def start!
  return if paused? || cancelled?
  self.incr_processing if incr_count?

  transition(:processing) do
    instrument('taskinator.task.processing', processing_payload) do
      start
    end
  end
end

#to_s ⇒ Object 



58
59
60
 
# File 'lib/taskinator/task.rb', line 58

def to_s
  "#<#{self.class.name}:#{uuid}>"
end