Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Initial commit to GitHub.

  • Loading branch information...
commit 4971591e9c1185f05ceec49004b8800e3ff188b8 0 parents
@fredjean fredjean authored
17 .gitignore
@@ -0,0 +1,17 @@
+*.gem
+*.rbc
+.bundle
+.config
+.yardoc
+Gemfile.lock
+InstalledFiles
+_yardoc
+coverage
+doc/
+lib/bundler/man
+pkg
+rdoc
+spec/reports
+test/tmp
+test/version_tmp
+tmp
4 Gemfile
@@ -0,0 +1,4 @@
+source 'https://rubygems.org'
+
+# Specify your gem's dependencies in simpler_workflow.gemspec
+gemspec
22 LICENSE
@@ -0,0 +1,22 @@
+Copyright (c) 2012 Frederic Jean
+
+MIT License
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
29 README.md
@@ -0,0 +1,29 @@
+# SimplerWorkflow
+
+TODO: Write a gem description
+
+## Installation
+
+Add this line to your application's Gemfile:
+
+ gem 'simpler_workflow'
+
+And then execute:
+
+ $ bundle
+
+Or install it yourself as:
+
+ $ gem install simpler_workflow
+
+## Usage
+
+TODO: Write usage instructions here
+
+## Contributing
+
+1. Fork it
+2. Create your feature branch (`git checkout -b my-new-feature`)
+3. Commit your changes (`git commit -am 'Added some feature'`)
+4. Push to the branch (`git push origin my-new-feature`)
+5. Create new Pull Request
2  Rakefile
@@ -0,0 +1,2 @@
+#!/usr/bin/env rake
+require "bundler/gem_tasks"
23 lib/simpler_workflow.rb
@@ -0,0 +1,23 @@
+require 'aws-sdk'
+
+module SimplerWorkflow
+ def SimplerWorkflow.domain(domain_name)
+ @domains ||= {}
+ @domains[domain_name.to_sym] ||= Domain.new(domain_name)
+ end
+
+ def SimplerWorkflow.swf
+ @swf ||= ::AWS::SimpleWorkflow.new
+ end
+
+ autoload :Version, 'simpler_workflow/version'
+ autoload :Domain, 'simpler_workflow/domain'
+ autoload :Workflow, 'simpler_workflow/workflow'
+ autoload :Activity, 'simpler_workflow/activity'
+end
+
+class Map
+ def Map.from_json(json)
+ from_hash(JSON.parse(json))
+ end
+end
125 lib/simpler_workflow/activity.rb
@@ -0,0 +1,125 @@
+module SimplerWorkflow
+ class Activity
+ attr_reader :domain, :name, :version, :options, :next_activity
+
+ def initialize(domain, name, version, options = {})
+ Activity.activities[[name, version]] ||= begin
+ default_options = {
+ :default_task_list => name,
+ :default_task_start_to_close_timeout => 1 * 60 * 60,
+ :default_task_schedule_to_start_timeout => 20,
+ :default_task_schedule_to_close_timeout => 1 * 60 * 60,
+ :default_task_heartbeat_timeout => :none
+ }
+ @options = default_options.merge(options)
+ @domain = domain
+ @name = name
+ @version = version
+ @failure_policy = :abort
+ self
+ end
+ end
+
+ def method_missing(meth_name, *args)
+ if @options.has_key?(meth_name.to_sym)
+ @options[meth_name.to_sym] = args[0]
+ else
+ super
+ end
+ end
+
+ def on_success(activity, version = nil)
+ case activity
+ when Hash
+ activity.symbolize_keys!
+ name = activity[:name].to_sym
+ version = activity[:version]
+ when String
+ name = activity.to_sym
+ when Symbol
+ name = activity
+ end
+ @next_activity = { :name => name, :version => version }
+ end
+
+ def on_fail(failure_policy)
+ @failure_policy = failure_policy.to_sym
+ end
+
+ def failure_policy
+ @failure_policy || :abort
+ end
+
+ def perform_activity(&block)
+ @perform_task = block
+ end
+
+ def perform_task(task)
+ logger.info("Performing task #{name}")
+ @perform_task.call(task)
+ rescue => e
+ logger.error e.message
+ logger.error e.backtrace.join("\n")
+ task.fail! :reason => e.message, :details => {:failure_policy => failure_policy}.to_json
+ end
+
+ def to_activity_type
+ domain.activity_types[name.to_s, version]
+ end
+
+ def start_activity_loop
+ Thread.new do
+ loop do
+ begin
+ logger.info("Starting activity_loop for #{name}")
+ domain.activity_tasks.poll(name.to_s) do |task|
+ logger.info("Received task...")
+ perform_task(task)
+ unless task.responded?
+ if next_activity
+ result = {:next_activity => next_activity}.to_json
+ task.complete!(:result => result)
+ else
+ task.complete!
+ end
+ end
+ end
+ rescue Timeout::Error => e
+ rescue => e
+ logger.error(e.message)
+ logger.error(e.backtrace.join("\n"))
+ end
+ end
+ end
+ end
+
+ def self.[](name, version = nil)
+ case name
+ when String
+ name = name.to_sym
+ when Hash
+ name.symbolize_keys!
+ version = name[:version]
+ name = name[:name]
+ end
+ activities[[name, version]]
+ end
+
+ def self.register(name, version, activity)
+ activities[[name, version]] = activity
+ end
+
+ protected
+ def self.activities
+ @activities ||= {}
+ end
+
+ def self.swf
+ SimplerWorkflow.swf
+ end
+
+ def logger
+ $logger || Rails.logger
+ end
+ end
+end
88 lib/simpler_workflow/domain.rb
@@ -0,0 +1,88 @@
+module SimplerWorkflow
+ class Domain
+ def initialize(domain_name, retention = :none)
+ domain_name = domain_name.to_s
+ @domain = swf.domains[domain_name]
+ unless swf.domains.include?(@domain)
+ @domain = swf.domains.create(domain_name, retention)
+ end
+ self
+ end
+
+ def Domain.[](domain_name)
+ @domains ||= {}
+ @domains[domain_name] ||= Domain.new(domain_name)
+ end
+
+ def register_workflow(name, version, &block)
+ unless workflow = Workflow[name, version]
+ workflow = Workflow.new(self, name, version)
+ end
+
+ workflow.instance_eval(&block) if block
+
+ begin
+ self.domain.workflow_types.register(name.to_s, version, workflow.options)
+ rescue ::AWS::SimpleWorkflow::Errors::TypeAlreadyExistsFault => e
+ # Instance already registered...
+ end
+ workflow
+ end
+
+ def workflows
+ Workflow
+ end
+
+ def start_workflow(name, version, input)
+ logger.info("Starting workflow[#{name},#{version}]")
+ workflow = Workflow[name, version] || Workflow.new(self, name, version)
+ workflow.start_workflow(input)
+ end
+
+ def activities
+ Activity
+ end
+
+ def activity_types
+ domain.activity_types
+ end
+
+ def register_activity(name, version, &block)
+ unless activity = Activity[name, version]
+ logger.info("Registering Activity[#{name},#{version}]")
+ activity = Activity.new(self, name, version)
+ end
+
+ activity.instance_eval(&block) if block
+
+ begin
+ self.domain.activity_types.register(name.to_s, version, activity.options)
+ rescue ::AWS::SimpleWorkflow::Errors::TypeAlreadyExistsFault
+ # Nothing to do, should probably log something here...
+ end
+
+ activity
+ end
+
+ def method_missing(meth_name, *args)
+ if domain.respond_to?(meth_name.to_sym)
+ domain.send(meth_name.to_sym, *args)
+ else
+ super
+ end
+ end
+
+ protected
+ def swf
+ SimplerWorkflow.swf
+ end
+
+ def domain
+ @domain
+ end
+
+ def logger
+ $logger || Rails.logger
+ end
+ end
+end
3  lib/simpler_workflow/version.rb
@@ -0,0 +1,3 @@
+module SimplerWorkflow
+ VERSION = "0.0.1"
+end
150 lib/simpler_workflow/workflow.rb
@@ -0,0 +1,150 @@
+module SimplerWorkflow
+ class Workflow
+ attr_reader :task_list, :domain, :name, :version, :options, :initial_activity_type
+
+ def initialize(domain, name, version, options = {})
+ Workflow.workflows[[name, version]] ||= begin
+ default_options = {
+ :default_task_list => name,
+ :default_task_start_to_close_timeout => 1 * 24 * 60 * 60,
+ :default_execution_start_to_close_timeout => 1 * 24 * 60 * 60,
+ :default_child_policy => :terminate
+ }
+ @options = default_options.merge(options)
+ @domain = domain
+ @name = name
+ @version = version
+ self
+ end
+ end
+
+ def initial_activity(name, version = nil)
+ if activity = Activity[name.to_sym, version]
+ @initial_activity_type = activity.to_activity_type
+ elsif activity = domain.activity_types[name.to_s, version]
+ @initial_activity_type = activity
+ end
+ end
+
+ def decision_loop
+ logger.info("Starting decision loop for #{name.to_s}, #{version} listening to #{task_list}")
+ domain.decision_tasks.poll(task_list) do |decision_task|
+ logger.info("Received decision task")
+ decision_task.new_events.each do |event|
+ logger.info("Processing #{event.event_type}")
+ case event.event_type
+ when 'WorkflowExecutionStarted'
+ start_execution(decision_task, event)
+ when 'ActivityTaskCompleted'
+ activity_completed(decision_task, event)
+ when 'ActivityTaskFailed'
+ activity_failed(decision_task, event)
+ end
+ end
+ end
+ rescue Timeout::Error => e
+ retry
+ end
+
+ def task_list
+ @options[:default_task_list][:name].to_s
+ end
+
+ def start_execution(decision_task, event)
+ logger.info "Starting the execution of the job."
+ if @on_start_execution && @on_start_execution.respond_to?(:call)
+ @on_start_execution.call(decision_task, event)
+ else
+ decision_task.schedule_activity_task initial_activity_type, :input => event.attributes.input
+ end
+ end
+
+ def activity_completed(decision_task, event)
+ if @on_activity_completed && @on_activity_completed.respond_to?(:call)
+ @on_activity_completed.call(decision_task, event)
+ else
+ if event.attributes.keys.include?(:result)
+ result = Map.from_json(event.attributes.result)
+ next_activity = result[:next_activity]
+ activity_type = domain.activity_types[next_activity[:name], next_activity[:version]]
+ decision_task.schedule_activity_task activity_type, :input => scheduled_event(decision_task, event).attributes.input
+ else
+ logger.info("Workflow #{name}, #{version} completed")
+ decision_task.complete_workflow_execution :result => 'success'
+ end
+ end
+ end
+
+ def activity_failed(decision_task, event)
+ logger.info("Activity failed.")
+ if @on_activity_failed && @on_activity_failed.respond_to?(:call)
+ @on_activity_failed.call(decision_task, event)
+ else
+ if event.attributes.keys.include?(:details)
+ details = Map.from_json(event.attributes.details)
+ case details.failure_policy.to_sym
+ when :abort
+ decision_task.cancel_workflow_execution
+ when :retry
+ logger.info("Retrying activity #{last_activity(decision_task, event).name} #{last_activity(decision_task, event).version}")
+ decision_task.schedule_activity_task last_activity(decision_task, event), :input => last_input(decision_task, event)
+ end
+ else
+ decision_task.cancel_workflow_execution
+ end
+ end
+ end
+
+ def start_workflow(input)
+ domain.workflow_types[name.to_s, version].start_execution(:input => input)
+ end
+
+ def on_start_execution(&block)
+ @on_start_execution = block
+ end
+
+ def on_activity_completed(&block)
+ @on_activity_completed = block
+ end
+
+ def on_activity_failed(&block)
+ @on_activity_failed = block
+ end
+
+ def self.[](name, version)
+ workflows[[name, version]]
+ end
+
+ def self.register(name, version, workflow)
+ workflows[[name, version]] = workflow
+ end
+
+ def method_missing(meth_name, *args)
+ if @options.has_key?(meth_name.to_sym)
+ @options[meth_name.to_sym] = args[0]
+ else
+ super
+ end
+ end
+
+ protected
+ def self.workflows
+ @workflows ||= {}
+ end
+
+ def self.swf
+ SimplerWorkflow.swf
+ end
+
+ def scheduled_event(decision_task, event)
+ decision_task.events.to_a[event.attributes.scheduled_event_id - 1]
+ end
+ def last_activity(decision_task, event)
+ scheduled_event(decision_task, event).attributes.activity_type
+ end
+
+ def last_input(decision_task, event)
+ scheduled_event(decision_task, event).attributes.input
+ end
+ end
+end
16 lib/simpler_workflow/workflow_collection.rb
@@ -0,0 +1,16 @@
+module SimplerWorkflow
+ class WorkflowCollection
+ def [](name, version)
+ registry[[name,version]]
+ end
+
+ def []=(name, version, value)
+ registry[[name, version]] = value
+ end
+
+ protected
+ def registry
+ @registry ||= {}
+ end
+ end
+end
21 simpler_workflow.gemspec
@@ -0,0 +1,21 @@
+# -*- encoding: utf-8 -*-
+require File.expand_path('../lib/simpler_workflow/version', __FILE__)
+
+Gem::Specification.new do |gem|
+ gem.authors = ["Frederic Jean"]
+ gem.email = ["fred@snugghome.com"]
+ gem.description = %q{TODO: Write a gem description}
+ gem.summary = %q{TODO: Write a gem summary}
+ gem.homepage = ""
+
+ gem.files = `git ls-files`.split($\)
+ gem.executables = gem.files.grep(%r{^bin/}).map{ |f| File.basename(f) }
+ gem.test_files = gem.files.grep(%r{^(test|spec|features)/})
+ gem.name = "simpler_workflow"
+ gem.require_paths = ["lib"]
+ gem.version = SimplerWorkflow::VERSION
+
+ gem.add_dependency 'aws-sdk', '~> 1.3.6'
+ gem.add_dependency 'map'
+ gem.add_development_dependency 'rspec'
+end
Please sign in to comment.
Something went wrong with that request. Please try again.