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) 


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

def aborted?
  state == 'aborted'
end

#acquire_lock? ⇒ Boolean

Is this task pending?

Returns:

  • (Boolean) 


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

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

#acts_as_task_task_queue ⇒ Object 



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

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



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

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) 


169
170
171
172
173
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 169

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



177
178
179
180
181
182
183
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 177

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) 


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

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

#execute! ⇒ Object

Execute task



200
201
202
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 200

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

#executed? ⇒ Boolean

Was this task already executed?

Returns:

  • (Boolean) 


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

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

#init_queue? ⇒ Boolean

Is this task waiting until the queue is initialized?

Returns:

  • (Boolean) 


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

def init_queue?
  state == 'init_queue'
end

#md5 ⇒ Object

Calculate md5-Checksum



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

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) 


109
110
111
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 109

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

#pending? ⇒ Boolean

Is this task pending?

Returns:

  • (Boolean) 


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

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

#prosa ⇒ Object

Placeholder. Please override it in your model.



187
188
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 187

def prosa
end

#reloading_config? ⇒ Boolean

Is the task_queue in state config_reload?

Returns:

  • (Boolean) 


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

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) 


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

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) 


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

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

#update_message(args = {}) ⇒ Object

Update the attributes perc_finished and message according to the args



99
100
101
102
103
104
105
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 99

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(rc_and_msg, 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



67
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
 
# File 'lib/queue_dispatcher/acts_as_task.rb', line 67

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

  if rc_and_msg.is_a?(QueueDispatcher::RcAndMsg)
    rc = rc_and_msg.rc
    output = rc_and_msg.output
    error_msg = rc_and_msg.error_msg
  elsif rc_and_msg.kind_of?(Hash)
    rc = rc_and_msg[:rc]
    output = rc_and_msg[:output]
    error_msg = rc_and_msg[:error_msg]
  end

  output ||= ""

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

  self.update_attributes :task_queue_id => nil if remove_from_queue

  rc
end