Class: Temporalio::Internal::Worker::WorkflowWorker::State

Inherits:
Object
  • Object
show all
Defined in:
lib/temporalio/internal/worker/workflow_worker.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(workflow_definitions:, bridge_worker:, logger:, metric_meter:, data_converter:, deadlock_timeout:, illegal_calls:, namespace:, task_queue:, disable_eager_activity_execution:, workflow_interceptors:, workflow_failure_exception_types:, unsafe_workflow_io_enabled:, assert_valid_local_activity:) ⇒ State

Returns a new instance of State.



201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 201

def initialize(
  workflow_definitions:, bridge_worker:, logger:, metric_meter:, data_converter:, deadlock_timeout:,
  illegal_calls:, namespace:, task_queue:, disable_eager_activity_execution:,
  workflow_interceptors:, workflow_failure_exception_types:, unsafe_workflow_io_enabled:,
  assert_valid_local_activity:
)
  @workflow_definitions = workflow_definitions
  @bridge_worker = bridge_worker
  @logger = logger
  @metric_meter = metric_meter
  @data_converter = data_converter
  @deadlock_timeout = deadlock_timeout
  @illegal_calls = illegal_calls
  @namespace = namespace
  @task_queue = task_queue
  @disable_eager_activity_execution = disable_eager_activity_execution
  @workflow_interceptors = workflow_interceptors
  @workflow_failure_exception_types = workflow_failure_exception_types
  @unsafe_workflow_io_enabled = unsafe_workflow_io_enabled
  @assert_valid_local_activity = assert_valid_local_activity

  @running_workflows = {}
  @running_workflows_mutex = Mutex.new
end

Instance Attribute Details

#assert_valid_local_activity ⇒ Object (readonly)

Returns the value of attribute assert_valid_local_activity.



194
195
196
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 194

def assert_valid_local_activity
  @assert_valid_local_activity
end

#bridge_worker ⇒ Object (readonly)

Returns the value of attribute bridge_worker.



194
195
196
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 194

def bridge_worker
  @bridge_worker
end

#data_converter ⇒ Object (readonly)

Returns the value of attribute data_converter.



194
195
196
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 194

def data_converter
  @data_converter
end

#deadlock_timeout ⇒ Object (readonly)

Returns the value of attribute deadlock_timeout.



194
195
196
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 194

def deadlock_timeout
  @deadlock_timeout
end

#disable_eager_activity_execution ⇒ Object (readonly)

Returns the value of attribute disable_eager_activity_execution.



194
195
196
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 194

def disable_eager_activity_execution
  @disable_eager_activity_execution
end

#illegal_calls ⇒ Object (readonly)

Returns the value of attribute illegal_calls.



194
195
196
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 194

def illegal_calls
  @illegal_calls
end

#logger ⇒ Object (readonly)

Returns the value of attribute logger.



194
195
196
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 194

def logger
  @logger
end

#metric_meter ⇒ Object (readonly)

Returns the value of attribute metric_meter.



194
195
196
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 194

def metric_meter
  @metric_meter
end

#namespace ⇒ Object (readonly)

Returns the value of attribute namespace.



194
195
196
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 194

def namespace
  @namespace
end

#on_eviction=(value) ⇒ Object (writeonly)

Sets the attribute on_eviction

Parameters:

  • value —

    the value to set the attribute on_eviction to.



199
200
201
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 199

def on_eviction=(value)
  @on_eviction = value
end

#task_queue ⇒ Object (readonly)

Returns the value of attribute task_queue.



194
195
196
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 194

def task_queue
  @task_queue
end

#unsafe_workflow_io_enabled ⇒ Object (readonly)

Returns the value of attribute unsafe_workflow_io_enabled.



194
195
196
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 194

def unsafe_workflow_io_enabled
  @unsafe_workflow_io_enabled
end

#workflow_definitions ⇒ Object (readonly)

Returns the value of attribute workflow_definitions.



194
195
196
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 194

def workflow_definitions
  @workflow_definitions
end

#workflow_failure_exception_types ⇒ Object (readonly)

Returns the value of attribute workflow_failure_exception_types.



194
195
196
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 194

def workflow_failure_exception_types
  @workflow_failure_exception_types
end

#workflow_interceptors ⇒ Object (readonly)

Returns the value of attribute workflow_interceptors.



194
195
196
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 194

def workflow_interceptors
  @workflow_interceptors
end

Instance Method Details

#evict_all ⇒ Object 



242
243
244
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 242

def evict_all
  @running_workflows_mutex.synchronize { @running_workflows.clear }
end

#evict_running_workflow(run_id, cache_remove_job) ⇒ Object 



237
238
239
240
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 237

def evict_running_workflow(run_id, cache_remove_job)
  @running_workflows_mutex.synchronize { @running_workflows.delete(run_id) }
  @on_eviction&.call(run_id, cache_remove_job)
end

#get_or_create_running_workflow(run_id) ⇒ Object

This can never be called at the same time for the same run ID on the same state object



227
228
229
230
231
232
233
234
235
 
# File 'lib/temporalio/internal/worker/workflow_worker.rb', line 227

def get_or_create_running_workflow(run_id, &)
  instance = @running_workflows_mutex.synchronize { @running_workflows[run_id] }
  # If instance is not there, we create it out of lock then store it under lock
  unless instance
    instance = yield
    @running_workflows_mutex.synchronize { @running_workflows[run_id] = instance }
  end
  instance
end