Class: AWS::Flow::WorkflowClient

Inherits:
GenericClient show all
Defined in:
lib/aws/decider/workflow_client.rb

Overview

Represents a client for a workflow execution.

Instance Attribute Summary collapse

Attributes inherited from GenericClient

#option_map

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from GenericClient

#_retry, #_retry_with_options, #bail_if_external, #decision_context, #exponential_retry, #reconfigure, #retry, #send_async, #with_opts

Constructor Details

#initialize(service, domain, workflow_class, options) ⇒ WorkflowClient

Creates a new AWS::Flow::WorkflowClient.

Parameters:



155
156
157
158
159
160
161
162
 
# File 'lib/aws/decider/workflow_client.rb', line 155

def initialize(service, domain, workflow_class, options)
  @service = service
  @domain = domain
  @workflow_class = workflow_class
  @options = options
  @failure_map = {}
  super
end

Dynamic Method Handling

This class handles dynamic methods through the method_missing method

#method_missing(method_name, *args, &block) ⇒ Object 



387
388
389
390
391
392
393
 
# File 'lib/aws/decider/workflow_client.rb', line 387

def method_missing(method_name, *args, &block)
  if is_execution_method(method_name)
    start_execution_method(method_name, *args, &block)
  else
    super(method_name, *args, &block)
  end
end

Instance Attribute Details

#domain ⇒ Object

The Amazon SWF domain used for this workflow.



137
138
139
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
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
 
# File 'lib/aws/decider/workflow_client.rb', line 137

class WorkflowClient < GenericClient
  attr_accessor :domain, :options

  # Creates a new {WorkflowClient}.
  #
  # @param service
  #   The Amazon SWF [Client](http://docs.aws.amazon.com/AWSRubySDK/latest/AWS/SimpleWorkflow/Client.html) to use for
  #   creating this {WorkflowClient}.
  #
  # @param domain
  #   The Amazon SWF [Domain](http://docs.aws.amazon.com/AWSRubySDK/latest/AWS/SimpleWorkflow/Domain.html) in which to
  #   start the workflow execution.
  #
  # @param workflow_class
  #
  # @param [Hash, WorkflowOptions] options
  #   Workflow options for this client.
  #
  def initialize(service, domain, workflow_class, options)
    @service = service
    @domain = domain
    @workflow_class = workflow_class
    @options = options
    @failure_map = {}
    super
  end


  # @api private
  def self.default_option_class; WorkflowOptions; end

  # Gets the events for this workflow client.
  #
  def events
    @execution.events if @execution
  end


  # Gets the current {DecisionContext} for the workflow client.
  #
  def get_decision_context
    @decision_context ||= FlowFiber.current[:decision_context]
    @decision_helper ||= @decision_context.decision_helper
  end


  # Begins executing this workflow.
  #
  # @param input
  #   Input to provide to the workflow.
  #
  # @param [Hash, StartWorkflowOptions] block
  #   A hash of {StartWorkflowOptions} to use for this workflow execution.
  #
  def start_execution(*input, &block)
    start_execution_method(nil, *input, &block)
  end

  def start_execution_method(method_name, *input, &block)
    if Utilities::is_external
      self.start_external_workflow(method_name, input, &block)
    else
      self.start_internal_workflow(method_name, input, &block)
    end
  end

  # Records a `WorkflowExecutionSignaled` event in the workflow execution history and creates a decision task for
  # the workflow execution. The event is recorded with the specified user-defined signal name and input (if
  # provided).
  #
  # @param signal_name
  #   The user-defined name of the signal.
  #
  # @param workflow_execution
  #   The workflow execution to signal.
  #
  # @param [Hash, SignalWorkflowOptions] block
  #   A block of {SignalWorkflowOptions} for the `WorkflowExecutionSignaled` event.
  #
  def signal_workflow_execution(signal_name = nil, workflow_execution = nil, &block)
    if Utilities::is_external
      self.signal_external_workflow(signal_name, workflow_execution, &block)
    else
      self.signal_internal_workflow(signal_name, workflow_execution, &block)
    end
  end


    # Called by {#signal_workflow_execution}.
    # @api private
    def signal_external_workflow(signal_name, workflow_execution, &block)
    options = Utilities::interpret_block_for_options(SignalWorkflowOptions, block)
    options.signal_name ||= signal_name
    if workflow_execution
      options.domain ||= workflow_execution.domain.name.to_s
      options.workflow_id ||= workflow_execution.workflow_id.to_s
    end
    @service.signal_workflow_execution(options.get_full_options)
  end

    # Called by {#signal_workflow_execution}.
    # @api private
  def signal_internal_workflow(signal_name, workflow_execution, &block)
    get_decision_context
    options = Utilities::interpret_block_for_options(SignalWorkflowOptions, block)
    # Unpack the workflow execution from the future
    workflow_execution = workflow_execution.workflow_execution if workflow_execution.respond_to? :workflow_execution
    options.signal_name ||= signal_name.to_s
    options.workflow_id ||= workflow_execution.workflow_id.get.to_s
    Utilities::merge_all_options(options)
    open_request = OpenRequestInfo.new
    decision_id = @decision_helper.get_next_id(:Signal)
    options.control ||= decision_id
    external_task do |external|
      external.initiate_task do |handle|
        @decision_helper[decision_id] = SignalDecisionStateMachine.new(decision_id, options)
        open_request.completion_handle = handle
        @decision_helper.scheduled_signals[decision_id] = open_request
      end
      external.cancellation_handler do |handle, cause|
        @decision_helper[decision_id].consume(:cancel)
        open_request = @decision_helper.scheduled_signal.delete(decision_id)
        raise "Signal #{decision_id} wasn't scheduled" unless open_request
        handle.complete
      end
    end
    return open_request.result
  end


  # Called by {#start_execution}.
  # @api private
  def start_internal_workflow(method_name, input = NoInput.new, &block)
    get_decision_context
    options = Utilities::interpret_block_for_options(StartWorkflowOptions, block)
    client_options = Utilities::client_options_from_method_name(method_name, @options)
    options = Utilities::merge_all_options(client_options, options)

    workflow_id_future, run_id_future = Future.new, Future.new
    minimal_domain = MinimalDomain.new(@domain.name.to_s)
    output = WorkflowFuture.new(AWS::Flow::MinimalWorkflowExecution.new(minimal_domain, workflow_id_future, run_id_future))
    new_options = StartWorkflowOptions.new(options)
    open_request = OpenRequestInfo.new
    workflow_id = new_options.workflow_id
    run_id = @decision_context.workflow_context.decision_task.workflow_execution.run_id
    workflow_id ||= @decision_helper.get_next_id(run_id.to_s + ":")
    workflow_id_future.set(workflow_id)
    error_handler do |t|
      t.begin do
        @data_converter = new_options.data_converter
        input = @data_converter.dump input unless input.empty?
        attributes = {}
        new_options.input ||= input unless input.empty?
        if @workflow_class != nil && new_options.execution_method.nil?
          new_options.execution_method = @workflow_class.entry_point
        end
        raise "Can't find an execution method for workflow #{@workflow_class}" if new_options.execution_method.nil?

        attributes[:options] = new_options
        attributes[:workflow_id] = workflow_id
        # TODO Use ChildWorkflowOptions
        attributes[:tag_list] = []

        external_task do |external|
          external.initiate_task do |handle|
            open_request.completion_handle = handle
            open_request.run_id = run_id_future
            open_request.description = output.workflow_execution
            @decision_helper.scheduled_external_workflows[workflow_id.to_s] = open_request
            @decision_helper[workflow_id.to_s] = ChildWorkflowDecisionStateMachine.new(workflow_id, attributes)
          end

          external.cancellation_handler do |handle, cause|
            state_machine = @decision_helper[workflow_id.to_s]
            if state_machine.current_state == :created
              open_request = @decision_helper.scheduled_external_workflows.delete(workflow_id)
              open_request.completion_handle.complete
            end
            state_machine.consume(:cancel)
          end
        end

        t.rescue(Exception) do |error|
          if error.is_a? ChildWorkflowFailedException
            details = @data_converter.load(error.details)
            error.details = details
            error.cause = details
          end
          @failure_map[workflow_id.to_s] = error
        end
        t.ensure do
          result = @data_converter.load open_request.result
          output.set(result)
          raise @failure_map[workflow_id.to_s] if @failure_map[workflow_id.to_s] && new_options.return_on_start
        end
      end
    end
    return output if new_options.return_on_start
    output.get
    this_failure = @failure_map[workflow_id.to_s]
    raise this_failure if this_failure
    return output.get
  end


  # Called by {#start_execution}.
  # @api private
  def start_external_workflow(method_name, input = NoInput.new, &block)
    options = Utilities::interpret_block_for_options(StartWorkflowOptions, block)
    client_options = Utilities::client_options_from_method_name(method_name, @options)
    options = Utilities::merge_all_options(client_options, options)

    @data_converter = options[:data_converter]
    # Basically, we want to avoid the special "NoInput, but allow stuff like nil in"
    if ! (input.class <= NoInput || input.empty?)
      options[:input] = @data_converter.dump input
    end
    if @workflow_class.nil?
      execution_method = @options.execution_method
      version = @options.version
    else
      workflow_type = method_name.nil? ? @workflow_class.workflows.first : @workflow_class.workflows.select { |x| x.options.execution_method.to_sym == method_name }.first
      execution_method = workflow_type.options.execution_method
      version = workflow_type.version
    end
    version = options[:version] ? options[:version] : version
    execution_method = options[:execution_method] ? options[:execution_method] : execution_method
    raise "Can't find an execution method for workflow #{workflow_class}" if execution_method.nil?
    # TODO A real workflowtype function
    workflow_name = @options.workflow_name || @options.prefix_name
    workflow_type_name = workflow_name.to_s + "." + execution_method.to_s

    task_list = options[:task_list]
    options[:task_list] = { :name => task_list } if options[:task_list]
    options[:workflow_id] ||= SecureRandom.uuid
    options[:domain] = @domain.name
    options[:workflow_type] = {
      :name => workflow_type_name.to_s,
      :version => version.to_s
    }
    [:prefix_name, :workflow_name, :version, :execution_method, :data_converter].each {|key| options.delete(key)}
    run_id = @service.start_workflow_execution(options)["runId"]
    this_workflow = @domain.workflow_executions.at(options[:workflow_id], run_id)
    this_workflow
  end

  def is_execution_method(method_name)
    (@workflow_class.workflows.map(&:options).map(&:execution_method).map(&:to_sym).include? method_name) || method_name == @workflow_class.entry_point
  end

  def method_missing(method_name, *args, &block)
    if is_execution_method(method_name)
      start_execution_method(method_name, *args, &block)
    else
      super(method_name, *args, &block)
    end
  end
  def request_cancel_workflow_execution(future)
    workflow_execution = future.workflow_execution
    run_id = workflow_execution.run_id.get
    workflow_id = workflow_execution.workflow_id.get
    state_machine = @decision_helper[workflow_id]
    state_machine.run_id = run_id
    state_machine.consume(:cancel)
  end
end

#options ⇒ Object

Workflow options for this client.



137
138
139
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
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
 
# File 'lib/aws/decider/workflow_client.rb', line 137

class WorkflowClient < GenericClient
  attr_accessor :domain, :options

  # Creates a new {WorkflowClient}.
  #
  # @param service
  #   The Amazon SWF [Client](http://docs.aws.amazon.com/AWSRubySDK/latest/AWS/SimpleWorkflow/Client.html) to use for
  #   creating this {WorkflowClient}.
  #
  # @param domain
  #   The Amazon SWF [Domain](http://docs.aws.amazon.com/AWSRubySDK/latest/AWS/SimpleWorkflow/Domain.html) in which to
  #   start the workflow execution.
  #
  # @param workflow_class
  #
  # @param [Hash, WorkflowOptions] options
  #   Workflow options for this client.
  #
  def initialize(service, domain, workflow_class, options)
    @service = service
    @domain = domain
    @workflow_class = workflow_class
    @options = options
    @failure_map = {}
    super
  end


  # @api private
  def self.default_option_class; WorkflowOptions; end

  # Gets the events for this workflow client.
  #
  def events
    @execution.events if @execution
  end


  # Gets the current {DecisionContext} for the workflow client.
  #
  def get_decision_context
    @decision_context ||= FlowFiber.current[:decision_context]
    @decision_helper ||= @decision_context.decision_helper
  end


  # Begins executing this workflow.
  #
  # @param input
  #   Input to provide to the workflow.
  #
  # @param [Hash, StartWorkflowOptions] block
  #   A hash of {StartWorkflowOptions} to use for this workflow execution.
  #
  def start_execution(*input, &block)
    start_execution_method(nil, *input, &block)
  end

  def start_execution_method(method_name, *input, &block)
    if Utilities::is_external
      self.start_external_workflow(method_name, input, &block)
    else
      self.start_internal_workflow(method_name, input, &block)
    end
  end

  # Records a `WorkflowExecutionSignaled` event in the workflow execution history and creates a decision task for
  # the workflow execution. The event is recorded with the specified user-defined signal name and input (if
  # provided).
  #
  # @param signal_name
  #   The user-defined name of the signal.
  #
  # @param workflow_execution
  #   The workflow execution to signal.
  #
  # @param [Hash, SignalWorkflowOptions] block
  #   A block of {SignalWorkflowOptions} for the `WorkflowExecutionSignaled` event.
  #
  def signal_workflow_execution(signal_name = nil, workflow_execution = nil, &block)
    if Utilities::is_external
      self.signal_external_workflow(signal_name, workflow_execution, &block)
    else
      self.signal_internal_workflow(signal_name, workflow_execution, &block)
    end
  end


    # Called by {#signal_workflow_execution}.
    # @api private
    def signal_external_workflow(signal_name, workflow_execution, &block)
    options = Utilities::interpret_block_for_options(SignalWorkflowOptions, block)
    options.signal_name ||= signal_name
    if workflow_execution
      options.domain ||= workflow_execution.domain.name.to_s
      options.workflow_id ||= workflow_execution.workflow_id.to_s
    end
    @service.signal_workflow_execution(options.get_full_options)
  end

    # Called by {#signal_workflow_execution}.
    # @api private
  def signal_internal_workflow(signal_name, workflow_execution, &block)
    get_decision_context
    options = Utilities::interpret_block_for_options(SignalWorkflowOptions, block)
    # Unpack the workflow execution from the future
    workflow_execution = workflow_execution.workflow_execution if workflow_execution.respond_to? :workflow_execution
    options.signal_name ||= signal_name.to_s
    options.workflow_id ||= workflow_execution.workflow_id.get.to_s
    Utilities::merge_all_options(options)
    open_request = OpenRequestInfo.new
    decision_id = @decision_helper.get_next_id(:Signal)
    options.control ||= decision_id
    external_task do |external|
      external.initiate_task do |handle|
        @decision_helper[decision_id] = SignalDecisionStateMachine.new(decision_id, options)
        open_request.completion_handle = handle
        @decision_helper.scheduled_signals[decision_id] = open_request
      end
      external.cancellation_handler do |handle, cause|
        @decision_helper[decision_id].consume(:cancel)
        open_request = @decision_helper.scheduled_signal.delete(decision_id)
        raise "Signal #{decision_id} wasn't scheduled" unless open_request
        handle.complete
      end
    end
    return open_request.result
  end


  # Called by {#start_execution}.
  # @api private
  def start_internal_workflow(method_name, input = NoInput.new, &block)
    get_decision_context
    options = Utilities::interpret_block_for_options(StartWorkflowOptions, block)
    client_options = Utilities::client_options_from_method_name(method_name, @options)
    options = Utilities::merge_all_options(client_options, options)

    workflow_id_future, run_id_future = Future.new, Future.new
    minimal_domain = MinimalDomain.new(@domain.name.to_s)
    output = WorkflowFuture.new(AWS::Flow::MinimalWorkflowExecution.new(minimal_domain, workflow_id_future, run_id_future))
    new_options = StartWorkflowOptions.new(options)
    open_request = OpenRequestInfo.new
    workflow_id = new_options.workflow_id
    run_id = @decision_context.workflow_context.decision_task.workflow_execution.run_id
    workflow_id ||= @decision_helper.get_next_id(run_id.to_s + ":")
    workflow_id_future.set(workflow_id)
    error_handler do |t|
      t.begin do
        @data_converter = new_options.data_converter
        input = @data_converter.dump input unless input.empty?
        attributes = {}
        new_options.input ||= input unless input.empty?
        if @workflow_class != nil && new_options.execution_method.nil?
          new_options.execution_method = @workflow_class.entry_point
        end
        raise "Can't find an execution method for workflow #{@workflow_class}" if new_options.execution_method.nil?

        attributes[:options] = new_options
        attributes[:workflow_id] = workflow_id
        # TODO Use ChildWorkflowOptions
        attributes[:tag_list] = []

        external_task do |external|
          external.initiate_task do |handle|
            open_request.completion_handle = handle
            open_request.run_id = run_id_future
            open_request.description = output.workflow_execution
            @decision_helper.scheduled_external_workflows[workflow_id.to_s] = open_request
            @decision_helper[workflow_id.to_s] = ChildWorkflowDecisionStateMachine.new(workflow_id, attributes)
          end

          external.cancellation_handler do |handle, cause|
            state_machine = @decision_helper[workflow_id.to_s]
            if state_machine.current_state == :created
              open_request = @decision_helper.scheduled_external_workflows.delete(workflow_id)
              open_request.completion_handle.complete
            end
            state_machine.consume(:cancel)
          end
        end

        t.rescue(Exception) do |error|
          if error.is_a? ChildWorkflowFailedException
            details = @data_converter.load(error.details)
            error.details = details
            error.cause = details
          end
          @failure_map[workflow_id.to_s] = error
        end
        t.ensure do
          result = @data_converter.load open_request.result
          output.set(result)
          raise @failure_map[workflow_id.to_s] if @failure_map[workflow_id.to_s] && new_options.return_on_start
        end
      end
    end
    return output if new_options.return_on_start
    output.get
    this_failure = @failure_map[workflow_id.to_s]
    raise this_failure if this_failure
    return output.get
  end


  # Called by {#start_execution}.
  # @api private
  def start_external_workflow(method_name, input = NoInput.new, &block)
    options = Utilities::interpret_block_for_options(StartWorkflowOptions, block)
    client_options = Utilities::client_options_from_method_name(method_name, @options)
    options = Utilities::merge_all_options(client_options, options)

    @data_converter = options[:data_converter]
    # Basically, we want to avoid the special "NoInput, but allow stuff like nil in"
    if ! (input.class <= NoInput || input.empty?)
      options[:input] = @data_converter.dump input
    end
    if @workflow_class.nil?
      execution_method = @options.execution_method
      version = @options.version
    else
      workflow_type = method_name.nil? ? @workflow_class.workflows.first : @workflow_class.workflows.select { |x| x.options.execution_method.to_sym == method_name }.first
      execution_method = workflow_type.options.execution_method
      version = workflow_type.version
    end
    version = options[:version] ? options[:version] : version
    execution_method = options[:execution_method] ? options[:execution_method] : execution_method
    raise "Can't find an execution method for workflow #{workflow_class}" if execution_method.nil?
    # TODO A real workflowtype function
    workflow_name = @options.workflow_name || @options.prefix_name
    workflow_type_name = workflow_name.to_s + "." + execution_method.to_s

    task_list = options[:task_list]
    options[:task_list] = { :name => task_list } if options[:task_list]
    options[:workflow_id] ||= SecureRandom.uuid
    options[:domain] = @domain.name
    options[:workflow_type] = {
      :name => workflow_type_name.to_s,
      :version => version.to_s
    }
    [:prefix_name, :workflow_name, :version, :execution_method, :data_converter].each {|key| options.delete(key)}
    run_id = @service.start_workflow_execution(options)["runId"]
    this_workflow = @domain.workflow_executions.at(options[:workflow_id], run_id)
    this_workflow
  end

  def is_execution_method(method_name)
    (@workflow_class.workflows.map(&:options).map(&:execution_method).map(&:to_sym).include? method_name) || method_name == @workflow_class.entry_point
  end

  def method_missing(method_name, *args, &block)
    if is_execution_method(method_name)
      start_execution_method(method_name, *args, &block)
    else
      super(method_name, *args, &block)
    end
  end
  def request_cancel_workflow_execution(future)
    workflow_execution = future.workflow_execution
    run_id = workflow_execution.run_id.get
    workflow_id = workflow_execution.workflow_id.get
    state_machine = @decision_helper[workflow_id]
    state_machine.run_id = run_id
    state_machine.consume(:cancel)
  end
end

Class Method Details

.default_option_class ⇒ 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. 



166
 
# File 'lib/aws/decider/workflow_client.rb', line 166

def self.default_option_class; WorkflowOptions; end

Instance Method Details

#events ⇒ Object

Gets the events for this workflow client.



170
171
172
 
# File 'lib/aws/decider/workflow_client.rb', line 170

def events
  @execution.events if @execution
end

#get_decision_context ⇒ Object

Gets the current DecisionContext for the workflow client.



177
178
179
180
 
# File 'lib/aws/decider/workflow_client.rb', line 177

def get_decision_context
  @decision_context ||= FlowFiber.current[:decision_context]
  @decision_helper ||= @decision_context.decision_helper
end

#is_execution_method(method_name) ⇒ Object 



383
384
385
 
# File 'lib/aws/decider/workflow_client.rb', line 383

def is_execution_method(method_name)
  (@workflow_class.workflows.map(&:options).map(&:execution_method).map(&:to_sym).include? method_name) || method_name == @workflow_class.entry_point
end

#request_cancel_workflow_execution(future) ⇒ Object 



394
395
396
397
398
399
400
401
 
# File 'lib/aws/decider/workflow_client.rb', line 394

def request_cancel_workflow_execution(future)
  workflow_execution = future.workflow_execution
  run_id = workflow_execution.run_id.get
  workflow_id = workflow_execution.workflow_id.get
  state_machine = @decision_helper[workflow_id]
  state_machine.run_id = run_id
  state_machine.consume(:cancel)
end

#signal_external_workflow(signal_name, workflow_execution, &block) ⇒ 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 by #signal_workflow_execution.



227
228
229
230
231
232
233
234
235
 
# File 'lib/aws/decider/workflow_client.rb', line 227

def signal_external_workflow(signal_name, workflow_execution, &block)
  options = Utilities::interpret_block_for_options(SignalWorkflowOptions, block)
  options.signal_name ||= signal_name
  if workflow_execution
    options.domain ||= workflow_execution.domain.name.to_s
    options.workflow_id ||= workflow_execution.workflow_id.to_s
  end
  @service.signal_workflow_execution(options.get_full_options)
end

#signal_internal_workflow(signal_name, workflow_execution, &block) ⇒ 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 by #signal_workflow_execution.



239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
 
# File 'lib/aws/decider/workflow_client.rb', line 239

def signal_internal_workflow(signal_name, workflow_execution, &block)
  get_decision_context
  options = Utilities::interpret_block_for_options(SignalWorkflowOptions, block)
  # Unpack the workflow execution from the future
  workflow_execution = workflow_execution.workflow_execution if workflow_execution.respond_to? :workflow_execution
  options.signal_name ||= signal_name.to_s
  options.workflow_id ||= workflow_execution.workflow_id.get.to_s
  Utilities::merge_all_options(options)
  open_request = OpenRequestInfo.new
  decision_id = @decision_helper.get_next_id(:Signal)
  options.control ||= decision_id
  external_task do |external|
    external.initiate_task do |handle|
      @decision_helper[decision_id] = SignalDecisionStateMachine.new(decision_id, options)
      open_request.completion_handle = handle
      @decision_helper.scheduled_signals[decision_id] = open_request
    end
    external.cancellation_handler do |handle, cause|
      @decision_helper[decision_id].consume(:cancel)
      open_request = @decision_helper.scheduled_signal.delete(decision_id)
      raise "Signal #{decision_id} wasn't scheduled" unless open_request
      handle.complete
    end
  end
  return open_request.result
end

#signal_workflow_execution(signal_name = nil, workflow_execution = nil, &block) ⇒ Object

Records a ‘WorkflowExecutionSignaled` event in the workflow execution history and creates a decision task for the workflow execution. The event is recorded with the specified user-defined signal name and input (if provided).

Parameters:

  • signal_name (defaults to: nil) —

    The user-defined name of the signal.

  • workflow_execution (defaults to: nil) —

    The workflow execution to signal.

  • block (Hash, SignalWorkflowOptions) —

    A block of SignalWorkflowOptions for the ‘WorkflowExecutionSignaled` event.



216
217
218
219
220
221
222
 
# File 'lib/aws/decider/workflow_client.rb', line 216

def signal_workflow_execution(signal_name = nil, workflow_execution = nil, &block)
  if Utilities::is_external
    self.signal_external_workflow(signal_name, workflow_execution, &block)
  else
    self.signal_internal_workflow(signal_name, workflow_execution, &block)
  end
end

#start_execution(*input, &block) ⇒ Object

Begins executing this workflow.

Parameters:



191
192
193
 
# File 'lib/aws/decider/workflow_client.rb', line 191

def start_execution(*input, &block)
  start_execution_method(nil, *input, &block)
end

#start_execution_method(method_name, *input, &block) ⇒ Object 



195
196
197
198
199
200
201
 
# File 'lib/aws/decider/workflow_client.rb', line 195

def start_execution_method(method_name, *input, &block)
  if Utilities::is_external
    self.start_external_workflow(method_name, input, &block)
  else
    self.start_internal_workflow(method_name, input, &block)
  end
end

#start_external_workflow(method_name, input = NoInput.new, &block) ⇒ 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 by #start_execution.



344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
 
# File 'lib/aws/decider/workflow_client.rb', line 344

def start_external_workflow(method_name, input = NoInput.new, &block)
  options = Utilities::interpret_block_for_options(StartWorkflowOptions, block)
  client_options = Utilities::client_options_from_method_name(method_name, @options)
  options = Utilities::merge_all_options(client_options, options)

  @data_converter = options[:data_converter]
  # Basically, we want to avoid the special "NoInput, but allow stuff like nil in"
  if ! (input.class <= NoInput || input.empty?)
    options[:input] = @data_converter.dump input
  end
  if @workflow_class.nil?
    execution_method = @options.execution_method
    version = @options.version
  else
    workflow_type = method_name.nil? ? @workflow_class.workflows.first : @workflow_class.workflows.select { |x| x.options.execution_method.to_sym == method_name }.first
    execution_method = workflow_type.options.execution_method
    version = workflow_type.version
  end
  version = options[:version] ? options[:version] : version
  execution_method = options[:execution_method] ? options[:execution_method] : execution_method
  raise "Can't find an execution method for workflow #{workflow_class}" if execution_method.nil?
  # TODO A real workflowtype function
  workflow_name = @options.workflow_name || @options.prefix_name
  workflow_type_name = workflow_name.to_s + "." + execution_method.to_s

  task_list = options[:task_list]
  options[:task_list] = { :name => task_list } if options[:task_list]
  options[:workflow_id] ||= SecureRandom.uuid
  options[:domain] = @domain.name
  options[:workflow_type] = {
    :name => workflow_type_name.to_s,
    :version => version.to_s
  }
  [:prefix_name, :workflow_name, :version, :execution_method, :data_converter].each {|key| options.delete(key)}
  run_id = @service.start_workflow_execution(options)["runId"]
  this_workflow = @domain.workflow_executions.at(options[:workflow_id], run_id)
  this_workflow
end

#start_internal_workflow(method_name, input = NoInput.new, &block) ⇒ 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 by #start_execution.



269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
 
# File 'lib/aws/decider/workflow_client.rb', line 269

def start_internal_workflow(method_name, input = NoInput.new, &block)
  get_decision_context
  options = Utilities::interpret_block_for_options(StartWorkflowOptions, block)
  client_options = Utilities::client_options_from_method_name(method_name, @options)
  options = Utilities::merge_all_options(client_options, options)

  workflow_id_future, run_id_future = Future.new, Future.new
  minimal_domain = MinimalDomain.new(@domain.name.to_s)
  output = WorkflowFuture.new(AWS::Flow::MinimalWorkflowExecution.new(minimal_domain, workflow_id_future, run_id_future))
  new_options = StartWorkflowOptions.new(options)
  open_request = OpenRequestInfo.new
  workflow_id = new_options.workflow_id
  run_id = @decision_context.workflow_context.decision_task.workflow_execution.run_id
  workflow_id ||= @decision_helper.get_next_id(run_id.to_s + ":")
  workflow_id_future.set(workflow_id)
  error_handler do |t|
    t.begin do
      @data_converter = new_options.data_converter
      input = @data_converter.dump input unless input.empty?
      attributes = {}
      new_options.input ||= input unless input.empty?
      if @workflow_class != nil && new_options.execution_method.nil?
        new_options.execution_method = @workflow_class.entry_point
      end
      raise "Can't find an execution method for workflow #{@workflow_class}" if new_options.execution_method.nil?

      attributes[:options] = new_options
      attributes[:workflow_id] = workflow_id
      # TODO Use ChildWorkflowOptions
      attributes[:tag_list] = []

      external_task do |external|
        external.initiate_task do |handle|
          open_request.completion_handle = handle
          open_request.run_id = run_id_future
          open_request.description = output.workflow_execution
          @decision_helper.scheduled_external_workflows[workflow_id.to_s] = open_request
          @decision_helper[workflow_id.to_s] = ChildWorkflowDecisionStateMachine.new(workflow_id, attributes)
        end

        external.cancellation_handler do |handle, cause|
          state_machine = @decision_helper[workflow_id.to_s]
          if state_machine.current_state == :created
            open_request = @decision_helper.scheduled_external_workflows.delete(workflow_id)
            open_request.completion_handle.complete
          end
          state_machine.consume(:cancel)
        end
      end

      t.rescue(Exception) do |error|
        if error.is_a? ChildWorkflowFailedException
          details = @data_converter.load(error.details)
          error.details = details
          error.cause = details
        end
        @failure_map[workflow_id.to_s] = error
      end
      t.ensure do
        result = @data_converter.load open_request.result
        output.set(result)
        raise @failure_map[workflow_id.to_s] if @failure_map[workflow_id.to_s] && new_options.return_on_start
      end
    end
  end
  return output if new_options.return_on_start
  output.get
  this_failure = @failure_map[workflow_id.to_s]
  raise this_failure if this_failure
  return output.get
end