Module: QueueDispatcher::ActsAsTask::InstanceMethods

Defined in:
lib/queue_dispatcher/acts_as_task.rb

Instance Method Summary collapse

Instance Method Details

#aborted? ⇒ Boolean

Was this task aborted?

Returns:

  • (Boolean) 


156
157
158
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 156

def aborted?
  state == 'aborted'
end

#acquire_lock? ⇒ Boolean

Is this task pending?

Returns:

  • (Boolean) 


132
133
134
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 132

def acquire_lock?
  acts_as_task_task_queue && acts_as_task_task_queue.running? && state == 'acquire_lock'
end

#acts_as_task_task_queue ⇒ Object 



54
55
56
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 54

def acts_as_task_task_queue
  self.send(self.class.acts_as_task_config.task_queue_class_name)
end

#args ⇒ Object

Add task_id to the args



60
61
62
63
64
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 60

def args
  a = super
  a[-1] = a.last.merge(task_id: self.id) if a && a.instance_of?(Array) && a.last.instance_of?(Hash)
  a
end

#dependent_tasks_executed? ⇒ Boolean

Are all dependent_tasks executed?

Returns:

  • (Boolean) 


174
175
176
177
178
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 174

def dependent_tasks_executed?
  state = true
  dependent_tasks.each{ |dt| state = false unless dt.executed? }
  state
end

#dependent_tasks_had_errors ⇒ Object

Check recursive, if one or more of the tasks, which this task is dependent on had errors



182
183
184
185
186
187
188
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 182

def dependent_tasks_had_errors
  error = false
  dependent_tasks.each do |t|
    error = true if t.state == 'error' || t.dependent_tasks_had_errors
  end
  error
end

#error? ⇒ Boolean

Had this task error(s)?

Returns:

  • (Boolean) 


150
151
152
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 150

def error?
  state == 'error' || state.blank?
end

#execute! ⇒ Object

Execute task



205
206
207
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 205

def execute!
  target.send(method_name, *args)
end

#executed? ⇒ Boolean

Was this task already executed?

Returns:

  • (Boolean) 


168
169
170
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 168

def executed?
  successful? || error? || aborted?
end

#init_queue? ⇒ Boolean

Is this task waiting until the queue is initialized?

Returns:

  • (Boolean) 


162
163
164
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 162

def init_queue?
  state == 'init_queue'
end

#md5 ⇒ Object

Calculate md5-Checksum



197
198
199
200
201
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 197

def md5
  attr_str = ''
  Task.attribute_names.each{ |a| attr_str += self.send(a).to_s }
  Digest('MD5').digest(attr_str)
end

#new? ⇒ Boolean

Is this task new?

Returns:

  • (Boolean) 


114
115
116
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 114

def new?
   state == 'new' || state == 'new_popped'
end

#pending? ⇒ Boolean

Is this task pending?

Returns:

  • (Boolean) 


120
121
122
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 120

def pending?
  acts_as_task_task_queue && state == 'new'
end

#prosa ⇒ Object

Placeholder. Please override it in your model.



192
193
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 192

def prosa
end

#reloading_config? ⇒ Boolean

Is the task_queue in state config_reload?

Returns:

  • (Boolean) 


126
127
128
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 126

def reloading_config?
  acts_as_task_task_queue && acts_as_task_task_queue.reloading_config? && acts_as_task_task_queue.tasks.where(state: 'new').first.id == id
end

#running? ⇒ Boolean

Is this task running?

Returns:

  • (Boolean) 


138
139
140
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 138

def running?
  acts_as_task_task_queue && acts_as_task_task_queue.running? && state == 'running'
end

#successful? ⇒ Boolean

Was this task finsihed successful?

Returns:

  • (Boolean) 


144
145
146
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 144

def successful?
  state == 'successful' || state == 'finished'
end

#update_message(args = {}) ⇒ Object

Update the attributes perc_finished and message according to the args



104
105
106
107
108
109
110
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 104

def update_message(args = {})
  msg = args[:msg]
  perc_finished = args[:perc_finished]

  self.update_attribute :message, msg if msg
  self.update_attribute :perc_finished, perc_finished if perc_finished
end

#update_state(result, remove_from_queue = false) ⇒ Object

This method updates the task state according to the return code of their corresponding command and removes it from the task_queue



68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 68

def update_state(result, remove_from_queue = false)
  rc = output = error_msg = nil

  if result.methods.include?(:rc) && result.methods.include?(:output) && result.methods.include?(:error_msg)
    rc        = result.rc
    output    = result.output
    error_msg = result.error_msg
    result    = nil
  elsif result.kind_of?(Hash)
    rc        = result[:rc]
    output    = result[:output]
    error_msg = result[:error_msg]
    result    = nil
  end

  output ||= ''

  if rc.nil? || rc == 0
    self.update_attributes :state         => 'successful',
                           :perc_finished => 100,
                           :message       => output.truncate(10256),
                           :result        => result
  else
    self.update_attributes :state     => 'error',
                           :error_msg => error_msg,
                           :message   => output.truncate(10256),
                           :result    => result
  end

  self.update_attributes :task_queue_id => nil if remove_from_queue

  rc
end