Class: Camunda::ExternalTask

Inherits:
Model
  • Object
show all
Includes:
VariableSerialization
Defined in:
lib/camunda/external_task.rb

Overview

task (step 3).“

Defined Under Namespace

Classes: SubmissionError

Class Method Summary collapse

Instance Method Summary collapse

Methods included from VariableSerialization

#serialize_variables

Methods inherited from Model

base_path, find_by!, worker_id

Class Method Details

.fetch_and_lock(topics, lock_duration: nil, long_polling_duration: nil) ⇒ Camunda::ExternalTask

Locking means that the task is reserved for this worker for a certain duration beginning with the time of fetching and prevents that another worker can fetch the same task while the lock is valid. Locking duration is set in the the Camunda::Workflow configuration. Before an external task can be completed, it must be locked.

Examples:

task = Camunda::ExternalTask.fetch_and_lock("CamundaWorkflow")

Parameters:

  • topics (Array<String>) —

    definition keys

  • lock_duration (Integer) (defaults to: nil)
  • long_polling_duration (Integer) (defaults to: nil)

Returns:



122
123
124
125
126
127
128
129
130
 
# File 'lib/camunda/external_task.rb', line 122

def self.fetch_and_lock(topics, lock_duration: nil, long_polling_duration: nil)
  long_polling_duration ||= long_polling_duration()
  lock_duration ||= lock_duration()
  topic_details = Array(topics).map do |topic|
    { topicName: topic, lockDuration: lock_duration }
  end
  fetchAndLock workerId: worker_id, maxTasks: max_polling_tasks, asyncResponseTimeout: long_polling_duration,
               topics: topic_details
end

.fetch_and_queue(topics, lock_duration: nil, long_polling_duration: nil) ⇒ Camunda::ExternalTask

Locking means that the task is reserved for this worker for a certain duration beginning with the time of fetching and prevents that another worker can fetch the same task while the lock is valid. Locking duration is set in the the Camunda::Workflow configuration. Before an external task can be completed, it must be locked.

This method calls fetch_and_lock and then queues the jobs that were retrieved

Examples:

task = Camunda::ExternalTask.fetch_and_queue("CamundaWorkflow")

Parameters:

  • topics (Array<String>) —

    definition keys

  • lock_duration (Integer) (defaults to: nil)
  • long_polling_duration (Integer) (defaults to: nil)

Returns:



143
144
145
146
147
148
149
150
 
# File 'lib/camunda/external_task.rb', line 143

def self.fetch_and_queue(topics, lock_duration: nil, long_polling_duration: nil)
  response = fetch_and_lock(topics, lock_duration: lock_duration, long_polling_duration: long_polling_duration)
  response.each do |task|
    task.queue_task
  rescue Camunda::MissingImplementationClass => e
    task.failure(e)
  end
end

.lock_duration ⇒ Integer

Default lock duration time is set to 14 days in Camunda::Workflow.configuration.

Returns:

  • (Integer) —

    default lock duration time



39
40
41
 
# File 'lib/camunda/external_task.rb', line 39

def self.lock_duration
  Camunda::Workflow.configuration.lock_duration.in_milliseconds
end

.long_polling_duration ⇒ Integer

Note:

long_polling_duration is defaulted to 30 seconds in Camunda::Workflow.configuration.

Returns default polling duration from Camunda::Workflow configuration.

Returns:

  • (Integer) —

    default polling duration from Camunda::Workflow configuration



27
28
29
 
# File 'lib/camunda/external_task.rb', line 27

def self.long_polling_duration
  Camunda::Workflow.configuration.long_polling_duration.in_milliseconds
end

.max_polling_tasks ⇒ Integer

Note:

Max polling tasks is defaulted to 2 in Camunda::Workflow.configuration.

Returns sets the max polling tasks.

Returns:

  • (Integer) —

    sets the max polling tasks



33
34
35
 
# File 'lib/camunda/external_task.rb', line 33

def self.max_polling_tasks
  Camunda::Workflow.configuration.max_polling_tasks
end

Instance Method Details

#bpmn_error(bpmn_exception) ⇒ Object

Reports the error to Camunda and creates an incident for the process instance.

Parameters:



55
56
57
58
59
 
# File 'lib/camunda/external_task.rb', line 55

def bpmn_error(bpmn_exception)
  self.class.request(:post, "external-task/#{id}/bpmnError",
                     workerId: worker_id, variables: serialize_variables(bpmn_exception.variables),
                     errorCode: bpmn_exception.error_code, errorMessage: bpmn_exception.message)
end

#complete(variables = {}) ⇒ Object

Completes the process instance of a fetched task

Parameters:

  • variables (Hash) (defaults to: {}) —

    submitted when starting the process definition

Raises:



64
65
66
67
68
69
 
# File 'lib/camunda/external_task.rb', line 64

def complete(variables={})
  self.class.request(:post, "external-task/#{id}/complete", workerId: worker_id,
                                                            variables: serialize_variables(variables)).tap do |response|
    raise SubmissionError, response.errors["message"] unless response.errors.blank?
  end
end

#failure(exception, input_variables = {}) ⇒ Object

Reports a failure to Camunda process definition and creates an incident for a process instance.

Parameters:

  • input_variables (Hash) (defaults to: {}) —

    process variables



45
46
47
48
49
50
51
 
# File 'lib/camunda/external_task.rb', line 45

def failure(exception, input_variables={})
  variables_information = "Input variables are #{input_variables.inspect}\n\n" if input_variables.present?
  self.class.request(:post, "external-task/#{id}/failure", workerId: worker_id, errorMessage: exception.message,
                                                           errorDetails:
                  variables_information.to_s + exception.message +
                  backtrace_cleaner.clean(exception.backtrace).join("\n"))
end

#queue_task ⇒ Camunda::ExternalTask

Queues the task to run at a specific time via ActiveJob

Examples:

# Below will retrieve current running process instances with the definition key "CamundaWorkflow"
# and queues the instances to be run at a specific time.
task = Camunda::ExternalTask.fetch_and_lock("CamundaWorkflow")
task.each(&:queue_task)

Returns:



96
97
98
 
# File 'lib/camunda/external_task.rb', line 96

def queue_task
  task_class.perform_later(id, variables)
end

#run_now ⇒ Camunda::ExternalTask

Runs the task using ActiveJob based on the classes created by bpmn_classes_generator. Before an external task can be run, you must #fetch_and_lock the task.

Examples:

# Below will retrieve current running process instances with the definition key "CamundaWorkflow"
# and run the instances.
task = Camunda::ExternalTask.fetch_and_lock("CamundaWorkflow")
task.each(&:run_now)

Returns:

See Also:



109
110
111
 
# File 'lib/camunda/external_task.rb', line 109

def run_now
  task_class_name.safe_constantize.perform_now id, variables
end

#task_class ⇒ Module

Checks to make sure an implementation class is available

Returns:

  • (Module) —

    return class name

Raises:



161
162
163
164
165
 
# File 'lib/camunda/external_task.rb', line 161

def task_class
  task_class_name.safe_constantize.tap do |klass|
    raise Camunda::MissingImplementationClass, task_class_name unless klass
  end
end

#task_class_name ⇒ String

Returns the class name which is supposed to implement this task

Returns:

  • (String) —

    modularized class name of bpmn task implementation



154
155
156
 
# File 'lib/camunda/external_task.rb', line 154

def task_class_name
  "#{process_definition_key}::#{activity_id}"
end

#variables ⇒ Object

deserializes JSON attributes from variables returned by Camunda API



79
80
81
82
83
84
85
86
87
 
# File 'lib/camunda/external_task.rb', line 79

def variables
  super.transform_values do |details|
    if details['type'] == 'Json'
      JSON.parse(details['value'])
    else
      details['value']
    end
  end
end

#worker_id ⇒ String

Note:

default is set to ‘0’ in Camunda::Workflow.configuration

Returns the worker id for an external task

Returns:

  • (String) 


74
75
76
 
# File 'lib/camunda/external_task.rb', line 74

def worker_id
  self.class.worker_id
end