Class: Phronomy::WorkflowRunner Private

Inherits:
Object
  • Object
show all
Includes:
Runnable
Defined in:
lib/phronomy/workflow_runner.rb

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Execution boundary for compiled Workflows.

WorkflowRunner separates three identities:

  • application session_id remains caller/tracing metadata;
  • thread_id identifies durable Workflow state;
  • fsm_session_id identifies one Runtime FSMSession execution.

Workflow persistence is synchronous at the repository contract but is always invoked through Runtime's OffloadPool from EventLoop-driven lifecycle paths.

Defined Under Namespace

Classes: Execution, WorkflowPersistenceCommand

Constant Summary collapse

FINISH =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future. 

:__end__

Instance Method Summary collapse

Methods included from Runnable

#batch, #trace

Constructor Details

#initialize(state_class:, entry_actions:, declared_states:, auto_transitions:, external_events:, entry_point:, exit_actions: {}, wait_state_names: [], persistence: nil) ⇒ WorkflowRunner

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of WorkflowRunner.



40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/phronomy/workflow_runner.rb', line 40

def initialize(
  state_class:,
  entry_actions:,
  declared_states:,
  auto_transitions:,
  external_events:,
  entry_point:,
  exit_actions: {},
  wait_state_names: [],
  persistence: nil
)
  @state_class = state_class
  @entry_actions = entry_actions
  @declared_states = declared_states
  @auto_state_set = auto_transitions.each_with_object({}) do |transition, set|
    set[transition[:from]] = true
  end
  @external_events = external_events
  @entry_point = entry_point
  @wait_state_names = wait_state_names
  @persistence = persistence
  @phase_machine_class = Workflow::PhaseMachineBuilder.new(
    entry_point: @entry_point,
    declared_states: @declared_states,
    wait_state_names: @wait_state_names,
    external_events: @external_events,
    entry_actions: @entry_actions,
    auto_transitions: auto_transitions,
    exit_actions: exit_actions
  ).build
end

Instance Method Details

#deliver_persistence_on_event_loop(command) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Called only by EventLoop for terminal Workflow persistence completion.



137
138
139
140
141
142
143
144
145
146
147
148
 
# File 'lib/phronomy/workflow_runner.rb', line 137

def deliver_persistence_on_event_loop(command)
  event_loop = Phronomy::Runtime.instance.event_loop
  event_loop.release_workflow(
    command.thread_id,
    owner_fsm_session_id: command.fsm_session_id
  )
  if command.error
    fail_task(command.result_task, command.error)
  else
    complete_task(command.result_task, command.result)
  end
end

#invoke(input, config: {}) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/phronomy/workflow_runner.rb', line 72

def invoke(input, config: {})
  ensure_blocking_call_allowed!(:invoke, :invoke_async)
  caller_meta = {}
  caller_meta[:user_id] = config[:user_id] if config[:user_id]
  caller_meta[:session_id] = config[:session_id] if config[:session_id]

  trace("workflow.invoke", input: input.inspect, **caller_meta) do |_span|
    result = start_new_execution(input, config).wait_result
    [result, nil]
  end
end

#invoke_deferred(input, config: {}) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



84
85
86
 
# File 'lib/phronomy/workflow_runner.rb', line 84

def invoke_deferred(input, config: {})
  start_new_execution(input, config)
end

#resume(state:, input: nil) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



95
96
97
 
# File 'lib/phronomy/workflow_runner.rb', line 95

def resume(state:, input: nil)
  send_event(state: state, event: :resume, input: input)
end

#send_event(state:, event:, input: nil) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/phronomy/workflow_runner.rb', line 99

def send_event(state:, event:, input: nil)
  ensure_blocking_call_allowed!(:send_event, :signal)
  current_phase = state.phase.to_sym
  event_name = resolve_resume_event(current_phase, event)
  thread_id = state.thread_id
  unless thread_id
    raise ArgumentError, "Halted WorkflowContext has no thread_id"
  end

  start_resume_execution(
    state,
    input: input,
    event_name: event_name,
    current_phase: current_phase
  ).wait_result
end

#signal(thread_id:, event:, payload: nil) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Posts an application-defined event to the currently live Workflow owner. thread_id is resolved to its Runtime-only fsm_session_id by EventLoop; the application session_id is deliberately unrelated to this routing.

Raises:

  • (ArgumentError) 


119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/phronomy/workflow_runner.rb', line 119

def signal(thread_id:, event:, payload: nil)
  raise ArgumentError, "thread_id is required" if thread_id.nil?

  event_name = event.to_sym
  unless @external_events.key?(event_name)
    raise ArgumentError,
      "Unknown event #{event_name.inspect}. " \
      "Valid events: #{@external_events.keys.inspect}"
  end

  Phronomy::Runtime.instance.event_loop.post_to_workflow(
    thread_id: thread_id,
    event: event_name,
    payload: payload
  )
end

#stream(input, config: {}, &observer) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Raises:

  • (ArgumentError) 


88
89
90
91
92
93
 
# File 'lib/phronomy/workflow_runner.rb', line 88

def stream(input, config: {}, &observer)
  ensure_blocking_call_allowed!(:stream, :invoke_async)
  raise ArgumentError, "stream requires a block" unless observer

  start_new_execution(input, config, stable_observer: observer).wait_result
end