Class: AWS::Flow::WorkflowClient

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

Overview

A client for a workflow execution.

Instance Attribute Summary collapse

Attributes inherited from GenericClient

#option_map

Instance Method Summary collapse

Methods inherited from GenericClient

#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:



108
109
110
111
112
113
114
115
 
# File 'lib/aws/decider/workflow_client.rb', line 108

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 



335
336
337
338
339
340
341
 
# File 'lib/aws/decider/workflow_client.rb', line 335

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

Instance Attribute Details

#domain ⇒ Object

The SWF domain used for this workflow.



90
91
92
93
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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
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
 
# File 'lib/aws/decider/workflow_client.rb', line 90

class WorkflowClient < GenericClient
  attr_accessor :domain, :options

  # Creates a new {WorkflowClient}
  #
  # @param service
  #   The SWF [Client](http://docs.aws.amazon.com/AWSRubySDK/latest/AWS/SimpleWorkflow/Client.html) to use for
  #   creating this {WorkflowClient}.
  #
  # @param domain
  #   The 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


  # @!visibility 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
  #
  # @param [Hash, StartWorkflowOptions] block
  #   A hash of {StartWorkflowOptions} to use for this workflow execution.
  #
  def start_execution(*input, &block)
    if Utilities::is_external
      self.start_external_workflow(input, &block)
    else
      self.start_internal_workflow(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}
    # @!visibility 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}
    # @!visibility 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
    execution_method = options.execution_method || @options.execution_method
    raise "You haven't specified an execution method!" if execution_method.nil?
    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}
  # @!visibility private
  def start_internal_workflow(input = NoInput.new, &block)
    get_decision_context
    options = Utilities::interpret_block_for_options(StartWorkflowOptions, block)
    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))
    options = Utilities::merge_all_options(@options, options)
    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.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}
  # @!visibility private
  def start_external_workflow(input = NoInput.new, &block)
    options = Utilities::interpret_block_for_options(StartWorkflowOptions, block)
    options = Utilities::merge_all_options(@options, options)
    @converter ||= YAMLDataConverter.new
    # Basically, we want to avoid the special "NoInput, but allow stuff like nil in"
    if ! (input.class <= NoInput || input.empty?)
      options[:input] = @converter.dump input
    end
    if @workflow_class.nil?
      execution_method = @options.execution_method
      version = @options.version
    else
      # TODO This is a nasty hack
      workflow_type = @workflow_class.workflows.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] ||= UUIDTools::UUID.random_create.to_s
    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(*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.



90
91
92
93
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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
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
 
# File 'lib/aws/decider/workflow_client.rb', line 90

class WorkflowClient < GenericClient
  attr_accessor :domain, :options

  # Creates a new {WorkflowClient}
  #
  # @param service
  #   The SWF [Client](http://docs.aws.amazon.com/AWSRubySDK/latest/AWS/SimpleWorkflow/Client.html) to use for
  #   creating this {WorkflowClient}.
  #
  # @param domain
  #   The 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


  # @!visibility 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
  #
  # @param [Hash, StartWorkflowOptions] block
  #   A hash of {StartWorkflowOptions} to use for this workflow execution.
  #
  def start_execution(*input, &block)
    if Utilities::is_external
      self.start_external_workflow(input, &block)
    else
      self.start_internal_workflow(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}
    # @!visibility 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}
    # @!visibility 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
    execution_method = options.execution_method || @options.execution_method
    raise "You haven't specified an execution method!" if execution_method.nil?
    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}
  # @!visibility private
  def start_internal_workflow(input = NoInput.new, &block)
    get_decision_context
    options = Utilities::interpret_block_for_options(StartWorkflowOptions, block)
    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))
    options = Utilities::merge_all_options(@options, options)
    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.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}
  # @!visibility private
  def start_external_workflow(input = NoInput.new, &block)
    options = Utilities::interpret_block_for_options(StartWorkflowOptions, block)
    options = Utilities::merge_all_options(@options, options)
    @converter ||= YAMLDataConverter.new
    # Basically, we want to avoid the special "NoInput, but allow stuff like nil in"
    if ! (input.class <= NoInput || input.empty?)
      options[:input] = @converter.dump input
    end
    if @workflow_class.nil?
      execution_method = @options.execution_method
      version = @options.version
    else
      # TODO This is a nasty hack
      workflow_type = @workflow_class.workflows.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] ||= UUIDTools::UUID.random_create.to_s
    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(*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

Instance Method Details

#events ⇒ Object

Gets the events for this workflow client



123
124
125
 
# File 'lib/aws/decider/workflow_client.rb', line 123

def events
  @execution.events if @execution
end

#get_decision_context ⇒ Object

Gets the current DecisionContext for the workflow client.



130
131
132
133
 
# File 'lib/aws/decider/workflow_client.rb', line 130

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

#is_execution_method(method_name) ⇒ Object 



331
332
333
 
# File 'lib/aws/decider/workflow_client.rb', line 331

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 



342
343
344
345
346
347
348
349
 
# File 'lib/aws/decider/workflow_client.rb', line 342

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_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.



165
166
167
168
169
170
171
 
# File 'lib/aws/decider/workflow_client.rb', line 165

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:



144
145
146
147
148
149
150
 
# File 'lib/aws/decider/workflow_client.rb', line 144

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