_how_to_build_workflow_engine_part_3.md

Workflow Engine をつくろう！ Part 3

記事は Qiita に書いています。

http://qiita.com/hakobera/items/cf570090943ea177938e

dag.rb

require 'tsort'

class DAG
  include TSort

  def initialize
    @tasks = {}
  end

  def tsort_each_node(&block)
    @tasks.each_key(&block)
  end

  def tsort_each_child(node, &block)
    @tasks.fetch(node).each(&block)
  end

  def add_task(task)
    _add_task(nil, task)
  end

  private

  def _add_task(parent, task)
    unless @tasks[task]    
      @tasks[task] = task.requires
      task.requires.each do |r|
        _add_task(task, r)
      end
    end
    task
  end
end

example.rb

require_relative 'workflow'
require_relative 'task'
require_relative 'local_file_target'

class GenerateDataTask < Task
  def output
    LocalFileTarget.new("/tmp/data_#{Time.now.strftime('%Y-%m-%d')}.txt")
  end

  def run
    File.open(output.path, "w") do |f|
      (1..10).each do |i|
        f.puts i
      end
    end
  end
end

class SumTask < Task
  def requires
    [ GenerateDataTask.new ]
  end

  def output
    LocalFileTarget.new("/tmp/output_#{Time.now.strftime('%Y-%m-%d')}.txt")
  end

  def run
    sum = 0
    File.foreach(input[0].path) do |line|
      sum += line.to_i
    end
    File.write(output.path, sum)
  end
end

Workflow.new.run(SumTask.new)

local_file_target.rb

require_relative 'target'

class LocalFileTarget < Target
  attr_reader :path

  def initialize(path)
    @path = path
  end

  def exists?
    File.exist?(path)
  end
end

target.rb

class Target
  def exists?
    raise NotImplementedError, "You must implement #{self.class}##{__method__}"
  end
end

task.rb

class Task

  def input
    @input ||= requires.map { |task| task.output }
  end

  def output
    raise NotImplementedError, "You must implement #{self.class}##{__method__}"
  end

  def run
    raise NotImplementedError, "You must implement #{self.class}##{__method__}"
  end

  def requires
    [] # If you need to define task dependencies, override in subclass
  end

  def eql?(other)
    self.hash == other.hash
  end

  def hash
    self.class.name.hash
  end
end

workflow.rb

require_relative 'dag'

class Workflow
  def run(task)
    dag = DAG.new
    dag.add_task(task)
    dag.tsort.each do |t|
      t.run unless t.output.exists?
    end
  end
end