Module: AWS::Flow::Workflows
Overview
Types and methods related to workflow execution. Extend this to implement a workflow decider.
Defined Under Namespace
Modules: InstanceMethods
Instance Attribute Summary collapse
-
#options ⇒ Object
Sets or returns the WorkflowOptions for this decider.
-
#version(arg = nil) ⇒ Object
deprecated
private
Deprecated.
Set the version in the WorkflowOptions passed in to the #workflow method.
Attributes included from Utilities::UpwardLookups
Class Method Summary collapse
Instance Method Summary collapse
- #_options ⇒ Object private
-
#activity_client(name, &block) ⇒ Object
Sets the activity client.
-
#entry_point(input = nil) ⇒ Object
deprecated
private
Deprecated.
Set the entry point with #workflow instead.
-
#get_state_method(get_state_method = nil, options = {}) ⇒ MethodPair
A MethodPair object.
- #look_upwards(variable) ⇒ Object
-
#signal(method_name, options = {}) ⇒ Object
Defines a signal for the workflow.
-
#signals ⇒ Hash
A hash of string(SignalName) => MethodPair(method, signalConverter) objects.
-
#workflow(*workflow_names, &block) ⇒ Object
Defines a new workflow.
Methods included from Utilities::UpwardLookups
held_properties, properties, property
Instance Attribute Details
#options ⇒ Object
Sets or returns the AWS::Flow::WorkflowOptions for this decider.
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 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 |
# File 'lib/aws/decider/decider.rb', line 245 module Workflows attr_accessor :version extend Utilities::UpwardLookups @precursors ||= [] def look_upwards(variable) precursors = self.ancestors.dup precursors.delete(self) results = precursors.map { |x| x.send(variable) if x.methods.map(&:to_sym).include? variable }.compact.flatten.uniq end property(:workflows, []) @workflows = [] def self.extended(base) base.send :include, InstanceMethods end # @deprecated Set the entry point with {Workflows#workflow} instead. # # @api private def entry_point(input=nil) if input @entry_point = input workflow_type = WorkflowType.new(self.to_s + "." + input.to_s, nil, WorkflowRegistrationOptions.new(:execution_method => input)) self.workflows.each { |workflow| workflow.name = self.to_s + "." + input.to_s } self.workflows.each do |workflow| workflow. = WorkflowRegistrationOptions.new(:execution_method => input) end self.workflows = self.workflows << workflow_type end return @entry_point if @entry_point raise "You must set an entry point on the workflow definition" end # @deprecated Set the version in the {WorkflowOptions} passed in to the {#workflow} method. # @api private def version(arg = nil) if arg self.workflows.each { |workflow| workflow.version = arg } self.workflows = self.workflows << WorkflowType.new(nil, arg, WorkflowOptions.new) end return @version end # Sets the activity client. # # @param name # Sets the client name for the activity client. # # @param block # A block of {ActivityOptions} for the activity client. # def activity_client(name, &block) = Utilities::(ActivityOptions, block) # TODO: Make sure this works for dynamic stuff begin activity_class = get_const(.prefix_name) rescue Exception => e #pass end = {} if activity_class values = activity_class.activities.map{|x| [x.name.split(".").last.to_sym, x.]} = Hash[*values.flatten] end # define_method(name) do # return @client if @client # @client ||= activity_class.activity_client.new(@decision_helper, options) # @client.decision_context = @decision_context # @client # end # else client_name = "@client_#{name}" define_method(name) do return instance_variable_get(client_name) if instance_variable_get(client_name) @decision_context ||= Fiber.current[:decision_context] @decision_helper ||= @decision_context.decision_helper @decision_helper. = instance_variable_set(client_name, GenericActivityClient.new(@decision_helper, )) instance_variable_get(client_name) end instance_variable_get(client_name) end # @api private def ; self.workflows; end # Defines a new workflow. # # @param workflow_names # The entry points (methods) that starts the workflow. # # @param block # A block of {WorflowRegistrationOptions} for the workflow. # def workflow(*workflow_names, &block) workflow_names.each do |workflow_name| = Utilities::(WorkflowRegistrationOptions, block) .execution_method = workflow_name prefix_name = .prefix_name || self.to_s workflow_type = WorkflowType.new(prefix_name.to_s + "." + workflow_name.to_s, .version, ) self.workflows = self.workflows << workflow_type end end # @return [MethodPair] # A {MethodPair} object. # def get_state_method(get_state_method = nil, = {}) data_converter = [:data_converter] @get_state_method = MethodPair.new(get_state_method, data_converter) unless get_state_method.nil? @get_state_method end # Defines a signal for the workflow. # # @param method_name # The signal method for the workflow. # # @param [SignalWorkflowOptions] options # The {SignalWorkflowOptions} for this signal. # def signal(method_name , = {}) data_converter = [:data_converter] signal_name = [:signal_name] signal_name ||= method_name.to_s data_converter ||= FlowConstants.default_data_converter @signals ||= {} @signals[signal_name] = MethodPair.new(method_name, data_converter) @signals end # @return [Hash] # A hash of string(SignalName) => MethodPair(method, signalConverter) objects def signals @signals end # Instance methods for {DecisionContext}. module InstanceMethods # Returns the {DecisionContext} instance. # @return [DecisionContext] # The {DecisionContext} instance. def decision_context FlowFiber.current[:decision_context] end # Returns the workflow ID. # # @return # The workflow ID. # def workflow_id self.decision_context.workflow_context.decision_task.workflow_execution.workflow_id end # Returns the decision helper for the decision context. This should be an instance of {DecisionHelper} or a # class derived from it. def run_id self.decision_context.workflow_context.decision_task.workflow_execution.run_id end def decision_helper FlowFiber.current[:decision_context].decision_helper end # Sets the activity client for this decision context. # # @param name # The name of the activity client. # # @param block # A block of {ActivityOptions} for the activity client. # def activity_client(name=nil, &block) = Utilities::(ActivityOptions, block) begin activity_class = get_const(.prefix_name) rescue Exception => e #pass end = {} if activity_class values = activity_class.activities.map{|x| [x.name.split(".").last.to_sym, x.]} = Hash[*values.flatten] end client = GenericActivityClient.new(self.decision_helper, ) self.class.send(:define_method, name) { client } if ! name.nil? client end # Creates a timer on the workflow that executes the supplied block after a specified delay. # # @param delay_seconds # The number of seconds to delay before executing the block. # # @param block # The block to execute when the timer expires. # def create_timer(delay_seconds, &block) self.decision_context.workflow_clock.create_timer(delay_seconds, block) end # Creates an asynchronous timer on the workflow that executes the supplied block after a specified delay. # # @param (see #create_timer) # # @deprecated # Use {#create_timer_async} instead. # # @api private def async_create_timer(delay_seconds, &block) task { self.decision_context.workflow_clock.create_timer(delay_seconds, block) } end # Creates an asynchronous timer on the workflow that executes the supplied block after a specified delay. # # @param (see #create_timer) # def create_timer_async(delay_seconds, &block) task { self.decision_context.workflow_clock.create_timer(delay_seconds, block) } end # Restarts the workflow as a new workflow execution. # # @param args # Arguments for this workflow execution, in JSON format. # # @param [ContinueAsNewOptions] block # The {ContinueAsNewOptions} for this workflow execution. # def continue_as_new(*args, &block) = Utilities::(ContinueAsNewOptions, block) @data_converter ||= YAMLDataConverter.new if ! args.empty? input = @data_converter.dump args .input = input end known_workflows = self.class.workflows # If there is only one workflow, we can unambiguously say that we should use that one. if known_workflows.length == 1 .precursors << known_workflows.first. end # If we can find a name that matches, use that one. if .execution_method matching_option = self.class.workflows.map(&:options).find {|x| x.execution_method == .execution_method } .precursors << matching_option unless matching_option.nil? end self.decision_context.workflow_context. = end end end |
#version(arg = 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.
Set the version in the AWS::Flow::WorkflowOptions passed in to the #workflow method.
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 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 |
# File 'lib/aws/decider/decider.rb', line 245 module Workflows attr_accessor :version extend Utilities::UpwardLookups @precursors ||= [] def look_upwards(variable) precursors = self.ancestors.dup precursors.delete(self) results = precursors.map { |x| x.send(variable) if x.methods.map(&:to_sym).include? variable }.compact.flatten.uniq end property(:workflows, []) @workflows = [] def self.extended(base) base.send :include, InstanceMethods end # @deprecated Set the entry point with {Workflows#workflow} instead. # # @api private def entry_point(input=nil) if input @entry_point = input workflow_type = WorkflowType.new(self.to_s + "." + input.to_s, nil, WorkflowRegistrationOptions.new(:execution_method => input)) self.workflows.each { |workflow| workflow.name = self.to_s + "." + input.to_s } self.workflows.each do |workflow| workflow. = WorkflowRegistrationOptions.new(:execution_method => input) end self.workflows = self.workflows << workflow_type end return @entry_point if @entry_point raise "You must set an entry point on the workflow definition" end # @deprecated Set the version in the {WorkflowOptions} passed in to the {#workflow} method. # @api private def version(arg = nil) if arg self.workflows.each { |workflow| workflow.version = arg } self.workflows = self.workflows << WorkflowType.new(nil, arg, WorkflowOptions.new) end return @version end # Sets the activity client. # # @param name # Sets the client name for the activity client. # # @param block # A block of {ActivityOptions} for the activity client. # def activity_client(name, &block) = Utilities::(ActivityOptions, block) # TODO: Make sure this works for dynamic stuff begin activity_class = get_const(.prefix_name) rescue Exception => e #pass end = {} if activity_class values = activity_class.activities.map{|x| [x.name.split(".").last.to_sym, x.]} = Hash[*values.flatten] end # define_method(name) do # return @client if @client # @client ||= activity_class.activity_client.new(@decision_helper, options) # @client.decision_context = @decision_context # @client # end # else client_name = "@client_#{name}" define_method(name) do return instance_variable_get(client_name) if instance_variable_get(client_name) @decision_context ||= Fiber.current[:decision_context] @decision_helper ||= @decision_context.decision_helper @decision_helper. = instance_variable_set(client_name, GenericActivityClient.new(@decision_helper, )) instance_variable_get(client_name) end instance_variable_get(client_name) end # @api private def ; self.workflows; end # Defines a new workflow. # # @param workflow_names # The entry points (methods) that starts the workflow. # # @param block # A block of {WorflowRegistrationOptions} for the workflow. # def workflow(*workflow_names, &block) workflow_names.each do |workflow_name| = Utilities::(WorkflowRegistrationOptions, block) .execution_method = workflow_name prefix_name = .prefix_name || self.to_s workflow_type = WorkflowType.new(prefix_name.to_s + "." + workflow_name.to_s, .version, ) self.workflows = self.workflows << workflow_type end end # @return [MethodPair] # A {MethodPair} object. # def get_state_method(get_state_method = nil, = {}) data_converter = [:data_converter] @get_state_method = MethodPair.new(get_state_method, data_converter) unless get_state_method.nil? @get_state_method end # Defines a signal for the workflow. # # @param method_name # The signal method for the workflow. # # @param [SignalWorkflowOptions] options # The {SignalWorkflowOptions} for this signal. # def signal(method_name , = {}) data_converter = [:data_converter] signal_name = [:signal_name] signal_name ||= method_name.to_s data_converter ||= FlowConstants.default_data_converter @signals ||= {} @signals[signal_name] = MethodPair.new(method_name, data_converter) @signals end # @return [Hash] # A hash of string(SignalName) => MethodPair(method, signalConverter) objects def signals @signals end # Instance methods for {DecisionContext}. module InstanceMethods # Returns the {DecisionContext} instance. # @return [DecisionContext] # The {DecisionContext} instance. def decision_context FlowFiber.current[:decision_context] end # Returns the workflow ID. # # @return # The workflow ID. # def workflow_id self.decision_context.workflow_context.decision_task.workflow_execution.workflow_id end # Returns the decision helper for the decision context. This should be an instance of {DecisionHelper} or a # class derived from it. def run_id self.decision_context.workflow_context.decision_task.workflow_execution.run_id end def decision_helper FlowFiber.current[:decision_context].decision_helper end # Sets the activity client for this decision context. # # @param name # The name of the activity client. # # @param block # A block of {ActivityOptions} for the activity client. # def activity_client(name=nil, &block) = Utilities::(ActivityOptions, block) begin activity_class = get_const(.prefix_name) rescue Exception => e #pass end = {} if activity_class values = activity_class.activities.map{|x| [x.name.split(".").last.to_sym, x.]} = Hash[*values.flatten] end client = GenericActivityClient.new(self.decision_helper, ) self.class.send(:define_method, name) { client } if ! name.nil? client end # Creates a timer on the workflow that executes the supplied block after a specified delay. # # @param delay_seconds # The number of seconds to delay before executing the block. # # @param block # The block to execute when the timer expires. # def create_timer(delay_seconds, &block) self.decision_context.workflow_clock.create_timer(delay_seconds, block) end # Creates an asynchronous timer on the workflow that executes the supplied block after a specified delay. # # @param (see #create_timer) # # @deprecated # Use {#create_timer_async} instead. # # @api private def async_create_timer(delay_seconds, &block) task { self.decision_context.workflow_clock.create_timer(delay_seconds, block) } end # Creates an asynchronous timer on the workflow that executes the supplied block after a specified delay. # # @param (see #create_timer) # def create_timer_async(delay_seconds, &block) task { self.decision_context.workflow_clock.create_timer(delay_seconds, block) } end # Restarts the workflow as a new workflow execution. # # @param args # Arguments for this workflow execution, in JSON format. # # @param [ContinueAsNewOptions] block # The {ContinueAsNewOptions} for this workflow execution. # def continue_as_new(*args, &block) = Utilities::(ContinueAsNewOptions, block) @data_converter ||= YAMLDataConverter.new if ! args.empty? input = @data_converter.dump args .input = input end known_workflows = self.class.workflows # If there is only one workflow, we can unambiguously say that we should use that one. if known_workflows.length == 1 .precursors << known_workflows.first. end # If we can find a name that matches, use that one. if .execution_method matching_option = self.class.workflows.map(&:options).find {|x| x.execution_method == .execution_method } .precursors << matching_option unless matching_option.nil? end self.decision_context.workflow_context. = end end end |
Class Method Details
.extended(base) ⇒ Object
256 257 258 |
# File 'lib/aws/decider/decider.rb', line 256 def self.extended(base) base.send :include, InstanceMethods end |
Instance Method Details
#_options ⇒ 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.
330 |
# File 'lib/aws/decider/decider.rb', line 330 def ; self.workflows; end |
#activity_client(name, &block) ⇒ Object
Sets the activity client.
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 |
# File 'lib/aws/decider/decider.rb', line 295 def activity_client(name, &block) = Utilities::(ActivityOptions, block) # TODO: Make sure this works for dynamic stuff begin activity_class = get_const(.prefix_name) rescue Exception => e #pass end = {} if activity_class values = activity_class.activities.map{|x| [x.name.split(".").last.to_sym, x.]} = Hash[*values.flatten] end # define_method(name) do # return @client if @client # @client ||= activity_class.activity_client.new(@decision_helper, options) # @client.decision_context = @decision_context # @client # end # else client_name = "@client_#{name}" define_method(name) do return instance_variable_get(client_name) if instance_variable_get(client_name) @decision_context ||= Fiber.current[:decision_context] @decision_helper ||= @decision_context.decision_helper @decision_helper. = instance_variable_set(client_name, GenericActivityClient.new(@decision_helper, )) instance_variable_get(client_name) end instance_variable_get(client_name) end |
#entry_point(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.
Set the entry point with #workflow instead.
263 264 265 266 267 268 269 270 271 272 273 274 275 |
# File 'lib/aws/decider/decider.rb', line 263 def entry_point(input=nil) if input @entry_point = input workflow_type = WorkflowType.new(self.to_s + "." + input.to_s, nil, WorkflowRegistrationOptions.new(:execution_method => input)) self.workflows.each { |workflow| workflow.name = self.to_s + "." + input.to_s } self.workflows.each do |workflow| workflow. = WorkflowRegistrationOptions.new(:execution_method => input) end self.workflows = self.workflows << workflow_type end return @entry_point if @entry_point raise "You must set an entry point on the workflow definition" end |
#get_state_method(get_state_method = nil, options = {}) ⇒ MethodPair
Returns A MethodPair object.
353 354 355 356 357 |
# File 'lib/aws/decider/decider.rb', line 353 def get_state_method(get_state_method = nil, = {}) data_converter = [:data_converter] @get_state_method = MethodPair.new(get_state_method, data_converter) unless get_state_method.nil? @get_state_method end |
#look_upwards(variable) ⇒ Object
249 250 251 252 253 |
# File 'lib/aws/decider/decider.rb', line 249 def look_upwards(variable) precursors = self.ancestors.dup precursors.delete(self) results = precursors.map { |x| x.send(variable) if x.methods.map(&:to_sym).include? variable }.compact.flatten.uniq end |
#signal(method_name, options = {}) ⇒ Object
Defines a signal for the workflow.
367 368 369 370 371 372 373 374 375 |
# File 'lib/aws/decider/decider.rb', line 367 def signal(method_name , = {}) data_converter = [:data_converter] signal_name = [:signal_name] signal_name ||= method_name.to_s data_converter ||= FlowConstants.default_data_converter @signals ||= {} @signals[signal_name] = MethodPair.new(method_name, data_converter) @signals end |
#signals ⇒ Hash
Returns A hash of string(SignalName) => MethodPair(method, signalConverter) objects.
380 381 382 |
# File 'lib/aws/decider/decider.rb', line 380 def signals @signals end |
#workflow(*workflow_names, &block) ⇒ Object
Defines a new workflow.
340 341 342 343 344 345 346 347 348 |
# File 'lib/aws/decider/decider.rb', line 340 def workflow(*workflow_names, &block) workflow_names.each do |workflow_name| = Utilities::(WorkflowRegistrationOptions, block) .execution_method = workflow_name prefix_name = .prefix_name || self.to_s workflow_type = WorkflowType.new(prefix_name.to_s + "." + workflow_name.to_s, .version, ) self.workflows = self.workflows << workflow_type end end |