Module: Shikibu::Channels

Defined in:
lib/shikibu/channels.rb

Overview

Channel-based messaging for workflow communication Supports three modes:

- broadcast: All subscribers receive all messages
- competing: Each message goes to exactly one subscriber
- direct: Point-to-point messaging to specific instance

Examples:

Broadcast (pub/sub)

# Workflow A
subscribe 'notifications', mode: :broadcast
message = receive 'notifications', timeout: 60

# Workflow B
publish 'notifications', { type: 'alert', text: 'Hello!' }

Competing consumers (work queue)

# Worker workflows (multiple instances)
subscribe 'tasks', mode: :competing
task = receive 'tasks'
process(task)

# Producer
publish 'tasks', { job_id: 123, action: 'process' }

Direct messaging (point-to-point)

# Parent workflow
child_id = start_workflow(ChildWorkflow, parent_id: instance_id)
subscribe 'results', mode: :direct
result = receive 'results', timeout: 300

# Child workflow
send_to parent_id, 'results', { status: 'done', data: result }

Defined Under Namespace

Classes: Message, Subscription

Class Method Summary collapse

Class Method Details

.publish(ctx, channel, data, metadata: nil) ⇒ String

Publish a message to a channel



187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
 
# File 'lib/shikibu/channels.rb', line 187

def publish(ctx, channel, data, metadata: nil)
  full_metadata = (metadata || {}).merge(
    source_instance_id: ctx.instance_id,
    source_workflow: ctx.workflow_name,
    published_at: Time.now.iso8601
  )

  result = ctx.storage.publish_message(
    channel: channel,
    data: data,
    metadata: full_metadata
  )

  # Call hook
  ctx.hooks&.on_event_sent&.call(channel, ctx.workflow_name, data)

  result
end

.receive(ctx, channel, timeout: nil, mode: :broadcast) ⇒ Message

Receive a message from a channel (blocks workflow until message arrives)

Raises:



140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
 
# File 'lib/shikibu/channels.rb', line 140

def receive(ctx, channel, timeout: nil, mode: :broadcast)
  mode_str = mode.to_s
  activity_id = ctx.generate_activity_id("receive:#{channel}")

  # For direct mode, use instance-specific channel
  actual_channel = if mode_str == ChannelMode::DIRECT
                     "#{channel}:#{ctx.instance_id}"
                   else
                     channel
                   end

  # Check cache during replay
  if ctx.replaying? && ctx.cached_result?(activity_id)
    cached = ctx.get_cached_result(activity_id)
    return build_message_from_cached(cached) if cached[:event_type] == EventType::CHANNEL_MESSAGE_RECEIVED

    # Timeout was recorded
    raise MessageTimeoutError.new(channel, timeout)
  end

  # Try to get pending message immediately
  message = try_receive_immediate(ctx, actual_channel, mode_str)
  if message
    # Call hook
    ctx.hooks&.on_event_received&.call(ctx.instance_id, actual_channel, message.data)

    record_message_received(ctx, activity_id, actual_channel, message)
    return message
  end

  # No message available, suspend workflow
  timeout_at = timeout ? Time.now + timeout : nil

  raise WaitForChannelSignal.new(
    channel: actual_channel,
    mode: mode_str == ChannelMode::DIRECT ? ChannelMode::COMPETING : mode_str,
    timeout_at: timeout_at,
    activity_id: activity_id
  )
end

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

Send a message directly to a specific workflow instance



213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
 
# File 'lib/shikibu/channels.rb', line 213

def send_to(ctx, target_instance_id, channel, data, metadata: nil)
  direct_channel = "#{channel}:#{target_instance_id}"

  full_metadata = (metadata || {}).merge(
    source_instance_id: ctx.instance_id,
    source_workflow: ctx.workflow_name,
    target_instance_id: target_instance_id,
    published_at: Time.now.iso8601
  )

  ctx.storage.publish_message(
    channel: direct_channel,
    data: data,
    metadata: full_metadata
  )
end

.subscribe(ctx, channel, mode: :broadcast) ⇒ Object

Subscribe to a channel

Raises:



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/shikibu/channels.rb', line 94

def subscribe(ctx, channel, mode: :broadcast)
  mode_str = mode.to_s
  validate_mode!(mode_str)

  # For direct mode, use instance-specific channel
  actual_channel = if mode_str == ChannelMode::DIRECT
                     "#{channel}:#{ctx.instance_id}"
                   else
                     channel
                   end

  # Determine actual mode (direct maps to competing)
  actual_mode = mode_str == ChannelMode::DIRECT ? ChannelMode::COMPETING : mode_str

  # Check for mode conflict
  existing_mode = ctx.storage.get_channel_mode(actual_channel)
  if existing_mode && existing_mode != actual_mode
    raise ChannelModeConflictError.new(channel, existing_mode, mode_str)
  end

  ctx.storage.subscribe_to_channel(
    instance_id: ctx.instance_id,
    channel: actual_channel,
    mode: actual_mode
  )

  Subscription.new(channel: actual_channel, mode: mode_str)
end

.subscriptions(ctx) ⇒ Array<Subscription>

List active subscriptions for an instance



249
250
251
252
253
254
255
256
257
258
259
260
261
 
# File 'lib/shikibu/channels.rb', line 249

def subscriptions(ctx)
  subs = ctx.storage.db[:channel_subscriptions]
            .where(instance_id: ctx.instance_id)
            .all

  subs.map do |row|
    Subscription.new(
      channel: row[:channel],
      mode: row[:mode],
      subscribed_at: row[:subscribed_at]
    )
  end
end

.try_receive(ctx, channel, mode: :broadcast) ⇒ Message?

Receive without blocking (returns nil if no message)



235
236
237
238
239
240
241
242
243
244
 
# File 'lib/shikibu/channels.rb', line 235

def try_receive(ctx, channel, mode: :broadcast)
  mode_str = mode.to_s
  actual_channel = if mode_str == ChannelMode::DIRECT
                     "#{channel}:#{ctx.instance_id}"
                   else
                     channel
                   end

  try_receive_immediate(ctx, actual_channel, mode_str)
end

.unsubscribe(ctx, channel) ⇒ Object

Unsubscribe from a channel



126
127
128
129
130
131
 
# File 'lib/shikibu/channels.rb', line 126

def unsubscribe(ctx, channel)
  ctx.storage.unsubscribe_from_channel(
    instance_id: ctx.instance_id,
    channel: channel
  )
end