em_transaction.rb

require 'eventmachine'

class Pipeline < EM::Completion
  def initialize(*steps)
    @steps = steps

    @steps.each_cons(2) do |pair|
      pair.first.callback do |input|
        EM.next_tick do
          pair.last.call(input)
        end
      end
    end

    @steps.last.callback do |input|
      succeed(input)
    end

    super()
  end

  def call(input)
    EM.next_tick do
      @steps.first.call(input)
    end
  end
end

class Step < EM::Completion
  def add_worker(&worker)
    @worker = worker
  end

  def call(input)
    succeed(@worker.call(input))
  end
end

class Transaction
  def self.start(input, &block)
    transaction = self.new
    transaction.instance_eval(&block)
    transaction.run(input)
  end

  def initialize
    @steps = []
  end

  def step(label, &block)
    step = Step.new
    step.add_worker(&block)
    @steps << step
  end

  def after(&block)
    @after = block
  end

  def run(input)
    pipeline = Pipeline.new(*@steps)

    pipeline.callback do
      @after.call
    end

    pipeline.call(input)
  end
end

EM.run do
  Transaction.start("hello") do
    step(:first) do |input|
      puts input
      input.upcase
    end

    step(:second) do |input|
      puts input
      "#{input} world"
    end

    step(:third) do |input|
      puts input
      input.upcase
    end

    step(:fourth) do |input|
      puts input
    end

    after do
      puts "transaction complete"
      EM.stop
    end
  end
end