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

find_by!, log_details?, 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:



126
127
128
129
130
131
132
133
134
 
# File 'lib/camunda/external_task.rb', line 126

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:



147
148
149
150
151
152
153
 
# File 'lib/camunda/external_task.rb', line 147

def self.fetch_and_queue(topics, lock_duration: nil, long_polling_duration: nil)
  fetch_and_lock(topics, lock_duration: lock_duration, long_polling_duration: long_polling_duration).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



35
36
37
 
# File 'lib/camunda/external_task.rb', line 35

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



23
24
25
 
# File 'lib/camunda/external_task.rb', line 23

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



29
30
31
 
# File 'lib/camunda/external_task.rb', line 29

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:



52
53
54
55
56
 
# File 'lib/camunda/external_task.rb', line 52

def bpmn_error(bpmn_exception)
  self.class.post_raw("#{collection_path}/#{id}/bpmnError",
                      workerId: worker_id, variables: serialize_variables(bpmn_exception.variables),
                      errorCode: bpmn_exception.error_code, errorMessage: bpmn_exception.message)[:response]
end

#collection_path ⇒ String

Helper method for instances since collection_path is a class method

Returns:

  • (String) 


78
79
80
 
# File 'lib/camunda/external_task.rb', line 78

def collection_path
  self.class.collection_path
end

#complete(variables = {}) ⇒ Object

Completes the process instance of a fetched task

Parameters:

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

    submitted when starting the process definition

Raises:



61
62
63
64
65
66
67
 
# File 'lib/camunda/external_task.rb', line 61

def complete(variables={})
  self.class.post_raw("#{collection_path}/#{id}/complete",
                      workerId: worker_id, variables: serialize_variables(variables))[:response]
      .tap do |response|
    raise SubmissionError, response.body[:data][:message] unless response.success?
  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



41
42
43
44
45
46
47
48
 
# File 'lib/camunda/external_task.rb', line 41

def failure(exception, input_variables={})
  variables_information = "Input variables are #{input_variables.inspect}\n\n" if input_variables.present?
  self.class.post_raw("#{collection_path}/#{id}/failure",
                      workerId: worker_id, errorMessage: exception.message,
                      errorDetails:
                        variables_information.to_s + exception.message +
                        backtrace_cleaner.clean(exception.backtrace).join("\n"))[:response]
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:



100
101
102
 
# File 'lib/camunda/external_task.rb', line 100

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:



113
114
115
 
# File 'lib/camunda/external_task.rb', line 113

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:



164
165
166
167
168
 
# File 'lib/camunda/external_task.rb', line 164

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



157
158
159
 
# File 'lib/camunda/external_task.rb', line 157

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

#variables ⇒ Object

deserializes JSON attributes from variables returned by Camunda API



83
84
85
86
87
88
89
90
91
 
# File 'lib/camunda/external_task.rb', line 83

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) 


72
73
74
 
# File 'lib/camunda/external_task.rb', line 72

def worker_id
  self.class.worker_id
end