Class: Shikibu::WorkflowContext

Inherits:
Object
  • Object
show all
Defined in:
lib/shikibu/context.rb

Overview

Context object passed to workflow execution Provides access to workflow state, history, and operations

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(instance_id:, workflow_name:, worker_id:, storage:, hooks: nil, history_cache: {}, replaying: false) ⇒ WorkflowContext 



10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/shikibu/context.rb', line 10

def initialize(
  instance_id:,
  workflow_name:,
  worker_id:,
  storage:,
  hooks: nil,
  history_cache: {},
  replaying: false
)
  @instance_id = instance_id
  @workflow_name = workflow_name
  @worker_id = worker_id
  @storage = storage
  @hooks = hooks
  @history_cache = history_cache
  @replaying = replaying
  @activity_counters = Hash.new(0)
  @compensations = []
  @direct_subscriptions = Set.new
  @last_activity_id = nil
end

Instance Attribute Details

#current_activity_id ⇒ Object

Returns the value of attribute current_activity_id.



8
9
10
 
# File 'lib/shikibu/context.rb', line 8

def current_activity_id
  @current_activity_id
end

#hooks ⇒ Object (readonly)

Returns the value of attribute hooks.



7
8
9
 
# File 'lib/shikibu/context.rb', line 7

def hooks
  @hooks
end

#instance_id ⇒ Object (readonly)

Returns the value of attribute instance_id.



7
8
9
 
# File 'lib/shikibu/context.rb', line 7

def instance_id
  @instance_id
end

#last_activity_id ⇒ Object (readonly)

Returns the value of attribute last_activity_id.



7
8
9
 
# File 'lib/shikibu/context.rb', line 7

def last_activity_id
  @last_activity_id
end

#storage ⇒ Object (readonly)

Returns the value of attribute storage.



7
8
9
 
# File 'lib/shikibu/context.rb', line 7

def storage
  @storage
end

#worker_id ⇒ Object (readonly)

Returns the value of attribute worker_id.



7
8
9
 
# File 'lib/shikibu/context.rb', line 7

def worker_id
  @worker_id
end

#workflow_name ⇒ Object (readonly)

Returns the value of attribute workflow_name.



7
8
9
 
# File 'lib/shikibu/context.rb', line 7

def workflow_name
  @workflow_name
end

Instance Method Details

#cache_result(activity_id, result) ⇒ Object

Cache an activity result



69
70
71
 
# File 'lib/shikibu/context.rb', line 69

def cache_result(activity_id, result)
  @history_cache[activity_id] = result
end

#cached_result?(activity_id) ⇒ Boolean

Check if activity has cached result



62
63
64
 
# File 'lib/shikibu/context.rb', line 62

def cached_result?(activity_id)
  @history_cache.key?(activity_id)
end

#compensations ⇒ Object

Get registered compensations (LIFO order)



94
95
96
 
# File 'lib/shikibu/context.rb', line 94

def compensations
  @compensations.reverse
end

#direct_subscriptions ⇒ Object

Get list of direct subscriptions



132
133
134
 
# File 'lib/shikibu/context.rb', line 132

def direct_subscriptions
  @direct_subscriptions.to_a
end

#generate_activity_id(name) ⇒ String

Generate a unique activity ID for deterministic replay



45
46
47
48
 
# File 'lib/shikibu/context.rb', line 45

def generate_activity_id(name)
  @activity_counters[name] += 1
  "#{name}:#{@activity_counters[name]}"
end

#get_cached_result(activity_id) ⇒ Object?

Get cached result for an activity (during replay)



53
54
55
56
57
 
# File 'lib/shikibu/context.rb', line 53

def get_cached_result(activity_id)
  return nil unless @history_cache.key?(activity_id)

  @history_cache[activity_id]
end

#in_transaction? ⇒ Boolean

Check if currently in a transaction



109
110
111
 
# File 'lib/shikibu/context.rb', line 109

def in_transaction?
  @storage.in_transaction?
end

#publish(channel, data, metadata: nil) ⇒ String

Publish a message to a channel



194
195
196
 
# File 'lib/shikibu/context.rb', line 194

def publish(channel, data, metadata: nil)
  Channels.publish(self, channel, data, metadata: metadata)
end

#receive(channel, timeout: nil, mode: :broadcast) ⇒ Channels::Message

Wait for a message on a channel



177
178
179
 
# File 'lib/shikibu/context.rb', line 177

def receive(channel, timeout: nil, mode: :broadcast)
  Channels.receive(self, channel, timeout: timeout, mode: mode)
end

#record_last_activity_id(activity_id) ⇒ Object

Record the last completed activity ID (for on_failure to reference)



33
34
35
 
# File 'lib/shikibu/context.rb', line 33

def record_last_activity_id(activity_id)
  @last_activity_id = activity_id
end

#recur(**new_input) ⇒ Object

Restart workflow with new input (tail recursion pattern)

Raises:



144
145
146
 
# File 'lib/shikibu/context.rb', line 144

def recur(**new_input)
  raise RecurSignal, new_input
end

#register_compensation(activity_id:, compensation_name:, args:) ⇒ Object

Register a compensation action (Romancy/Edda compatible)



77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/shikibu/context.rb', line 77

def register_compensation(activity_id:, compensation_name:, args:)
  @compensations << {
    activity_id: activity_id,
    compensation_name: compensation_name,
    args: args
  }

  # Also persist to storage
  @storage.push_compensation(
    instance_id: @instance_id,
    activity_id: activity_id,
    activity_name: compensation_name,
    args: args
  )
end

#replaying? ⇒ Boolean

Check if currently replaying from history



38
39
40
 
# File 'lib/shikibu/context.rb', line 38

def replaying?
  @replaying
end

#send_to(target_instance_id, channel, data, metadata: nil) ⇒ String

Send a message directly to another workflow instance



204
205
206
 
# File 'lib/shikibu/context.rb', line 204

def send_to(target_instance_id, channel, data, metadata: nil)
  Channels.send_to(self, target_instance_id, channel, data, metadata: metadata)
end

#session ⇒ Object

Access database session (within the same transaction)



99
100
101
 
# File 'lib/shikibu/context.rb', line 99

def session
  @storage.db
end

#sleep(seconds, timer_id: nil) ⇒ Object

Wait for a timer to expire



151
152
153
 
# File 'lib/shikibu/context.rb', line 151

def sleep(seconds, timer_id: nil)
  sleep_until(Time.now + seconds, timer_id: timer_id)
end

#sleep_until(until_time, timer_id: nil) ⇒ Object

Wait until a specific time

Raises:



158
159
160
161
162
163
164
165
166
167
168
169
170
 
# File 'lib/shikibu/context.rb', line 158

def sleep_until(until_time, timer_id: nil)
  timer_id ||= generate_activity_id('timer')

  # Check if already expired in history
  return get_cached_result(timer_id) if replaying? && cached_result?(timer_id)

  # Register timer and suspend
  raise WaitForTimerSignal.new(
    timer_id: timer_id,
    expires_at: until_time,
    activity_id: timer_id
  )
end

#subscribe(channel, mode: :broadcast) ⇒ Channels::Subscription

Subscribe to a channel



117
118
119
120
121
122
 
# File 'lib/shikibu/context.rb', line 117

def subscribe(channel, mode: :broadcast)
  mode_str = mode.to_s
  @direct_subscriptions.add(channel) if mode_str == ChannelMode::DIRECT

  Channels.subscribe(self, channel, mode: mode)
end

#subscriptions ⇒ Array<Channels::Subscription>

List all active subscriptions



138
139
140
 
# File 'lib/shikibu/context.rb', line 138

def subscriptions
  Channels.subscriptions(self)
end

#transaction ⇒ Object

Execute a block within a transaction



104
105
106
 
# File 'lib/shikibu/context.rb', line 104

def transaction(&)
  @storage.transaction(&)
end

#try_receive(channel, mode: :broadcast) ⇒ Channels::Message?

Try to receive without blocking (returns nil if no message)



185
186
187
 
# File 'lib/shikibu/context.rb', line 185

def try_receive(channel, mode: :broadcast)
  Channels.try_receive(self, channel, mode: mode)
end

#unsubscribe(channel) ⇒ Object

Unsubscribe from a channel



126
127
128
129
 
# File 'lib/shikibu/context.rb', line 126

def unsubscribe(channel)
  @direct_subscriptions.delete(channel)
  Channels.unsubscribe(self, channel)
end